mq消息订阅实现5

ibzdst-core/src/main/java/cn/ibizlab/core/util/config/RocktcermqCustmerRunner.java

@@ -19,6 +19,7 @@ import org.springframework.stereotype.Component;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
 
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -37,31 +38,32 @@ public class RocktcermqCustmerRunner implements ApplicationRunner {
 
     @Override
     public void run(ApplicationArguments args) {
-        List<RuleEngine> ruleEngines = ruleEngineMapper.selectList(new LambdaQueryWrapper<RuleEngine>().select(RuleEngine::getEngineId, RuleEngine::getExtParams));
+        List<RuleEngine> ruleEngines = ruleEngineMapper.selectList(new LambdaQueryWrapper<RuleEngine>()
+                .select(RuleEngine::getEngineId, RuleEngine::getExtParams));
         if (!StringUtils.isEmpty(ruleEngines)) {
+            DefaultMQPushConsumer consumer = null;
             for (int i = 0, len = ruleEngines.size(); i < len; i++) {
-                int finalI = i;
-                String topic = getParamTopic(ruleEngines.get(finalI).getExtParams());
-                if (isAutoCreate(ruleEngines.get(finalI).getExtParams()) && topic != null) {
+                Map map = getParamMap(ruleEngines.get(i).getExtParams());
+                if ("Y".equals(map.get("isAuto")) && !StringUtils.isEmpty(map.get("topic"))) {
                     log.info("ruleEngineConsumer 正在创建---------------------------------------");
-                    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(ruleEngines.get(finalI).getEngineId());
+                    consumer = new DefaultMQPushConsumer(ruleEngines.get(i).getEngineId());
                     consumer.setNamesrvAddr(namesrvAddr);
                     consumer.setConsumeThreadMin(1);
                     consumer.setConsumeThreadMax(1);
                     consumer.setConsumeMessageBatchMaxSize(1);
                     // 设置监听
-                    GeneralConsumeMsgListenerProcessor messageListener = new GeneralConsumeMsgListenerProcessor(ruleEngines.get(finalI).getEngineId());
+                    GeneralConsumeMsgListenerProcessor messageListener = new GeneralConsumeMsgListenerProcessor(ruleEngines.get(i).getEngineId());
                     consumer.registerMessageListener(messageListener);
                     consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
                     consumer.setInstanceName("dstMQMsgConsumer");
                     try {
-                        consumer.subscribe(topic, "*");
+                        consumer.subscribe((String) map.get("topic"), "*");
                         consumer.start();
-                        if (dataMap.get(ruleEngines.get(finalI)) == null) {
-                            dataMap.put(ruleEngines.get(finalI).getEngineId(), consumer);
-                            log.info("ruleEngineConsumer 创建成功 groupName={}, topics={}, namesrvAddr={}", ruleEngines.get(finalI).getEngineId(), ruleEngines.get(finalI).getModelId(), namesrvAddr);
+                        if (StringUtils.isEmpty(dataMap.get(ruleEngines.get(i)))) {
+                            dataMap.put(ruleEngines.get(i).getEngineId(), consumer);
+                            log.info("ruleEngineConsumer 创建成功 groupName={}, topics={}, namesrvAddr={}", ruleEngines.get(i).getEngineId(), ruleEngines.get(i).getModelId(), namesrvAddr);
                         } else {
-                            log.warn("ruleEngineConsumer 不能重复创建 groupName={}, topics={}, namesrvAddr={}", ruleEngines.get(finalI).getEngineId(), ruleEngines.get(finalI).getModelId(), namesrvAddr);
+                            log.warn("ruleEngineConsumer 不能重复创建 groupName={}, topics={}, namesrvAddr={}", ruleEngines.get(i).getEngineId(), ruleEngines.get(i).getModelId(), namesrvAddr);
                         }
                     } catch (MQClientException e) {
                         e.printStackTrace();
@@ -73,41 +75,21 @@ public class RocktcermqCustmerRunner implements ApplicationRunner {
     }
 
     /**
-     * 判断是否自动创建
+     * 获取参数map
      *
      * @param extParams
      * @return
      */
-    public boolean isAutoCreate(String extParams) {
+    public Map getParamMap(String extParams) {
         JSONArray objects = JSON.parseArray(extParams);
-        boolean flat = true;
+        HashMap map = new HashMap();
         if (!CollectionUtils.isEmpty(objects)) {
             for (int j = 0, len1 = objects.size(); j < len1; j++) {
-                String property = objects.getJSONObject(j).getString("property");
-                if ("isAuto".equals(property)) {
-                    if ("N".equals(objects.getJSONObject(j).getString("value"))) {
-                        flat = false;
-                    }
-                    break;
-                }
+                map.put(objects.getJSONObject(j).getString("property"), objects.getJSONObject(j).getString("value"));
             }
         }
-        return flat;
+        return map;
     }
 
 
-    public String getParamTopic(String extParams) {
-        JSONArray objects = JSON.parseArray(extParams);
-        boolean flat = true;
-        if (!CollectionUtils.isEmpty(objects)) {
-            for (int j = 0, len1 = objects.size(); j < len1; j++) {
-                String property = objects.getJSONObject(j).getString("property");
-                if ("topic".equals(property)) {
-                    return objects.getJSONObject(j).getString("value");
-                }
-            }
-        }
-        return null;
-    }
-
 }