test

ibzdst-core/src/main/java/cn/ibizlab/core/extensions/util/MQConsumeMsgListenerProcessor.java

@@ -31,38 +31,38 @@ public class MQConsumeMsgListenerProcessor implements MessageListenerOrderly {
 
     @Override
     public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
-        synchronized (lockObject) {
-            if (CollectionUtils.isEmpty(list)) {
-                log.info("MQ接收消息为空，直接返回成功");
-                return ConsumeOrderlyStatus.SUCCESS;
-            }
-            MessageExt messageExt = list.get(0);
-            log.info("MQ接收到的消息为：" + messageExt.toString());
-            try {
-                String topic = messageExt.getTopic();
-                String tags = messageExt.getTags();
-                String body = new String(messageExt.getBody(), "utf-8");
+        if (CollectionUtils.isEmpty(list)) {
+            log.info("MQ接收消息为空，直接返回成功");
+            return ConsumeOrderlyStatus.SUCCESS;
+        }
+        System.out.println(System.currentTimeMillis());
+        MessageExt messageExt = list.get(0);
+        log.info("MQ接收到的消息为：" + messageExt.toString());
+        try {
+            String topic = messageExt.getTopic();
+            String tags = messageExt.getTags();
+            String body = new String(messageExt.getBody(), "utf-8");
 
-                log.info("MQ消息topic={}, tags={}, 消息内容={}", topic, tags, body);
+            log.info("MQ消息topic={}, tags={}, 消息内容={}", topic, tags, body);
 
-            } catch (Exception e) {
-                log.error("获取MQ消息内容异常{}", e);
-            }
+        } catch (Exception e) {
+            log.error("获取MQ消息内容异常{}", e);
+        }
 
-            // 处理规则引擎构建消息
-            if (ruleEngineTopic.equalsIgnoreCase(messageExt.getTopic())) {
-                this.processRuleEngineData(messageExt);
-            }
-            return ConsumeOrderlyStatus.SUCCESS;
+        // 处理规则引擎构建消息
+        if (ruleEngineTopic.equalsIgnoreCase(messageExt.getTopic())) {
+            this.processRuleEngineData(messageExt);
         }
+        return ConsumeOrderlyStatus.SUCCESS;
     }
 
     protected void processRuleEngineData(MessageExt messageExt){
         try {
         String body = new String(messageExt.getBody(), "utf-8");
 
-        EngineMQMsg engineMQMsg = JSON.parseObject(body, new TypeReference<EngineMQMsg>() {});
-        ruleEngineExService.processData(engineMQMsg);
+        Thread.sleep(10000);
+//        EngineMQMsg engineMQMsg = JSON.parseObject(body, new TypeReference<EngineMQMsg>() {});
+//        ruleEngineExService.processData(engineMQMsg);
         } catch (Exception e) {
             log.error("获取MQ消息内容异常{}",e);
         }