one-YI
/
cloud-server
5
2
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

test(): 完成责任数路由逻辑优化,实现持久化测试

dev
面包骑士 2024-10-10 15:56:13 +08:00
parent f797c7cae8
commit 1abe99919b
5 changed files with 23 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/StrategyHandler.java
View File

@ -1,7 +1,9 @@
package com.muyu.data.processing.strategy;
import com.github.yulichang.toolkit.SpringContentUtils;
import com.muyu.common.redis.service.RedisService;
import com.muyu.data.processing.strategy.core.EndStrategy;
import org.springframework.data.redis.core.RedisTemplate;
/**
* 策略控制者接口
@ -13,7 +15,6 @@ import com.muyu.data.processing.strategy.core.EndStrategy;
*/
public interface StrategyHandler<T,R> {
RedisService redisService = new RedisService();
@SuppressWarnings("rawtypes")
StrategyHandler DEFAULT = param -> new EndStrategy();

10
cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/core/BasicStrategy.java
View File

@ -40,12 +40,10 @@ public class BasicStrategy extends abstractStrategyRouter<HashMap<String, BasicD
public Temporary2 apply(HashMap<String, BasicData> basicDataMap) {
log.info("开始执行基础校验节点。。。");
basicDataMap.put(CacheNameEnums.STORAGE.getCode(), null);
CacheNameEnums.getCodes().forEach(code-> {
// 如果缓存信息不为空，则说明车辆需要处理该事件
if (ObjectUtils.isNotEmpty(cacheUtils.hasKey(code, basicDataMap.get("VIN").getKey()))){
basicDataMap.put(code, null);
}
});
CacheNameEnums.getCodes()
.stream()
.filter(code -> cacheUtils.hasKey(code, basicDataMap.get("VIN").getKey()))
.forEach(code-> basicDataMap.put(code, null));
return applyStrategy(basicDataMap);
}
}

6
cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/core/RoutingStrategy.java
View File

@ -8,6 +8,7 @@ import com.muyu.data.processing.strategy.StrategyHandler;
import com.muyu.data.processing.strategy.abstractStrategyRouter;
import com.muyu.data.processing.strategy.branch.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.springframework.stereotype.Component;
/**
@ -36,11 +37,12 @@ public class RoutingStrategy extends abstractStrategyRouter<HashMap<String, Ba
@Override
protected StrategyMapper<HashMap<String, BasicData>, Temporary2> registerStrategy() {
log.info("路由节点已通过。。。");
return param -> {
log.info("路由节点已通过。。。");
// 编写路由规则
for (String code : map.keySet()) {
if(param.containsKey(code)){
if(ObjectUtils.isNotEmpty(param.get(code))){
param.remove(code);
return map.get(code);
}

5
cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/leaves/DataStorageStrategy.java
View File

@ -1,5 +1,6 @@
package com.muyu.data.processing.strategy.leaves;
import com.github.yulichang.toolkit.SpringContentUtils;
import com.muyu.common.iotdb.config.IotDBSessionConfig;
import com.muyu.data.processing.domain.BasicData;
import java.util.HashMap;
@ -11,6 +12,7 @@ import com.muyu.data.processing.strategy.core.RoutingStrategy;
import lombok.extern.slf4j.Slf4j;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.pool.SessionPool;
import org.springframework.stereotype.Component;
/**
@ -27,6 +29,7 @@ import org.springframework.stereotype.Component;
@Component
public class DataStorageStrategy extends abstractStrategyRouter<HashMap<String, BasicData>, Temporary2>
implements StrategyHandler<HashMap<String, BasicData>, Temporary2> {
private final SessionPool sessionPool = SpringContentUtils.getBean(SessionPool.class);
@Override
protected StrategyMapper<HashMap<String, BasicData>, Temporary2> registerStrategy() {
@ -70,12 +73,12 @@ public class DataStorageStrategy extends abstractStrategyRouter<HashMap<String,
.append(values.substring(0, values.length() - 1))
.append(")");
try {
log.info("拼接sql语句: [{}]", sql);
new IotDBSessionConfig().getSessionPool().executeNonQueryStatement(sql.toString());
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
throw new RuntimeException(e);
}
log.info("成功执行sql语句: [{}]", sql);
}
}

14
cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/leaves/RealTimeAlarmStrategy.java
View File

@ -1,5 +1,6 @@
package com.muyu.data.processing.strategy.leaves;
import com.github.yulichang.toolkit.SpringContentUtils;
import com.muyu.data.processing.domain.BasicData;
import java.util.HashMap;
@ -8,6 +9,7 @@ import com.muyu.data.processing.strategy.StrategyHandler;
import com.muyu.data.processing.strategy.abstractStrategyRouter;
import com.muyu.data.processing.strategy.core.RoutingStrategy;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
/**
@ -24,6 +26,8 @@ import org.springframework.stereotype.Component;
@Component
public class RealTimeAlarmStrategy extends abstractStrategyRouter<HashMap<String, BasicData>, Temporary2>
implements StrategyHandler<HashMap<String, BasicData>, Temporary2> {
// private final RedisTemplate<String,String> redisTemplate = SpringContentUtils.getBean(RedisTemplate.class);
@Override
protected StrategyMapper<HashMap<String, BasicData>, Temporary2> registerStrategy() {
log.info("实时数据处理分支已完成。。。");
@ -33,11 +37,11 @@ public class RealTimeAlarmStrategy extends abstractStrategyRouter<HashMap<String
@Override
public Temporary2 apply(HashMap<String, BasicData> basicDataMap) {
log.info("开始执行实时数据处理节点。。。");
String vin = basicDataMap.get("VIN").getKey();
basicDataMap.keySet().forEach(key -> {
BasicData basicData = basicDataMap.get(key);
redisService.setCacheObject(vin+":"+basicData.getKey(), basicData.getValue());
});
// String vin = basicDataMap.get("VIN").getKey();
// basicDataMap.keySet().forEach(key -> {
// BasicData basicData = basicDataMap.get(key);
// redisTemplate.opsForValue().set(vin+":"+basicData.getKey(), basicData.getValue());
// });
return applyStrategy(basicDataMap);
}
}