Merge remote-tracking branch 'dev.saas/dev' into dev

cloud-common/cloud-common-caffeine/src/main/java/com/muyu/common/caffeine/utils/CacheUtils.java

@@ -38,7 +38,6 @@ public class CacheUtils {
     /**
      * 清除缓存 - 根据缓存分区名称
      * @param cacheName 缓存分区名称
-     * @return 操作结果
      */
     public void delCache(String cacheName) {
         getCache(cacheName).invalidate();

cloud-common/cloud-common-kafka/src/main/java/com/muyu/common/kafka/constants/KafkaConstants.java

@@ -8,7 +8,7 @@ package com.muyu.common.kafka.constants;
  */
 public class KafkaConstants {
 
-    public final static String KafkaTopic = "kafka_topic";
+    public final static String KafkaTopic = "kafka_topic2";
 
-    public final static String KafkaGrop = "kafka_grop";
+    public final static String KafkaGrop = "kafka_grop2";
 }

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

@@ -32,18 +32,23 @@ public class BasicStrategy extends abstractStrategyRouter<HashMap<String, BasicD
 
     @Override
     protected StrategyMapper<HashMap<String, BasicData>, Temporary2> registerStrategy() {
-        log.info("基础校验节点已通过。。。");
         return param -> new RoutingStrategy();
     }
 
     @Override
     public Temporary2 apply(HashMap<String, BasicData> basicDataMap) {
         log.info("开始执行基础校验节点。。。");
+        StringBuilder str = new StringBuilder();
         basicDataMap.put(CacheNameEnums.STORAGE.getCode(), null);
+        str.append(CacheNameEnums.getInfo(CacheNameEnums.STORAGE.getCode())).append(",");
         CacheNameEnums.getCodes()
                 .stream()
-                .filter(code -> cacheUtils.hasKey(code, basicDataMap.get("VIN").getKey()))
-                .forEach(code-> basicDataMap.put(code, null));
+                .filter(code -> cacheUtils.hasKey(code, basicDataMap.get("VIN").getValue()))
+                .forEach(code-> {
+                    basicDataMap.put(code, null);
+                    str.append(CacheNameEnums.getInfo(code)).append(",");
+                });
+        log.info("基础校验节点已通过, 需执行分支:{}", str.substring(0, str.length()-1));
         return applyStrategy(basicDataMap);
     }
 }

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

@@ -42,7 +42,7 @@ public class RoutingStrategy   extends abstractStrategyRouter<HashMap<String, Ba
             log.info("路由节点已通过。。。");
             // 编写路由规则
             for (String code : map.keySet()) {
-                if(ObjectUtils.isNotEmpty(param.get(code))){
+                if(param.containsKey(code)){
                     param.remove(code);
                     return map.get(code);
                 }

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

@@ -44,7 +44,8 @@ public class DataStorageStrategy extends abstractStrategyRouter<HashMap<String,
         return applyStrategy(basicDataMap);
     }
 
-    private void addCarData(HashMap<String, BasicData> hashMap) {
+    private void addCarData(HashMap<String, BasicData> basicDataMap) {
+        HashMap<String, BasicData> hashMap = new HashMap<>(basicDataMap);
         StringBuilder sql = new StringBuilder("insert into root.one.");
         sql.append(hashMap.get("firmCode").getValue())
                 .append(".")

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

@@ -1,14 +1,17 @@
 package com.muyu.data.processing.strategy.leaves;
 
 import com.github.yulichang.toolkit.SpringContentUtils;
+import com.muyu.common.redis.service.RedisService;
 import com.muyu.data.processing.domain.BasicData;
 import java.util.HashMap;
+import java.util.Set;
 
 import com.muyu.data.processing.domain.Temporary2;
 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.apache.commons.lang3.ObjectUtils;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Component;
 
@@ -26,7 +29,7 @@ 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);
+    private final RedisService redisService = SpringContentUtils.getBean(RedisService.class);
 
     @Override
     protected StrategyMapper<HashMap<String, BasicData>, Temporary2> registerStrategy() {
@@ -37,11 +40,12 @@ 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);
-//            redisTemplate.opsForValue().set(vin+":"+basicData.getKey(), basicData.getValue());
-//        });
+        String vin = basicDataMap.get("VIN").getValue();
+        Set<String> keySet = basicDataMap.keySet();
+        keySet.stream()
+                .filter(key1 -> ObjectUtils.isNotEmpty(basicDataMap.get(key1)))
+                .map(basicDataMap::get)
+                .forEach(basicData -> redisService.setCacheObject(vin+":"+basicData.getKey(), basicData.getValue()));
         return applyStrategy(basicDataMap);
     }
 }