mq消息订阅实现4

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

@@ -2,9 +2,10 @@ package cn.ibizlab.core.util.config;
 
 import cn.ibizlab.core.extensions.util.GeneralConsumeMsgListenerProcessor;
 import cn.ibizlab.core.rule.domain.RuleEngine;
-import cn.ibizlab.core.rule.mapper.RuleEngineExtendMapper;
+import cn.ibizlab.core.rule.mapper.RuleEngineMapper;
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONArray;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
 import org.apache.rocketmq.client.exception.MQClientException;
@@ -27,7 +28,7 @@ import java.util.concurrent.ConcurrentHashMap;
 @ConditionalOnExpression("${rocketmq.consumer.enabled}&&'${rocketmq.consumer.namesrvAddr:}'!=''")
 public class RocktcermqCustmerRunner implements ApplicationRunner {
     @Autowired
-    private RuleEngineExtendMapper ruleEngineMapper;
+    private RuleEngineMapper ruleEngineMapper;
 
     @Value("${rocketmq.consumer.namesrvAddr:}")
     private String namesrvAddr;
@@ -36,36 +37,37 @@ public class RocktcermqCustmerRunner implements ApplicationRunner {
 
     @Override
     public void run(ApplicationArguments args) {
-        List<RuleEngine> ruleEngines = ruleEngineMapper.selectCustomerList();
+        List<RuleEngine> ruleEngines = ruleEngineMapper.selectList(new LambdaQueryWrapper<RuleEngine>().select(RuleEngine::getEngineId, RuleEngine::getExtParams));
         if (!StringUtils.isEmpty(ruleEngines)) {
             for (int i = 0, len = ruleEngines.size(); i < len; i++) {
                 int finalI = i;
-                    if (isAutoCreate(ruleEngines.get(finalI).getExtParams())) {
-                        log.info("ruleEngineConsumer 正在创建---------------------------------------");
-                        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(ruleEngines.get(finalI).getEngineId());
-                        consumer.setNamesrvAddr(namesrvAddr);
-                        consumer.setConsumeThreadMin(1);
-                        consumer.setConsumeThreadMax(1);
-                        consumer.setConsumeMessageBatchMaxSize(1);
-                        // 设置监听
-                        GeneralConsumeMsgListenerProcessor messageListener = new GeneralConsumeMsgListenerProcessor(ruleEngines.get(finalI).getEngineId());
-                        consumer.registerMessageListener(messageListener);
-                        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
-                        consumer.setInstanceName("dstMQMsgConsumer");
-                        try {
-                            consumer.subscribe(ruleEngines.get(finalI).getModelId(), "*");
-                            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);
-                            } else {
-                                log.warn("ruleEngineConsumer 不能重复创建 groupName={}, topics={}, namesrvAddr={}", ruleEngines.get(finalI).getEngineId(), ruleEngines.get(finalI).getModelId(), namesrvAddr);
-                            }
-                        } catch (MQClientException e) {
-                            e.printStackTrace();
-                            log.error("ruleEngineConsumer 创建失败!");
+                String topic = getParamTopic(ruleEngines.get(finalI).getExtParams());
+                if (isAutoCreate(ruleEngines.get(finalI).getExtParams()) && topic != null) {
+                    log.info("ruleEngineConsumer 正在创建---------------------------------------");
+                    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(ruleEngines.get(finalI).getEngineId());
+                    consumer.setNamesrvAddr(namesrvAddr);
+                    consumer.setConsumeThreadMin(1);
+                    consumer.setConsumeThreadMax(1);
+                    consumer.setConsumeMessageBatchMaxSize(1);
+                    // 设置监听
+                    GeneralConsumeMsgListenerProcessor messageListener = new GeneralConsumeMsgListenerProcessor(ruleEngines.get(finalI).getEngineId());
+                    consumer.registerMessageListener(messageListener);
+                    consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
+                    consumer.setInstanceName("dstMQMsgConsumer");
+                    try {
+                        consumer.subscribe(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);
+                        } else {
+                            log.warn("ruleEngineConsumer 不能重复创建 groupName={}, topics={}, namesrvAddr={}", ruleEngines.get(finalI).getEngineId(), ruleEngines.get(finalI).getModelId(), namesrvAddr);
                         }
+                    } catch (MQClientException e) {
+                        e.printStackTrace();
+                        log.error("ruleEngineConsumer 创建失败!");
                     }
+                }
             }
         }
     }
@@ -93,4 +95,19 @@ public class RocktcermqCustmerRunner implements ApplicationRunner {
         return flat;
     }
 
+
+    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;
+    }
+
 }