feat(REQ-3004) - 调整日志报告的逻辑

workflow-engine-core/src/main/java/cn/axzo/workflow/core/conf/SupportRefreshProperties.java

@@ -75,6 +75,9 @@ public class SupportRefreshProperties {
     @Value("${workflow.useNewToAdminApi:false}")
     private Boolean useNewToAdminApi;
 
+    @Value("${workflow.sendDingTalk:true}")
+    private Boolean sendDingTalk;
+
     @Value("${workflow.esSyncBatchSize:10}")
     private Integer esSyncBatchSize;
 }

workflow-engine-server/src/main/java/cn/axzo/workflow/server/common/aspectj/ErrorReportAspect.java

@@ -1,6 +1,7 @@
 package cn.axzo.workflow.server.common.aspectj;
 
 import cn.axzo.workflow.core.common.event.ApiLogEvent;
+import cn.axzo.workflow.core.conf.SupportRefreshProperties;
 import cn.axzo.workflow.server.common.annotation.EnvConfig;
 import cn.axzo.workflow.server.common.annotation.ErrorReporter;
 import cn.axzo.workflow.server.common.config.property.WorkflowProperties;
@@ -47,8 +48,8 @@ import static cn.azxo.framework.common.constatns.Constants.CTX_LOG_ID_MDC;
 public class ErrorReportAspect implements Ordered {
     @Value("${spring.profiles.active}")
     private String profile;
-    @Value("${workflow.sendDingTalk:true}")
+    @Resource
-    private Boolean sendDingTalk;
+    private SupportRefreshProperties refreshProperties;
     @Resource
     private ApplicationEventPublisher applicationEventPublisher;
     @Resource
@@ -141,7 +142,7 @@ public class ErrorReportAspect implements Ordered {
         EnvConfig[] envConfigs = errorReporter.envConfig();
         for (EnvConfig envConfig : envConfigs) {
             if (Arrays.asList(envConfig.profiles()).contains(profile)) {
-                boolean filterSendDingTalk = sendDingTalk;
+                boolean filterSendDingTalk = refreshProperties.getSendDingTalk();
                 if (workflowProperties.getFilterSendDingTalk().contains(operation.summary())) {
                     filterSendDingTalk = false;
                 }

workflow-engine-server/src/main/java/cn/axzo/workflow/server/mq/inside/consumer/ElasticSearchBatchSyncListener.java

@@ -1,6 +1,11 @@
 package cn.axzo.workflow.server.mq.inside.consumer;
 
+import cn.axzo.framework.domain.data.IdHelper;
+import cn.axzo.framework.rocketmq.Event;
+import cn.axzo.framework.rocketmq.utils.TraceUtils;
 import cn.axzo.workflow.core.conf.SupportRefreshProperties;
+import com.alibaba.fastjson.JSONObject;
+import com.google.common.base.Strings;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
@@ -9,12 +14,15 @@ import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
 import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
 import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
+import org.slf4j.MDC;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
+import org.springframework.util.CollectionUtils;
 
 import javax.annotation.PostConstruct;
 import javax.annotation.Resource;
 import java.nio.charset.StandardCharsets;
+import java.util.Map;
 
 /**
  * 批量同步 ES
@@ -42,10 +50,10 @@ public class ElasticSearchBatchSyncListener {
         consumer.subscribe(topic, "*");
         consumer.setConsumeMessageBatchMaxSize(refreshProperties.getEsSyncBatchSize());
         consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
-            log.warn("拉取消息条数：{}", msgs.size());
             try {
                 for (MessageExt msg : msgs) {
                     String msgBody = new String(msg.getBody(), StandardCharsets.UTF_8);
+                    Event event = convertEvent(msg, msgBody);
                 }
             } catch (Exception e) {
                 return ConsumeConcurrentlyStatus.RECONSUME_LATER;
@@ -54,4 +62,33 @@ public class ElasticSearchBatchSyncListener {
         });
 //        consumer.start();
     }
+
+    private Event convertEvent(MessageExt msg, String msgBody) {
+        try {
+            return JSONObject.parseObject(msgBody, Event.class);
+        } catch (Exception e) {
+            log.error("====MQ CONSUMER {} ===={}, msgId = {}, parse event error, event = {}",
+                    getTraceId(msg), msg.getMsgId(), msgBody, e.getMessage(), e);
+            throw e;
+        }
+    }
+
+    public String getTraceId(MessageExt msg) {
+        Map<String, String> headers = msg.getProperties();
+        MDC.put(TraceUtils.CTX_LOG_ID, headers.get(TraceUtils.TRACE_ID));
+        MDC.put(TraceUtils.TRACE_ID, headers.get(TraceUtils.TRACE_ID));
+        MDC.put(TraceUtils.TRACE_ID_IN_MDC, headers.get(TraceUtils.TRACE_ID));
+        if (log.isDebugEnabled()) {
+            log.debug("received message, topic={}, headers={}", topic, headers);
+        }
+        String traceId = null;
+        if (!CollectionUtils.isEmpty(headers) && headers.containsKey(TraceUtils.TRACE_ID)) {
+            traceId = headers.get(TraceUtils.TRACE_ID);
+        }
+        if (Strings.isNullOrEmpty(traceId)) {
+            traceId = IdHelper.get32UUID();
+        }
+        TraceUtils.putTraceId(traceId);
+        return traceId;
+    }
 }