版本分支提交

elab-alert/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>elab-parent</artifactId>
         <groupId>com.elab.core</groupId>
-        <version>2.0.5.1-SNAPSHOT</version>
+        <version>2.0.5.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>elab-alert</artifactId>

elab-annotation/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>elab-parent</artifactId>
         <groupId>com.elab.core</groupId>
-        <version>2.0.5.1-SNAPSHOT</version>
+        <version>2.0.5.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

elab-cache/pom.xml

@@ -6,7 +6,7 @@
     <parent>
         <groupId>com.elab.core</groupId>
         <artifactId>elab-parent</artifactId>
-        <version>2.0.5.1-SNAPSHOT</version>
+        <version>2.0.5.2-SNAPSHOT</version>
     </parent>
     <groupId>com.elab.cache</groupId>
     <artifactId>elab-cache</artifactId>

elab-core/pom.xml

@@ -7,7 +7,7 @@
     <parent>
         <groupId>com.elab.core</groupId>
         <artifactId>elab-parent</artifactId>
-        <version>2.0.5.1-SNAPSHOT</version>
+        <version>2.0.5.2-SNAPSHOT</version>
     </parent>
 
     <groupId>com.elab.core</groupId>

elab-core/src/main/java/com/elab/core/async/pruducer/TaskProducer.java

@@ -32,6 +32,10 @@ public class TaskProducer implements ITaskProducer {
 
     private TaskStoreProcess handler;
 
+    private volatile Long startTime = 0L;
+
+    private Integer maxWarnTime = 5 * 60 * 1000;
+
     public TaskProducer(TaskExecutorQueue taskExecutorQueue) {
         this.taskExecutorQueue = taskExecutorQueue;
     }
@@ -53,7 +57,12 @@ public class TaskProducer implements ITaskProducer {
         TaskExecutorDecoration taskExecutorDecoration = new TaskExecutorDecoration(taskStoreData, taskExecutor);
         boolean isConsumer = this.taskExecutorQueue.getRealTimeQueue().offer(taskExecutorDecoration);
         if (!isConsumer) {
-            logger.info("队列满了!");
+            long currentTimeMillis = System.currentTimeMillis();
+            long current = currentTimeMillis - startTime;
+            if (current > maxWarnTime) {
+                startTime = currentTimeMillis;
+                logger.error("队列满了!");
+            }
         }
     }
 

elab-core/src/main/java/com/elab/core/componts/LoadContextTool.java

@@ -0,0 +1,36 @@
+package com.elab.core.componts;
+
+import java.util.concurrent.CountDownLatch;
+
+/**
+ * @Module 加载工具类
+ * @Description 判断应用是否加载完毕
+ * @Author liukaixiong
+ * @Date 2020/12/24 14:30
+ */
+public class LoadContextTool {
+
+    private volatile boolean isLoadFinish = false;
+
+    private CountDownLatch countDownLatch;
+
+    public LoadContextTool(Integer countLatch) {
+        this.countDownLatch = new CountDownLatch(countLatch);
+    }
+
+    public void awaitFinish() {
+        if (!isLoadFinish) {
+            try {
+                countDownLatch.await();
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+        }
+    }
+
+    public void loadFinish() {
+        this.isLoadFinish = true;
+        countDownLatch.countDown();
+    }
+
+}

elab-core/src/main/java/com/elab/core/utils/DataUtils.java

@@ -2,7 +2,6 @@ package com.elab.core.utils;
 
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONObject;
-import com.elab.core.utils.ObjectUtils;
 import org.apache.commons.lang3.StringUtils;
 
 import java.util.Arrays;
@@ -66,10 +65,29 @@ public class DataUtils {
         return JSON.parseObject(text);
     }
 
+    public static Integer getStringByIntegerValue(String num) {
+        if (num != null) {
+            try {
+                return Integer.valueOf(num);
+            } catch (Exception e) {
+
+            }
+        }
+        return null;
+    }
+
     public static Integer getIntegerValue(Integer value, int defaultValue) {
         if (value == null) {
             return defaultValue;
         }
         return value;
     }
+
+    public static <T> T getDefaultValue(Object obj, Object defaultValue) {
+        if (ObjectUtils.isEmpty(obj)) {
+            return (T) defaultValue;
+        }
+        return (T) obj;
+    }
+
 }

elab-db/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>com.elab.core</groupId>
         <artifactId>elab-parent</artifactId>
-        <version>2.0.5.1-SNAPSHOT</version>
+        <version>2.0.5.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>elab-db</artifactId>

elab-db/src/main/java/com/elab/core/dao/BasicBaseDao.java

@@ -16,6 +16,7 @@ import com.elab.core.spring.method.DefaultCheckSQLProcess;
 import com.elab.core.spring.method.SQLEventHandler;
 import com.elab.core.sql.config.SqlCommandType;
 import com.elab.core.utils.BeanPropertyUtils;
+import com.google.common.collect.Sets;
 import org.apache.commons.lang3.ArrayUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -57,6 +58,9 @@ public class BasicBaseDao implements ApplicationContextAware,
 
     private ApplicationContext applicationContext;
 
+    private Set<Class> basicTypes = Sets.newHashSet(String.class, Integer.class, Double.class, Long.class, Float.class, Date.class);
+
+
     @Override
     public void afterPropertiesSet() throws Exception {
         Map<String, CheckSqlProcess> beansOfType = this.applicationContext.getBeansOfType(CheckSqlProcess.class);
@@ -243,6 +247,15 @@ public class BasicBaseDao implements ApplicationContextAware,
         return ts;
     }
 
+    public <T> List<T> executeQueryBasicTypeList(String sql, Object o, Class<T> elementType) throws Exception {
+        JdbcParamsModel jdbcParamsModel = commonParseSql(sql, o);
+        long start = System.currentTimeMillis();
+        List<T> ts = (List<T>) this.jdbcTemplate.query(jdbcParamsModel.getSql(), jdbcParamsModel.getObjects(), new SingleColumnRowMapper<>(elementType));
+        long time = System.currentTimeMillis() - start;
+        logger.debug(" SQL 执行耗时 : " + time);
+        return ts;
+    }
+
     /**
      * 查询一个集合对象并返回指定的类型
      *
@@ -488,7 +501,7 @@ public class BasicBaseDao implements ApplicationContextAware,
         try {
             map = this.jdbcTemplate.queryForMap(jdbcParamsModel.getSql(), jdbcParamsModel.getObjects());
         } catch (EmptyResultDataAccessException e) {
-            
+
         } catch (IncorrectResultSizeDataAccessException e) {
             logger.error("数据异常,只要查询一个,但是结果集是多个!", e);
         }

elab-es/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>elab-parent</artifactId>
         <groupId>com.elab.core</groupId>
-        <version>2.0.5.1-SNAPSHOT</version>
+        <version>2.0.5.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

elab-kafka/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>elab-parent</artifactId>
         <groupId>com.elab.core</groupId>
-        <version>2.0.5.1-SNAPSHOT</version>
+        <version>2.0.5.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

elab-kafka/src/main/java/com/elab/core/kafka/config/KafkaConsumerConfig.java

@@ -1,10 +1,12 @@
 package com.elab.core.kafka.config;
 
+import com.elab.core.kafka.consumer.AbstractBatchKafkaConsumer;
 import com.elab.core.kafka.consumer.AbstractKafkaConsumer;
+import com.elab.core.kafka.consumer.BatchKafkaMsgListener;
 import com.elab.core.kafka.consumer.KafkaMsgListener;
-import com.elab.core.kafka.monitor.TopicMonitorRule;
+import com.elab.core.kafka.sender.KafkaMonitorProducer;
 import com.elab.core.kafka.sender.KafkaProducerImpl;
-import com.jay.monitor.data.client.ext.MonitorRuleFactory;
+import com.elab.log.ext.CatIdMonitorRule;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -16,6 +18,7 @@ import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
@@ -49,7 +52,7 @@ public class KafkaConsumerConfig {
     @Autowired(required = false)
     private ProducerListener<String, String> producerListeners;
 
-//    @Bean
+    //    @Bean
 //    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory(KafkaProperties kafkaProperties) {
 //        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
 //        factory.setConsumerFactory(consumerFactory(kafkaProperties));
@@ -61,16 +64,21 @@ public class KafkaConsumerConfig {
 //        return factory;
 //    }
 
+    @Bean
+    @ConditionalOnMissingBean(value = {KafkaMonitorProducer.class})
+    public KafkaMonitorProducer kafkaMonitorProducer() {
+        return new KafkaMonitorProducer();
+    }
+
     @Bean
     @ConditionalOnBean(value = {AbstractKafkaConsumer.class})
-    public KafkaMessageListenerContainer<String, String> kafkaMessageListenerContainer(KafkaProperties kafkaProperties, List<AbstractKafkaConsumer> kafkaConsumer,@Autowired(required = false) MonitorRuleFactory monitorRuleFactory) {
+    public KafkaMessageListenerContainer<String, String> kafkaMessageListenerContainer(KafkaProperties kafkaProperties, List<AbstractKafkaConsumer> kafkaConsumer) {
         List<String> topicList = kafkaConsumer.stream().map(AbstractKafkaConsumer::subscribeTopic).collect(Collectors.toList());
         logger.info("关注kafka的topic:" + topicList);
         ContainerProperties containerProps = new ContainerProperties(topicList.toArray(new String[topicList.size()]));
 
-        KafkaMsgListener kafkaMsgListener = new KafkaMsgListener(kafkaConsumer,monitorRuleFactory);
+        KafkaMsgListener kafkaMsgListener = new KafkaMsgListener(kafkaConsumer);
         // 包装成监控
-        //  KafkaMonitorConsumerListener kafkaMonitorConsumerListener = new KafkaMonitorConsumerListener(kafkaMsgListener, monitorRuleFactory);
 
         containerProps.setMessageListener(kafkaMsgListener);
         containerProps.setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
@@ -83,8 +91,32 @@ public class KafkaConsumerConfig {
     }
 
     @Bean
-    public TopicMonitorRule topicMonitorRule() {
-        return new TopicMonitorRule();
+    @ConditionalOnBean(value = {AbstractBatchKafkaConsumer.class})
+    public KafkaMessageListenerContainer<String, String> batchKafkaMessageListenerContainer(KafkaProperties kafkaProperties, List<AbstractBatchKafkaConsumer> kafkaConsumer) {
+
+        List<String> topicList = kafkaConsumer.stream().map(AbstractBatchKafkaConsumer::subscribeTopic).collect(Collectors.toList());
+
+        logger.info("关注kafka的topic:" + topicList);
+        ContainerProperties containerProps = new ContainerProperties(topicList.toArray(new String[topicList.size()]));
+
+        BatchKafkaMsgListener kafkaMsgListener = new BatchKafkaMsgListener(kafkaConsumer);
+
+        // 包装成监控
+        containerProps.setMessageListener(kafkaMsgListener);
+        containerProps.setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
+        containerProps.setSyncCommits(true);
+        DefaultKafkaConsumerFactory<String, String> cf =
+                new DefaultKafkaConsumerFactory<String, String>(getConfigConsumerMap(kafkaProperties));
+        KafkaMessageListenerContainer<String, String> container =
+                new KafkaMessageListenerContainer<>(cf, containerProps);
+
+        return container;
+    }
+
+    @Bean
+    @ConditionalOnMissingBean(value = {CatIdMonitorRule.class})
+    public CatIdMonitorRule topicMonitorRule() {
+        return new CatIdMonitorRule();
     }
 
 

elab-kafka/src/main/java/com/elab/core/kafka/consumer/AbstractBatchKafkaConsumer.java

@@ -0,0 +1,19 @@
+package com.elab.core.kafka.consumer;
+
+import org.springframework.kafka.listener.BatchAcknowledgingMessageListener;
+
+/**
+ * @Module TODO
+ * @Description TODO
+ * @Author liukaixiong
+ * @Date 2020/12/28 15:16
+ */
+public abstract class AbstractBatchKafkaConsumer implements BatchAcknowledgingMessageListener<String, String> {
+
+    /**
+     * 订阅的topic消息
+     *
+     * @return
+     */
+    public abstract String subscribeTopic();
+}

elab-kafka/src/main/java/com/elab/core/kafka/consumer/BatchKafkaMsgListener.java

@@ -0,0 +1,84 @@
+package com.elab.core.kafka.consumer;
+
+import com.elab.log.utils.CatCrossProcess;
+import com.jay.monitor.data.client.MonitorSendProducer;
+import com.jay.monitor.data.client.utils.MonitorUtils;
+import com.jay.monitor.data.core.model.serializable.MQDataDTO;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.kafka.listener.BatchAcknowledgingMessageListener;
+import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.util.Assert;
+
+import java.util.List;
+
+/**
+ * @Module 监听器
+ * @Description kafka批量消费监听器
+ * @Author liukaixiong
+ * @Date 2020/10/27 19:14
+ */
+public class BatchKafkaMsgListener implements BatchAcknowledgingMessageListener<String, String>, InitializingBean {
+
+    private Logger logger = LoggerFactory.getLogger(getClass());
+    private List<AbstractBatchKafkaConsumer> messageListeners;
+
+    public BatchKafkaMsgListener(List<AbstractBatchKafkaConsumer> kafkaConsumer) {
+        Assert.notNull(kafkaConsumer, "请实现一个AbstractKafkaConsumer的子类,并注册到ioc容器中");
+        this.messageListeners = kafkaConsumer;
+    }
+
+    @Override
+    public void onMessage(List<ConsumerRecord<String, String>> batchData, Acknowledgment acknowledgment) {
+        if (batchData == null || batchData.size() == 0) {
+            return;
+        }
+        // 默认不支持单个topic不同业务的情况
+        ConsumerRecord<String, String> data = batchData.get(0);
+        try {
+            CatCrossProcess.processTransactionMsg("kafka-batch-" + data.topic(), data.partition() + "", (transaction) -> {
+                long startTime = System.currentTimeMillis();
+                try {
+                    String topic = data.topic();
+                    this.messageListeners.forEach((listener) -> {
+                        if (listener.subscribeTopic().equals(topic)) {
+                            listener.onMessage(batchData, acknowledgment);
+                        }
+                    });
+                    long requestTime = System.currentTimeMillis() - startTime;
+                    if (acknowledgment != null) {
+                        sendMonitorData(batchData, 1, requestTime);
+                    } else {
+                        sendMonitorData(batchData, -1, requestTime);
+                    }
+                } catch (Exception e) {
+                    transaction.setStatus(e);
+                    logger.error("kafka消费失败", e);
+                    sendMonitorData(batchData, -1, System.currentTimeMillis() - startTime);
+                    throw e;
+                }
+                return null;
+            });
+        } catch (Exception e) {
+            // 基本不会出现
+            e.printStackTrace();
+            logger.error("未知异常", e);
+        }
+    }
+
+
+    private void sendMonitorData(List<ConsumerRecord<String, String>> batchData, Integer status, Long time) {
+        for (int i = 0; i < batchData.size(); i++) {
+            ConsumerRecord<String, String> data = batchData.get(i);
+            MQDataDTO mqDataDTO = MonitorUtils.builderConsumerMqDataDTO(data.topic(), data.key(), data.partition() + "", data.offset(), data.value(), status, time);
+            MonitorSendProducer.sendMsg(data.topic(), mqDataDTO);
+        }
+    }
+
+    @Override
+    public void afterPropertiesSet() throws Exception {
+
+    }
+}

elab-kafka/src/main/java/com/elab/core/kafka/consumer/KafkaMsgListener.java

@@ -3,7 +3,6 @@ package com.elab.core.kafka.consumer;
 import com.alibaba.fastjson.JSON;
 import com.elab.log.utils.CatCrossProcess;
 import com.jay.monitor.data.client.MonitorSendProducer;
-import com.jay.monitor.data.client.ext.MonitorRuleFactory;
 import com.jay.monitor.data.client.utils.MonitorUtils;
 import com.jay.monitor.data.core.model.serializable.MQDataDTO;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -28,18 +27,16 @@ public class KafkaMsgListener implements AcknowledgingMessageListener<String, St
     private Logger logger = LoggerFactory.getLogger(getClass());
     private List<AbstractKafkaConsumer> messageListeners;
 
-    private MonitorRuleFactory monitorRuleFactory;
-
-    public KafkaMsgListener(List<AbstractKafkaConsumer> kafkaConsumer, MonitorRuleFactory monitorRuleFactory) {
+    public KafkaMsgListener(List<AbstractKafkaConsumer> kafkaConsumer) {
         Assert.notNull(kafkaConsumer, "请实现一个AbstractKafkaConsumer的子类,并注册到ioc容器中");
         this.messageListeners = kafkaConsumer;
-        this.monitorRuleFactory = monitorRuleFactory;
     }
 
     @Override
     public void onMessage(ConsumerRecord<String, String> data, Acknowledgment acknowledgment) {
         Map<String, String> valueMap = JSON.parseObject(data.value(), Map.class);
         CatCrossProcess.buildRemoteMQMsg("kafka-" + data.topic(), data.partition() + "", valueMap, () -> {
+            long startTime = System.currentTimeMillis();
             try {
                 String topic = data.topic();
                 this.messageListeners.forEach((listener) -> {
@@ -47,23 +44,24 @@ public class KafkaMsgListener implements AcknowledgingMessageListener<String, St
                         listener.onMessage(data, acknowledgment);
                     }
                 });
+                long requestTime = System.currentTimeMillis() - startTime;
                 if (acknowledgment != null) {
-                    sendMonitorData(data, 1);
+                    sendMonitorData(data, 1, requestTime);
                 } else {
-                    sendMonitorData(data, -1);
+                    sendMonitorData(data, -1, requestTime);
                 }
             } catch (Exception e) {
                 logger.error("kafka消费失败", e);
-                sendMonitorData(data, -1);
+                sendMonitorData(data, -1, System.currentTimeMillis() - startTime);
                 throw e;
             }
             return acknowledgment;
         });
     }
 
-    private void sendMonitorData(ConsumerRecord<String, String> data, Integer status) {
-        MQDataDTO mqDataDTO = MonitorUtils.getConsumerMqDataDTO(this.monitorRuleFactory.getMqRuleMap(), data.topic(), data.key(), data.partition(), data.offset(), data.value(), status);
-        MonitorSendProducer.sendMsg(mqDataDTO);
+    private void sendMonitorData(ConsumerRecord<String, String> data, Integer status, Long time) {
+        MQDataDTO mqDataDTO = MonitorUtils.builderConsumerMqDataDTO(data.topic(), data.key(), data.partition() + "", data.offset(), data.value(), status, time);
+        MonitorSendProducer.sendMsg(data.topic(), mqDataDTO);
     }
 
     @Override

elab-kafka/src/main/java/com/elab/core/kafka/monitor/TopicMonitorRule.java

@@ -1,30 +0,0 @@
-package com.elab.core.kafka.monitor;
-
-import com.alibaba.fastjson.JSON;
-import com.dianping.cat.Cat;
-import com.elab.log.model.CatCrossIdModel;
-import com.jay.monitor.data.client.ext.MqMonitorRuleCallback;
-import com.jay.monitor.data.core.model.serializable.MQDataDTO;
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.function.BiConsumer;
-
-public class TopicMonitorRule extends MqMonitorRuleCallback {
-
-    @Override
-    public Map<String, BiConsumer<String, MQDataDTO>> getRuleMap() {
-        Map<String, BiConsumer<String, MQDataDTO>> map = new HashMap<>();
-
-        map.put("DEFAULT_ALL", (value, mqData) -> {
-            CatCrossIdModel catCrossIdModel = JSON.parseObject(value, CatCrossIdModel.class);
-            mqData.setRootLogId(catCrossIdModel.get_catRootMessageId());
-            mqData.setParentLogId(catCrossIdModel.get_catParentMessageId());
-            // 这里不用get_catChildMessageId的原因是因为生产者是不确定下游的消费者是谁的。
-            // 这里还是用当前线程产生的LogId。这个id非常重要。
-            mqData.setLogId(Cat.getCurrentMessageId());
-        });
-
-        return map;
-    }
-}

elab-kafka/src/main/java/com/elab/core/kafka/sender/KafkaMonitorProducer.java

@@ -0,0 +1,60 @@
+package com.elab.core.kafka.sender;
+
+import com.elab.log.utils.CatCrossProcess;
+import com.jay.monitor.data.client.MonitorSendProducer;
+import com.jay.monitor.data.client.utils.MonitorUtils;
+import com.jay.monitor.data.core.model.serializable.MQDataDTO;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.kafka.support.ProducerListener;
+import org.springframework.stereotype.Component;
+
+/**
+ * @Module 生产者监听器
+ * @Description kafka发送的时候会回调该接口告诉你成功或者失败
+ * @Author liukaixiong
+ * @Date 2020/10/29 16:10
+ */
+@Component
+public class KafkaMonitorProducer implements ProducerListener<String, String>, InitializingBean {
+    private Logger logger = LoggerFactory.getLogger(getClass());
+
+    private String defaultType = "Kafka";
+
+    @Override
+    public void onSuccess(String topic, Integer partition, String key, String value, RecordMetadata recordMetadata) {
+        try {
+            CatCrossProcess.createRemoteMQMsg(defaultType, topic, (msgIdMap) -> {
+                logger.debug("topic :" + topic + " 分区 : " + partition + " key : " + key + " value : " + value);
+                MQDataDTO mq = MonitorUtils.builderProducerMqDataDTO(topic, key, recordMetadata.partition() + "", recordMetadata.offset(), value, 1);
+                MonitorSendProducer.sendMsg(topic, mq);
+                return null;
+            });
+
+        } catch (Exception e) {
+            logger.warn("success监控发送器发送失败" + e.getMessage());
+        }
+    }
+
+    @Override
+    public void onError(String topic, Integer partition, String key, String value, Exception exception) {
+        try {
+            CatCrossProcess.createRemoteMQMsg(defaultType, topic, (msgIdMap) -> {
+                logger.debug("topic :" + topic + " 分区 : " + partition + " key : " + key + " value : " + value);
+                logger.error("发送失败", exception);
+                MQDataDTO mq = MonitorUtils.builderProducerMqDataDTO(topic, key, null + "", null, value, 1);
+                MonitorSendProducer.sendMsg(topic, mq);
+                return null;
+            });
+        } catch (Exception e) {
+            logger.warn("error监控发送器发送失败 :" + e.getMessage());
+        }
+    }
+
+    @Override
+    public void afterPropertiesSet() throws Exception {
+
+    }
+}

elab-kafka/src/main/java/com/elab/core/kafka/sender/KafkaProducerImpl.java

@@ -33,12 +33,14 @@ public class KafkaProducerImpl<K, V> extends KafkaTemplate<K, V> {
         ListenableFuture<SendResult<K, V>> recordMetadata = (ListenableFuture<SendResult<K, V>>) CatCrossProcess.createRemoteMQMsg(defaultType, topic, (msgContextMap) -> {
 //            // 将上下文消息携带到请求中
             JSONObject valueObject = JSON.parseObject(record.value().toString());
+            logger.debug("发送消息内容:" + record.value());
             msgContextMap.forEach((K, V) -> {
                 valueObject.put(K, V);
             });
             ProducerRecord<K, V> producerRecord = new ProducerRecord<K, V>(topic, record.partition(), record.timestamp(), record.key(), (V) valueObject.toJSONString());
             try {
                 ListenableFuture<SendResult<K, V>> listenableFuture = KafkaProducerImpl.super.doSend(producerRecord);
+                logger.debug("消息推送成功!" + listenableFuture.get().toString());
                 return listenableFuture;
             } catch (Exception e) {
                 LOG.error("推送kafka失败", e);

elab-kafka/src/test/java/com/elab/core/kafka/monitor/TestTopicMonitorRule.java

@@ -2,11 +2,12 @@ package com.elab.core.kafka.monitor;
 
 import com.alibaba.fastjson.JSON;
 import com.elab.log.model.CatCrossIdModel;
-import com.jay.monitor.data.client.ext.MqMonitorRuleCallback;
-import com.jay.monitor.data.core.model.serializable.MQDataDTO;
+import com.jay.monitor.data.client.ext.MonitorRuleCallback;
+import com.jay.monitor.data.core.model.serializable.base.BaseDTO;
+import com.jay.monitor.data.core.model.serializable.base.SearchIndexData;
+import com.jay.monitor.data.core.model.serializable.base.TranceLogData;
 import org.springframework.stereotype.Component;
 
-import java.util.HashMap;
 import java.util.Map;
 import java.util.function.BiConsumer;
 
@@ -17,25 +18,29 @@ import java.util.function.BiConsumer;
  * @Date 2020/10/30 19:42
  */
 @Component
-public class TestTopicMonitorRule extends MqMonitorRuleCallback {
+public class TestTopicMonitorRule implements MonitorRuleCallback {
 
     @Override
-    public Map<String, BiConsumer<String, MQDataDTO>> getRuleMap() {
-        Map<String, BiConsumer<String, MQDataDTO>> map = new HashMap<>();
-
-        map.put("DEFAULT_ALL", (value, mqData) -> {
+    public void registerRule(Map<String, BiConsumer<String, ? super BaseDTO>> ruleMap) {
+        ruleMap.put("DEFAULT_ALL", (value, mqData) -> {
             CatCrossIdModel catCrossIdModel = JSON.parseObject(value, CatCrossIdModel.class);
-            mqData.setRootLogId(catCrossIdModel.get_catRootMessageId());
-            mqData.setParentLogId(catCrossIdModel.get_catParentMessageId());
-            mqData.setLogId(catCrossIdModel.get_catChildMessageId());
-        });
+            if (mqData instanceof TranceLogData) {
+                TranceLogData tranceLogData = (TranceLogData) mqData;
+                tranceLogData.setRootLogId(catCrossIdModel.get_catRootMessageId());
+                tranceLogData.setParentLogId(catCrossIdModel.get_catParentMessageId());
+                tranceLogData.setLogId(catCrossIdModel.get_catChildMessageId());
+            }
 
-        map.put("uat-mysql-dts", (value, mqData) -> {
-            mqData.setGroupName("aaaa");
-            mqData.setGroupKeyName("bbbb");
-            mqData.setDataId("cccc");
         });
 
-        return map;
+        ruleMap.put("uat-mysql-dts", (value, mqData) -> {
+            if (mqData instanceof SearchIndexData) {
+                SearchIndexData indexData = (SearchIndexData) mqData;
+                indexData.setGroupName("aaaa");
+                indexData.setGroupKeyName("bbbb");
+                indexData.setDataId("cccc");
+            }
+        });
     }
+
 }

elab-log/pom.xml

@@ -7,7 +7,7 @@
     <parent>
         <groupId>com.elab.core</groupId>
         <artifactId>elab-parent</artifactId>
-        <version>2.0.5.1-SNAPSHOT</version>
+        <version>2.0.5.2-SNAPSHOT</version>
     </parent>
 
     <groupId>com.elab.log</groupId>

elab-log/src/main/java/com/elab/log/MsgUtils.java

@@ -50,4 +50,14 @@ public class MsgUtils {
     public static void sendMetricForCount(String name) {
         Cat.logMetricForCount(name);
     }
+
+    /**
+     * 总耗时统计
+     *
+     * @param name
+     * @param durationInMillis
+     */
+    public static void sendMetricForDuration(String name, long durationInMillis) {
+        Cat.logMetricForDuration(name, durationInMillis);
+    }
 }

elab-log/src/main/java/com/elab/log/ext/CatIdMonitorRule.java

@@ -0,0 +1,29 @@
+package com.elab.log.ext;
+
+import com.alibaba.fastjson.JSON;
+import com.dianping.cat.Cat;
+import com.elab.log.model.CatCrossIdModel;
+import com.jay.monitor.data.client.ext.MonitorRuleCallback;
+import com.jay.monitor.data.core.model.serializable.base.BaseDTO;
+import com.jay.monitor.data.core.model.serializable.base.TranceLogData;
+
+import java.util.Map;
+import java.util.function.BiConsumer;
+
+public class CatIdMonitorRule implements MonitorRuleCallback {
+
+    @Override
+    public void registerRule(Map<String, BiConsumer<String, ? super BaseDTO>> ruleMap) {
+        ruleMap.put(MonitorRuleCallback.allMatchRuleType, (value, mqData) -> {
+            if (mqData instanceof TranceLogData) {
+                CatCrossIdModel catCrossIdModel = JSON.parseObject(value, CatCrossIdModel.class);
+                TranceLogData tranceLogData = (TranceLogData) mqData;
+                tranceLogData.setRootLogId(catCrossIdModel.get_catRootMessageId());
+                tranceLogData.setParentLogId(catCrossIdModel.get_catParentMessageId());
+                // 这里不用get_catChildMessageId的原因是因为生产者是不确定下游的消费者是谁的。
+                // 这里还是用当前线程产生的LogId。这个id非常重要。
+                tranceLogData.setLogId(Cat.getCurrentMessageId());
+            }
+        });
+    }
+}

elab-log/src/main/java/com/elab/log/ext/CatRealTaskExecutor.java

@@ -5,7 +5,6 @@ import com.elab.core.async.TaskEnums;
 import com.elab.core.async.model.TaskStoreData;
 import com.elab.core.async.pruducer.TaskProducer;
 import com.elab.log.utils.CatCrossProcess;
-import com.jay.monitor.data.client.MonitorSendProducer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -35,10 +34,10 @@ public class CatRealTaskExecutor implements RealTaskExecutor {
     public void run() {
         Map<String, String> dataMap = taskStoreData.getDataMap();
 
-        Boolean invoke = CatCrossProcess.buildRemoteMsg("privateQueue", taskEnums.name(), dataMap, () -> {
+        CatCrossProcess.buildRemoteMsg("privateQueue", taskEnums.name(), dataMap, () -> {
             try {
                 realTaskExecutor.run();
-                MonitorSendProducer.sendMsg(taskStoreData);
+                //MonitorSendProducer.sendMsg(taskStoreData);
                 return true;
             } catch (Exception e) {
                 logger.error("线程消费失败", e);

elab-log/src/main/java/com/elab/log/ext/FunctionProcess.java

@@ -0,0 +1,15 @@
+package com.elab.log.ext;
+
+/**
+ * @Module TODO
+ * @Description TODO
+ * @Author liukaixiong
+ * @Date 2020/12/28 18:36
+ */
+public interface FunctionProcess<T, R> {
+    /**
+     * @param obj
+     * @return
+     */
+    R process(T obj) throws Exception;
+}

elab-log/src/main/java/com/elab/log/utils/CatCrossProcess.java

@@ -3,6 +3,7 @@ package com.elab.log.utils;
 import com.dianping.cat.Cat;
 import com.dianping.cat.message.Event;
 import com.dianping.cat.message.Transaction;
+import com.elab.log.ext.FunctionProcess;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.web.context.request.RequestAttributes;
 import org.springframework.web.context.request.RequestContextHolder;
@@ -238,4 +239,26 @@ public class CatCrossProcess {
         }
         return transaction;
     }
+
+    /**
+     * 执行事务消息
+     *
+     * @param name     名称
+     * @param value    值
+     * @param consumer -> 具体的消费逻辑
+     * @return
+     */
+    public static <T> T processTransactionMsg(String name, String value, FunctionProcess<Transaction, T> consumer) throws Exception {
+        Transaction transaction = getCrossTransactionMsg(name, value, null);
+        try {
+            T process = consumer.process(transaction);
+            transaction.setSuccessStatus();
+            return process;
+        } catch (Throwable e) {
+            transaction.setStatus(e);
+            throw e;
+        } finally {
+            transaction.complete();
+        }
+    }
 }

elab-log/src/main/java/com/elab/log/utils/EventProcessUtils.java

@@ -21,6 +21,29 @@ public class EventProcessUtils {
         Cat.logEvent(key, value, Event.SUCCESS, null);
     }
 
+    /**
+     * 写入事件
+     *
+     * @param key
+     * @param value
+     * @param nameValuePairs
+     */
+    public static void write(String key, String value, String nameValuePairs) {
+        Cat.logEvent(key, value, Event.SUCCESS, nameValuePairs);
+    }
+
+    /**
+     * 写入事件
+     *
+     * @param key
+     * @param value
+     * @param success
+     * @param nameValuePairs
+     */
+    public static void write(String key, String value, String success, String nameValuePairs) {
+        Cat.logEvent(key, value, success, nameValuePairs);
+    }
+
     /**
      * 插入一个Event失败的事件
      *
@@ -31,6 +54,17 @@ public class EventProcessUtils {
         Cat.logEvent(key, value, "1", null);
     }
 
+    /**
+     * 写入失败事件
+     *
+     * @param key
+     * @param value
+     * @param kv
+     */
+    public static void writeFail(String key, String value, String kv) {
+        Cat.logEvent(key, value, "1", kv);
+    }
+
     /**
      * 第三方埋点标记
      *

elab-mongodb/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>elab-parent</artifactId>
         <groupId>com.elab.core</groupId>
-        <version>2.0.5.1-SNAPSHOT</version>
+        <version>2.0.5.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

elab-mq/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>elab-parent</artifactId>
         <groupId>com.elab.core</groupId>
-        <version>2.0.5.1-SNAPSHOT</version>
+        <version>2.0.5.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
@@ -18,15 +18,19 @@
             <scope>provided</scope>
         </dependency>
         <dependency>
-            <groupId>com.elab.core</groupId>
-            <artifactId>elab-core</artifactId>
-            <version>${project.version}</version>
-        </dependency>
-        <dependency>
-            <groupId>com.elab.core</groupId>
-            <artifactId>elab-spring</artifactId>
+            <groupId>com.elab.log</groupId>
+            <artifactId>elab-log</artifactId>
             <version>${project.version}</version>
-        </dependency>
+        </dependency><!--        <dependency>-->
+<!--            <groupId>com.elab.core</groupId>-->
+<!--            <artifactId>elab-core</artifactId>-->
+<!--            <version>${project.version}</version>-->
+<!--        </dependency>-->
+<!--        <dependency>-->
+<!--            <groupId>com.elab.core</groupId>-->
+<!--            <artifactId>elab-spring</artifactId>-->
+<!--            <version>${project.version}</version>-->
+<!--        </dependency>-->
         <dependency>
             <groupId>com.elab.core</groupId>
             <artifactId>elab-db</artifactId>

elab-mq/src/main/java/com/elab/mq/config/RocketMQConfiguration.java

@@ -3,6 +3,7 @@ package com.elab.mq.config;
 import com.aliyun.openservices.ons.api.Consumer;
 import com.aliyun.openservices.ons.api.ONSFactory;
 import com.aliyun.openservices.ons.api.PropertyKeyConst;
+import com.elab.log.ext.CatIdMonitorRule;
 import com.elab.mq.listener.AbstractMessageListener;
 import com.elab.mq.listener.MessageListenerWrapper;
 import com.elab.mq.msg.IMsgProducerFacade;
@@ -13,6 +14,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
@@ -43,6 +45,12 @@ public class RocketMQConfiguration {
         return producerBean;
     }
 
+    @Bean
+    @ConditionalOnMissingBean(value = {CatIdMonitorRule.class})
+    public CatIdMonitorRule topicMonitorRule() {
+        return new CatIdMonitorRule();
+    }
+
     @Bean
     @ConditionalOnBean(value = {AbstractMessageListener.class})
     public Consumer consumerBean(@Autowired Environment env, @Autowired MessageListenerWrapper
@@ -50,7 +58,6 @@ public class RocketMQConfiguration {
         Consumer consumer = ONSFactory.createConsumer(mqProperties(env));
         if (messageListeners.size() > 0) {
             Map<String, Set<String>> topicMap = new LinkedHashMap<>();
-            Set<String> existTopic = new HashSet<>();
             for (int i = 0; i < messageListeners.size(); i++) {
                 AbstractMessageListener abstractMessageListener = messageListeners.get(i);
                 String topic = abstractMessageListener.topic();

elab-mq/src/main/java/com/elab/mq/consts/MqConstants.java

@@ -0,0 +1,18 @@
+package com.elab.mq.consts;
+
+/**
+ * @Module TODO
+ * @Description TODO
+ * @Author liukaixiong
+ * @Date 2020/12/17 14:24
+ */
+public class MqConstants {
+
+    public final static Integer MSG_OK = 1;
+    public final static Integer MSG_NO = -1;
+    public final static Integer MSG_WAIT = 0;
+
+    public final static String MQ_TYPE = "RMQ";
+
+
+}

elab-mq/src/main/java/com/elab/mq/dao/IConsumerDao.java

@@ -1,14 +1,14 @@
-package com.elab.mq.dao;
-
-import com.elab.core.aop.annotations.XmlGroupName;
-import com.elab.core.dao.IBaseDaoSupport;
-import com.elab.mq.model.ConsumerEntity;
-
-/**
- * @author liuhx
- * @create 2019/05/17 18:36
- * @email liuhx@elab-plus.com
- **/
-@XmlGroupName("consumer")
-public interface IConsumerDao extends IBaseDaoSupport<ConsumerEntity> {
-}
+//package com.elab.mq.dao;
+//
+//import com.elab.core.aop.annotations.XmlGroupName;
+//import com.elab.core.dao.IBaseDaoSupport;
+//import com.elab.mq.model.MqConsumerLogEntity;
+//
+///**
+// * @author liuhx
+// * @create 2019/05/17 18:36
+// * @email liuhx@elab-plus.com
+// **/
+//@XmlGroupName("consumer")
+//public interface IConsumerDao extends IBaseDaoSupport<MqConsumerLogEntity> {
+//}

elab-mq/src/main/java/com/elab/mq/dao/IProducerDao.java

@@ -1,15 +1,15 @@
-package com.elab.mq.dao;
-
-import com.elab.core.aop.annotations.XmlGroupName;
-import com.elab.core.dao.IBaseDaoSupport;
-import com.elab.mq.model.ProducerEntity;
-
-/**
- * @author liuhx
- * @create 2019/05/16 19:42
- * @email liuhx@elab-plus.com
- **/
-@XmlGroupName("producer")
-public interface IProducerDao extends IBaseDaoSupport<ProducerEntity> {
-
-}
+//package com.elab.mq.dao;
+//
+//import com.elab.core.aop.annotations.XmlGroupName;
+//import com.elab.core.dao.IBaseDaoSupport;
+//import com.elab.mq.model.MqProducerLogEntity;
+//
+///**
+// * @author liuhx
+// * @create 2019/05/16 19:42
+// * @email liuhx@elab-plus.com
+// **/
+//@XmlGroupName("producer")
+//public interface IProducerDao extends IBaseDaoSupport<MqProducerLogEntity> {
+//
+//}

elab-mq/src/main/java/com/elab/mq/listener/AbstractMessageListener.java

@@ -4,21 +4,12 @@ import com.aliyun.openservices.ons.api.Action;
 import com.aliyun.openservices.ons.api.ConsumeContext;
 import com.aliyun.openservices.ons.api.Message;
 import com.aliyun.openservices.ons.api.MessageListener;
-import com.dianping.cat.Cat;
 import com.dianping.cat.message.Transaction;
-import com.elab.core.utils.DateUtils;
 import com.elab.log.utils.CatCrossProcess;
-import com.elab.mq.dao.IConsumerDao;
-import com.elab.mq.model.ConsumerEntity;
 import com.elab.mq.model.MessageModel;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Value;
-
-import java.util.LinkedHashMap;
-import java.util.Map;
-import java.util.Properties;
 
 /**
  * 抽象成客户端能够使用的方法
@@ -30,15 +21,11 @@ public abstract class AbstractMessageListener implements MessageListener {
 
     private Logger logger = LoggerFactory.getLogger(getClass());
 
-    @Autowired
-    private IConsumerDao consumerDao;
-
-    @Value("${spring.application.name}")
-    private String moduleName;
+    @Autowired(required = false)
+    private ConsumerInterceptor consumerInterceptor = new LoggerConsumerInterceptor();
 
     private final String MQ_CONSUMER = "MQ_CONSUMER";
 
-
     /**
      * 客户端关注的topic消息,实现了该方法,消息监听被触发时会与之匹配。
      *
@@ -55,67 +42,6 @@ public abstract class AbstractMessageListener implements MessageListener {
         return "*";
     }
 
-
-    private ConsumerEntity init(MessageModel message) throws Exception {
-        String msgID = message.getMsgID();
-        Integer producerId = 0;
-
-        if (message.getProducerId() != null) {
-            producerId = Integer.valueOf(message.getProducerId());
-        }
-
-        String topic = message.getTopic();
-        String tag = message.getTag();
-        ConsumerEntity consumerEntity = new ConsumerEntity();
-        consumerEntity.setProducerId(producerId);
-        consumerEntity.setTopicId(topic);
-        consumerEntity.setModuleName(moduleName);
-        consumerEntity.setStatus(1);
-        consumerEntity.setMsgId(msgID);
-        consumerEntity.setTag(tag);
-        consumerEntity.setModuleMethod(getClass().getName());
-        ConsumerEntity oldConsumerEntity = consumerDao.selectByObject(consumerEntity);
-        if (oldConsumerEntity != null && oldConsumerEntity.getConsumerStatus().equals(1)) {
-            logger.warn("消费者验证失败，已经消费成功过一次，不允许重复消费 消费编号 : " + oldConsumerEntity.getId());
-            // throw new BusinessException("请不要重复消费，key：" + message);
-            return oldConsumerEntity;
-        }
-        logger.debug("消费者验证通过，消费者消费的数据匹配上生产者数据，");
-        if (oldConsumerEntity != null) {
-            oldConsumerEntity.setRetryCount(oldConsumerEntity.getRetryCount() + 1);
-            oldConsumerEntity.setConsumerStatus(-1);
-            oldConsumerEntity.setUpdated(DateUtils.getCurrentDateTime());
-            oldConsumerEntity.setCatId(Cat.getCurrentMessageId());
-            consumerDao.updateById(oldConsumerEntity);
-            logger.debug("触发消息重试 当前重试次数 : " + oldConsumerEntity.getRetryCount());
-        } else {
-            oldConsumerEntity = new ConsumerEntity();
-            oldConsumerEntity.setProducerId(producerId);
-            String content = new String(message.getBody());
-            if (content.length() < 5000) {
-                oldConsumerEntity.setContent(content);
-            } else {
-                oldConsumerEntity.setContent(content.substring(0, 5000));
-            }
-            oldConsumerEntity.setHouseId(message.getHouseId());
-            oldConsumerEntity.setModuleName(moduleName);
-            oldConsumerEntity.setStatus(1);
-            oldConsumerEntity.setCreated(DateUtils.getCurrentDateTime());
-            oldConsumerEntity.setRetryCount(0);
-            oldConsumerEntity.setTopicId(topic);
-            oldConsumerEntity.setConsumerStatus(0);
-            oldConsumerEntity.setMsgId(msgID);
-            oldConsumerEntity.setTag(tag);
-            oldConsumerEntity.setCatId(Cat.getCurrentMessageId());
-            oldConsumerEntity.setModuleMethod(getClass().getName());
-            int id = consumerDao.insert(oldConsumerEntity);
-            oldConsumerEntity.setId(id);
-            logger.debug(" 一条全新的消息，消息编号：" + id);
-        }
-        return oldConsumerEntity;
-    }
-
-
     /**
      * 客户端的业务逻辑实现
      *
@@ -130,39 +56,35 @@ public abstract class AbstractMessageListener implements MessageListener {
         String topic = message.getTopic();
         String tag = message.getTag();
 
-        Map<String, String> msgMap = conversionPropertyToMap(message.getUserProperties());
-
-        Transaction t = CatCrossProcess.getCrossTransactionMsg(MQ_CONSUMER, topic + "_" + tag + "_" + getClass().getSimpleName(), msgMap);
-
+        Transaction t = CatCrossProcess.getCrossTransactionMsg(MQ_CONSUMER, topic + "_" + tag + "_" + getClass().getSimpleName(), null);
         MessageModel messageModel = new MessageModel(message);
+
         logger.debug("消息处理被触发 : " + message.toString());
+
         Action action = null;
+        long start = System.currentTimeMillis();
         try {
-            ConsumerEntity oldConsumerEntity = init(messageModel);
+            //boolean isConsumer = checkRepeatMsg(messageModel);
+
+            boolean isConsumer = consumerInterceptor.check(messageModel);
             //如果已经成功过一次
-            if (oldConsumerEntity.getConsumerStatus() == 1) {
-                logger.warn(" 重复消费 过滤掉. 消费编号 : " + oldConsumerEntity.getId());
+            if (!isConsumer) {
                 t.setSuccessStatus();
                 return Action.CommitMessage;
             }
 
-            //logger.debug("更新消费者数据： " + ObjectUtils.objectParseJsonStr(oldConsumerEntity));
             action = consume0(messageModel, consumeContext);
-            // 锁争抢太严重了。
-//            if (action == null || Action.ReconsumeLater.equals(action)) {
-//                oldConsumerEntity.setConsumerStatus(-1);
-//                oldConsumerEntity.setUpdated(new Date());
-//                consumerDao.updateById(oldConsumerEntity);
-//            } else {
-//                oldConsumerEntity.setConsumerStatus(1);
-//                oldConsumerEntity.setUpdated(new Date());
-//                consumerDao.updateById(oldConsumerEntity);
-//            }
+            long time = System.currentTimeMillis() - start;
+            messageModel.setInvokeTime(time);
             t.setSuccessStatus();
+            //saveConsumerStatus(messageModel, MqConstants.MSG_OK, time);
+            this.consumerInterceptor.success(messageModel);
         } catch (Exception e) {
             t.setStatus(e);
             logger.error("消息处理异常 : ", e);
-            e.printStackTrace();
+            messageModel.setInvokeTime(System.currentTimeMillis() - start);
+            this.consumerInterceptor.error(messageModel, e);
+            //saveConsumerStatus(messageModel, MqConstants.MSG_NO, System.currentTimeMillis() - start);
             return Action.ReconsumeLater;
         } finally {
             t.complete();
@@ -171,16 +93,5 @@ public abstract class AbstractMessageListener implements MessageListener {
         return action;
     }
 
-    private Map<String, String> conversionPropertyToMap(Properties userProperties) {
-        if (userProperties == null) {
-            return null;
-        }
-
-        Map<String, String> msgMap = new LinkedHashMap<>();
-        userProperties.stringPropertyNames().forEach((key) -> {
-            msgMap.put(key, userProperties.getProperty(key));
-        });
-        return msgMap;
-    }
 
 }

elab-mq/src/main/java/com/elab/mq/listener/ConsumerInterceptor.java

@@ -0,0 +1,35 @@
+package com.elab.mq.listener;
+
+import com.elab.mq.model.MessageModel;
+
+/**
+ * @Module TODO
+ * @Description TODO
+ * @Author liukaixiong
+ * @Date 2020/12/30 14:57
+ */
+public interface ConsumerInterceptor {
+
+    /**
+     * 检查数据是否符合
+     *
+     * @param messageModel
+     * @return
+     */
+    public boolean check(MessageModel messageModel);
+
+    /**
+     * 成功数据回调
+     *
+     * @param messageModel 消息内容
+     */
+    public void success(MessageModel messageModel);
+
+    /**
+     * 失败数据回调
+     *
+     * @param messageModel
+     */
+    public void error(MessageModel messageModel,Throwable e);
+
+}

elab-mq/src/main/java/com/elab/mq/listener/LoggerConsumerInterceptor.java

@@ -0,0 +1,38 @@
+package com.elab.mq.listener;
+
+import com.aliyun.openservices.ons.api.SendResult;
+import com.elab.mq.model.MessageModel;
+import com.elab.mq.msg.ProducerInterceptor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @Module TODO
+ * @Description TODO
+ * @Author liukaixiong
+ * @Date 2020/12/30 15:02
+ */
+public class LoggerConsumerInterceptor implements ConsumerInterceptor, ProducerInterceptor {
+    private Logger logger = LoggerFactory.getLogger(getClass());
+
+    @Override
+    public boolean check(MessageModel messageModel) {
+        logger.debug("检测数据 : " + messageModel.getObject(String.class));
+        return true;
+    }
+
+    @Override
+    public void success(MessageModel messageModel, SendResult result) {
+        logger.debug("发送成功:" + result);
+    }
+
+    @Override
+    public void error(MessageModel messageModel, Throwable e) {
+        logger.debug("消费失败:" + messageModel.getMsgID(), e);
+    }
+
+    @Override
+    public void success(MessageModel messageModel) {
+        logger.debug("消费:" + messageModel.getMsgID());
+    }
+}

elab-mq/src/main/java/com/elab/mq/listener/MessageListenerWrapper.java

@@ -4,13 +4,17 @@ import com.aliyun.openservices.ons.api.Action;
 import com.aliyun.openservices.ons.api.ConsumeContext;
 import com.aliyun.openservices.ons.api.Message;
 import com.aliyun.openservices.ons.api.MessageListener;
+import com.elab.log.utils.CatCrossProcess;
 import com.google.common.base.Splitter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
+import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Map;
+import java.util.Properties;
 
 /**
  * 订阅组件适配器
@@ -24,6 +28,8 @@ public class MessageListenerWrapper implements MessageListener {
     @Autowired(required = false)
     private List<AbstractMessageListener> messageListeners;
 
+    private String defaultRmqGroup = "rmq_group";
+
     private Logger logger = LoggerFactory.getLogger(MessageListenerWrapper.class);
 
     @Override
@@ -31,21 +37,25 @@ public class MessageListenerWrapper implements MessageListener {
         Action result = Action.CommitMessage;
         if (messageListeners != null) {
             try {
-                for (int i = 0; i < messageListeners.size(); i++) {
-                    AbstractMessageListener abstractMessageListener = messageListeners.get(i);
-                    String currentTopic = message.getTopic();
-                    String currentTag = message.getTag();
-                    String topic = abstractMessageListener.topic();
-                    String tag = abstractMessageListener.tag();
-                    if (topic.equals(currentTopic) && ("*".equals(tag) || isContainsTags(tag, currentTag)
-                    )) {
-                        Action actionResult = abstractMessageListener.consume(message, consumeContext);
-                        if (actionResult == null || Action.ReconsumeLater == actionResult) {
-                            logger.warn(" MQ 消费失败 : " + abstractMessageListener);
-                            result = Action.ReconsumeLater;
+                Map<String, String> msgMap = conversionPropertyToMap(message.getUserProperties());
+
+                result = CatCrossProcess.buildRemoteMsg(defaultRmqGroup, message.getTopic(), msgMap, () -> {
+                    for (int i = 0; i < messageListeners.size(); i++) {
+                        AbstractMessageListener abstractMessageListener = messageListeners.get(i);
+                        String currentTopic = message.getTopic();
+                        String currentTag = message.getTag();
+                        String topic = abstractMessageListener.topic();
+                        String tag = abstractMessageListener.tag();
+                        if (topic.equals(currentTopic) && ("*".equals(tag) || isContainsTags(tag, currentTag))) {
+                            Action actionResult = abstractMessageListener.consume(message, consumeContext);
+                            if (actionResult == null || Action.ReconsumeLater == actionResult) {
+                                logger.warn(" MQ 消费失败 : " + abstractMessageListener);
+                                return Action.ReconsumeLater;
+                            }
                         }
                     }
-                }
+                    return Action.CommitMessage;
+                });
             } catch (Exception e) {
                 e.printStackTrace();
                 return Action.ReconsumeLater;
@@ -77,21 +87,33 @@ public class MessageListenerWrapper implements MessageListener {
         return flag;
     }
 
-    public static void main(String[] args) {
-        String text = "LOGIN || AUTHORIZATION_MOBILE || SHARE_XCX || BAOBEI || BAOBEI_SHOUDONG || GZXM || IM_VIDEO ||" +
-                " SPKF || QIANYUE || DAOFANG || LIUDIAN || RENGOU";
-        String tag = "Test";
-        List<String> tagList = Splitter.on("||").trimResults().omitEmptyStrings().splitToList(text);
-        List<String> tags = Splitter.on("||").trimResults().omitEmptyStrings().splitToList(tag);
-
-        if (isContainsTags(text, tag)) {
-            System.out.println("yes");
-        } else {
-            System.out.println("no");
+    private Map<String, String> conversionPropertyToMap(Properties userProperties) {
+        if (userProperties == null) {
+            return null;
         }
 
-        System.out.println(tagList);
-
-
+        Map<String, String> msgMap = new LinkedHashMap<>();
+        userProperties.stringPropertyNames().forEach((key) -> {
+            msgMap.put(key, userProperties.getProperty(key));
+        });
+        return msgMap;
     }
+
+//    public static void main(String[] args) {
+//        String text = "LOGIN || AUTHORIZATION_MOBILE || SHARE_XCX || BAOBEI || BAOBEI_SHOUDONG || GZXM || IM_VIDEO ||" +
+//                " SPKF || QIANYUE || DAOFANG || LIUDIAN || RENGOU";
+//        String tag = "Test";
+//        List<String> tagList = Splitter.on("||").trimResults().omitEmptyStrings().splitToList(text);
+//        List<String> tags = Splitter.on("||").trimResults().omitEmptyStrings().splitToList(tag);
+//
+//        if (isContainsTags(text, tag)) {
+//            System.out.println("yes");
+//        } else {
+//            System.out.println("no");
+//        }
+//
+//        System.out.println(tagList);
+//
+//
+//    }
 }

elab-mq/src/main/java/com/elab/mq/model/ConsumerEntity.java

@@ -1,235 +1,235 @@
-package com.elab.mq.model;
-
-import io.swagger.annotations.ApiModel;
-import io.swagger.annotations.ApiModelProperty;
-
-import javax.persistence.Column;
-import javax.persistence.Id;
-import javax.persistence.Table;
-import java.util.Date;
-
-/**
- * @author liuhx
- * @create 2019/05/16 19:32
- * @email liuhx@elab-plus.com
- **/
-@Table(name = "mq_consumer", catalog = "elab_db")
-@ApiModel(description = "消息队列生产者 实体")
-public class ConsumerEntity {
-
-    @Id
-    @ApiModelProperty(name = "id", value = "主键")
-    private Integer id;
-
-    @Column(name = "producer_id")
-    @ApiModelProperty(name = "producer_id", value = "生产者id")
-    private Integer producerId;
-
-    @Column(name = "module_name")
-    @ApiModelProperty(name = "moduleName", value = "微服务模块名")
-    private String moduleName;
-    @Column(name = "msg_id")
-    @ApiModelProperty(name = "msgId", value = "MQ的消息编号")
-    private String msgId;
-
-    @Column(name = "module_method")
-    @ApiModelProperty(name = "moduleMethod", value = "模块方法")
-    private String moduleMethod;
-    @Column(name = "tag")
-    @ApiModelProperty(name = "tag", value = "标签 业务区分标识")
-    private String tag;
-    @Column(name = "cat_id")
-    @ApiModelProperty(name = "catId", value = "监控链路编号")
-    private String catId;
-
-    @Column(name = "house_id")
-    @ApiModelProperty(name = "houseId", value = "项目id")
-    private Integer houseId;
-
-    @Column(name = "topic_id")
-    @ApiModelProperty(name = "topicId", value = "队列通道id")
-    private String topicId;
-
-    @Column(name = "content")
-    @ApiModelProperty(name = "content", value = "内容")
-    private String content;
-
-    @Column(name = "consumer_status")
-    @ApiModelProperty(name = "consumerStatus", value = "消费者状态 1成功-1失败")
-    private Integer consumerStatus;
-
-    @Column(name = "retry_count")
-    @ApiModelProperty(name = "retryCount", value = "失败重试次数")
-    private Integer retryCount;
-
-    /**
-     * 状态：1  有效  -1  无效
-     */
-    @Column(name = "status")
-    @ApiModelProperty(name = "status", value = "状态：1  有效  -1  无效")
-    private Integer status;
-
-    /**
-     * 创建时间
-     */
-    @Column(name = "created")
-    @ApiModelProperty(name = "created", value = "创建时间")
-    private Date created;
-
-    /**
-     * 修改时间
-     */
-    @Column(name = "updated")
-    @ApiModelProperty(name = "updated", value = "修改时间")
-    private Date updated;
-
-    /**
-     * 创建者
-     */
-    @Column(name = "creator")
-    @ApiModelProperty(name = "creator", value = "创建者")
-    private String creator;
-
-    /**
-     * 修改者
-     */
-    @Column(name = "updator")
-    @ApiModelProperty(name = "updator", value = "修改者")
-    private String updator;
-
-    public String getTag() {
-        return tag;
-    }
-
-    public String getCatId() {
-        return catId;
-    }
-
-    public void setCatId(String catId) {
-        this.catId = catId;
-    }
-
-    public void setTag(String tag) {
-        this.tag = tag;
-    }
-
-    public String getMsgId() {
-        return msgId;
-    }
-
-    public void setMsgId(String msgId) {
-        this.msgId = msgId;
-    }
-
-    public String getModuleMethod() {
-        return moduleMethod;
-    }
-
-    public void setModuleMethod(String moduleMethod) {
-        this.moduleMethod = moduleMethod;
-    }
-
-    public Integer getId() {
-        return id;
-    }
-
-    public void setId(Integer id) {
-        this.id = id;
-    }
-
-    public Integer getProducerId() {
-        return producerId;
-    }
-
-    public void setProducerId(Integer producerId) {
-        this.producerId = producerId;
-    }
-
-    public String getModuleName() {
-        return moduleName;
-    }
-
-    public void setModuleName(String moduleName) {
-        this.moduleName = moduleName;
-    }
-
-    public Integer getHouseId() {
-        return houseId;
-    }
-
-    public void setHouseId(Integer houseId) {
-        this.houseId = houseId;
-    }
-
-    public String getTopicId() {
-        return topicId;
-    }
-
-    public void setTopicId(String topicId) {
-        this.topicId = topicId;
-    }
-
-    public String getContent() {
-        return content;
-    }
-
-    public void setContent(String content) {
-        this.content = content;
-    }
-
-    public Integer getConsumerStatus() {
-        return consumerStatus;
-    }
-
-    public void setConsumerStatus(Integer consumerStatus) {
-        this.consumerStatus = consumerStatus;
-    }
-
-    public Integer getRetryCount() {
-        return retryCount;
-    }
-
-    public void setRetryCount(Integer retryCount) {
-        this.retryCount = retryCount;
-    }
-
-    public Integer getStatus() {
-        return status;
-    }
-
-    public void setStatus(Integer status) {
-        this.status = status;
-    }
-
-    public Date getCreated() {
-        return created;
-    }
-
-    public void setCreated(Date created) {
-        this.created = created;
-    }
-
-    public Date getUpdated() {
-        return updated;
-    }
-
-    public void setUpdated(Date updated) {
-        this.updated = updated;
-    }
-
-    public String getCreator() {
-        return creator;
-    }
-
-    public void setCreator(String creator) {
-        this.creator = creator;
-    }
-
-    public String getUpdator() {
-        return updator;
-    }
-
-    public void setUpdator(String updator) {
-        this.updator = updator;
-    }
-}
+//package com.elab.mq.model;
+//
+//import io.swagger.annotations.ApiModel;
+//import io.swagger.annotations.ApiModelProperty;
+//
+//import javax.persistence.Column;
+//import javax.persistence.Id;
+//import javax.persistence.Table;
+//import java.util.Date;
+//
+///**
+// * @author liuhx
+// * @create 2019/05/16 19:32
+// * @email liuhx@elab-plus.com
+// **/
+//@Table(name = "mq_consumer", catalog = "elab_db")
+//@ApiModel(description = "消息队列生产者 实体")
+//public class ConsumerEntity {
+//
+//    @Id
+//    @ApiModelProperty(name = "id", value = "主键")
+//    private Integer id;
+//
+//    @Column(name = "producer_id")
+//    @ApiModelProperty(name = "producer_id", value = "生产者id")
+//    private Integer producerId;
+//
+//    @Column(name = "module_name")
+//    @ApiModelProperty(name = "moduleName", value = "微服务模块名")
+//    private String moduleName;
+//    @Column(name = "msg_id")
+//    @ApiModelProperty(name = "msgId", value = "MQ的消息编号")
+//    private String msgId;
+//
+//    @Column(name = "module_method")
+//    @ApiModelProperty(name = "moduleMethod", value = "模块方法")
+//    private String moduleMethod;
+//    @Column(name = "tag")
+//    @ApiModelProperty(name = "tag", value = "标签 业务区分标识")
+//    private String tag;
+//    @Column(name = "cat_id")
+//    @ApiModelProperty(name = "catId", value = "监控链路编号")
+//    private String catId;
+//
+//    @Column(name = "house_id")
+//    @ApiModelProperty(name = "houseId", value = "项目id")
+//    private Integer houseId;
+//
+//    @Column(name = "topic_id")
+//    @ApiModelProperty(name = "topicId", value = "队列通道id")
+//    private String topicId;
+//
+//    @Column(name = "content")
+//    @ApiModelProperty(name = "content", value = "内容")
+//    private String content;
+//
+//    @Column(name = "consumer_status")
+//    @ApiModelProperty(name = "consumerStatus", value = "消费者状态 1成功-1失败")
+//    private Integer consumerStatus;
+//
+//    @Column(name = "retry_count")
+//    @ApiModelProperty(name = "retryCount", value = "失败重试次数")
+//    private Integer retryCount;
+//
+//    /**
+//     * 状态：1  有效  -1  无效
+//     */
+//    @Column(name = "status")
+//    @ApiModelProperty(name = "status", value = "状态：1  有效  -1  无效")
+//    private Integer status;
+//
+//    /**
+//     * 创建时间
+//     */
+//    @Column(name = "created")
+//    @ApiModelProperty(name = "created", value = "创建时间")
+//    private Date created;
+//
+//    /**
+//     * 修改时间
+//     */
+//    @Column(name = "updated")
+//    @ApiModelProperty(name = "updated", value = "修改时间")
+//    private Date updated;
+//
+//    /**
+//     * 创建者
+//     */
+//    @Column(name = "creator")
+//    @ApiModelProperty(name = "creator", value = "创建者")
+//    private String creator;
+//
+//    /**
+//     * 修改者
+//     */
+//    @Column(name = "updator")
+//    @ApiModelProperty(name = "updator", value = "修改者")
+//    private String updator;
+//
+//    public String getTag() {
+//        return tag;
+//    }
+//
+//    public String getCatId() {
+//        return catId;
+//    }
+//
+//    public void setCatId(String catId) {
+//        this.catId = catId;
+//    }
+//
+//    public void setTag(String tag) {
+//        this.tag = tag;
+//    }
+//
+//    public String getMsgId() {
+//        return msgId;
+//    }
+//
+//    public void setMsgId(String msgId) {
+//        this.msgId = msgId;
+//    }
+//
+//    public String getModuleMethod() {
+//        return moduleMethod;
+//    }
+//
+//    public void setModuleMethod(String moduleMethod) {
+//        this.moduleMethod = moduleMethod;
+//    }
+//
+//    public Integer getId() {
+//        return id;
+//    }
+//
+//    public void setId(Integer id) {
+//        this.id = id;
+//    }
+//
+//    public Integer getProducerId() {
+//        return producerId;
+//    }
+//
+//    public void setProducerId(Integer producerId) {
+//        this.producerId = producerId;
+//    }
+//
+//    public String getModuleName() {
+//        return moduleName;
+//    }
+//
+//    public void setModuleName(String moduleName) {
+//        this.moduleName = moduleName;
+//    }
+//
+//    public Integer getHouseId() {
+//        return houseId;
+//    }
+//
+//    public void setHouseId(Integer houseId) {
+//        this.houseId = houseId;
+//    }
+//
+//    public String getTopicId() {
+//        return topicId;
+//    }
+//
+//    public void setTopicId(String topicId) {
+//        this.topicId = topicId;
+//    }
+//
+//    public String getContent() {
+//        return content;
+//    }
+//
+//    public void setContent(String content) {
+//        this.content = content;
+//    }
+//
+//    public Integer getConsumerStatus() {
+//        return consumerStatus;
+//    }
+//
+//    public void setConsumerStatus(Integer consumerStatus) {
+//        this.consumerStatus = consumerStatus;
+//    }
+//
+//    public Integer getRetryCount() {
+//        return retryCount;
+//    }
+//
+//    public void setRetryCount(Integer retryCount) {
+//        this.retryCount = retryCount;
+//    }
+//
+//    public Integer getStatus() {
+//        return status;
+//    }
+//
+//    public void setStatus(Integer status) {
+//        this.status = status;
+//    }
+//
+//    public Date getCreated() {
+//        return created;
+//    }
+//
+//    public void setCreated(Date created) {
+//        this.created = created;
+//    }
+//
+//    public Date getUpdated() {
+//        return updated;
+//    }
+//
+//    public void setUpdated(Date updated) {
+//        this.updated = updated;
+//    }
+//
+//    public String getCreator() {
+//        return creator;
+//    }
+//
+//    public void setCreator(String creator) {
+//        this.creator = creator;
+//    }
+//
+//    public String getUpdator() {
+//        return updator;
+//    }
+//
+//    public void setUpdator(String updator) {
+//        this.updator = updator;
+//    }
+//}

elab-mq/src/main/java/com/elab/mq/model/MessageModel.java

@@ -16,11 +16,9 @@ import java.util.Properties;
 public class MessageModel<T> extends Message {
 
     private T object;
+
     private String groupId;
-    /**
-     * 房源编号
-     */
-    private Integer houseId;
+
     /**
      * 模块名称
      */
@@ -30,6 +28,10 @@ public class MessageModel<T> extends Message {
      */
     private String producerId;
 
+    /**
+     * 消费者运行耗时
+     */
+    private Long invokeTime;
 
     /**
      * 使用该model传递参数
@@ -85,15 +87,15 @@ public class MessageModel<T> extends Message {
     /**
      * 获取对应的转换对象，切记必须为JSON格式的
      *
-     * @param cls
+     * @param clazz
      * @return
      */
-    public T getObject(Class<T> cls) {
+    public <T> T getObject(Class<T> clazz) {
         byte[] body = getBody();
         if (body != null) {
             String bodyString = new String(body);
             if (StringUtils.isNotEmpty(bodyString)) {
-                return JSON.parseObject(bodyString, cls);
+                return JSON.parseObject(bodyString, clazz);
             }
         }
         return null;
@@ -122,14 +124,6 @@ public class MessageModel<T> extends Message {
         this.groupId = groupId;
     }
 
-    public Integer getHouseId() {
-        return houseId;
-    }
-
-    public void setHouseId(Integer houseId) {
-        this.houseId = houseId;
-    }
-
     public String getModuleName() {
         return moduleName;
     }
@@ -154,4 +148,11 @@ public class MessageModel<T> extends Message {
         return userProperties;
     }
 
+    public Long getInvokeTime() {
+        return invokeTime;
+    }
+
+    public void setInvokeTime(Long invokeTime) {
+        this.invokeTime = invokeTime;
+    }
 }

elab-mq/src/main/java/com/elab/mq/model/MqConsumerLogEntity.java

@@ -0,0 +1,199 @@
+package com.elab.mq.model;
+
+import javax.persistence.Column;
+import javax.persistence.Table;
+import java.io.Serializable;
+import java.util.Date;
+
+/**
+ * mq_consumer_log
+ * @author 
+ */
+@Table(name = "mq_consumer_log", catalog = "elab_db")
+public class MqConsumerLogEntity implements Serializable {
+    private Integer id;
+
+    @Column(name = "producer_log_id")
+    private Integer producerLogId;
+
+    @Column(name="group_id")
+    private String groupId;
+    /**
+     * 消息类型 : RMQ / KAFKA
+     */
+    @Column(name = "msg_type")
+    private String msgType;
+
+    /**
+     * 应用的名称
+     */
+    @Column(name = "application_name")
+    private String applicationName;
+
+    /**
+     * 应用的编号
+     */
+    @Column(name = "application_ip")
+    private String applicationIp;
+
+    /**
+     * 主题
+     */
+    private String topic;
+
+    /**
+     * rmq是tag,kafka是分区index
+     */
+    @Column(name = "partition_name")
+    private String partitionName;
+
+    /**
+     * 消息的key
+     */
+    @Column(name = "msg_key")
+    private String msgKey;
+
+    /**
+     * rmq是消息编号,kafka是offset
+     */
+    @Column(name = "msg_id")
+    private String msgId;
+
+    /**
+     * 应用的消费者方法定位
+     */
+    @Column(name = "module_method")
+    private String moduleMethod;
+
+    /**
+     * 1成功 -1失败
+     */
+    @Column(name = "invoke_status")
+    private Integer invokeStatus;
+
+    /**
+     * 日志编号
+     */
+    @Column(name = "log_id")
+    private String logId;
+
+    /**
+     * 创建时间
+     */
+    @Column(name = "created_time")
+    private Date createdTime;
+
+    public String getGroupId() {
+        return groupId;
+    }
+
+    public void setGroupId(String groupId) {
+        this.groupId = groupId;
+    }
+
+    private static final long serialVersionUID = 1L;
+
+    public Integer getId() {
+        return id;
+    }
+
+    public void setId(Integer id) {
+        this.id = id;
+    }
+
+    public Integer getProducerLogId() {
+        return producerLogId;
+    }
+
+    public void setProducerLogId(Integer producerLogId) {
+        this.producerLogId = producerLogId;
+    }
+
+    public String getMsgType() {
+        return msgType;
+    }
+
+    public void setMsgType(String msgType) {
+        this.msgType = msgType;
+    }
+
+    public String getApplicationName() {
+        return applicationName;
+    }
+
+    public void setApplicationName(String applicationName) {
+        this.applicationName = applicationName;
+    }
+
+    public String getApplicationIp() {
+        return applicationIp;
+    }
+
+    public void setApplicationIp(String applicationIp) {
+        this.applicationIp = applicationIp;
+    }
+
+    public String getTopic() {
+        return topic;
+    }
+
+    public void setTopic(String topic) {
+        this.topic = topic;
+    }
+
+    public String getPartitionName() {
+        return partitionName;
+    }
+
+    public void setPartitionName(String partitionName) {
+        this.partitionName = partitionName;
+    }
+
+    public String getMsgKey() {
+        return msgKey;
+    }
+
+    public void setMsgKey(String msgKey) {
+        this.msgKey = msgKey;
+    }
+
+    public String getMsgId() {
+        return msgId;
+    }
+
+    public void setMsgId(String msgId) {
+        this.msgId = msgId;
+    }
+
+    public String getModuleMethod() {
+        return moduleMethod;
+    }
+
+    public void setModuleMethod(String moduleMethod) {
+        this.moduleMethod = moduleMethod;
+    }
+
+    public Integer getInvokeStatus() {
+        return invokeStatus;
+    }
+
+    public void setInvokeStatus(Integer invokeStatus) {
+        this.invokeStatus = invokeStatus;
+    }
+
+    public String getLogId() {
+        return logId;
+    }
+
+    public void setLogId(String logId) {
+        this.logId = logId;
+    }
+
+    public Date getCreatedTime() {
+        return createdTime;
+    }
+
+    public void setCreatedTime(Date createdTime) {
+        this.createdTime = createdTime;
+    }
+}

elab-mq/src/main/java/com/elab/mq/model/MqProducerLogEntity.java

@@ -0,0 +1,176 @@
+package com.elab.mq.model;
+
+import javax.persistence.Column;
+import javax.persistence.Table;
+import java.io.Serializable;
+import java.util.Date;
+
+/**
+ * mq_producer_log
+ * @author 
+ */
+@Table(name = "mq_producer_log", catalog = "elab_db")
+public class MqProducerLogEntity implements Serializable {
+    private Integer id;
+
+    /**
+     * 消息类型 : RMQ / KAFKA
+     */
+    @Column(name = "msg_type")
+    private String msgType;
+
+    @Column(name="group_id")
+    private String groupId;
+
+    /**
+     * 应用的名称
+     */
+    @Column(name = "application_name")
+    private String applicationName;
+
+    /**
+     * 应用的编号
+     */
+    @Column(name = "application_ip")
+    private String applicationIp;
+
+    /**
+     * 主题
+     */
+    private String topic;
+
+    /**
+     * rmq是tag,kafka是分区index
+     */
+    @Column(name = "partition_name")
+    private String partitionName;
+
+    /**
+     * 消息的key
+     */
+    @Column(name = "msg_key")
+    private String msgKey;
+
+    /**
+     * rmq是消息编号,kafka是offset
+     */
+    @Column(name = "msg_id")
+    private String msgId;
+
+    /**
+     * 1成功 -1失败
+     */
+    @Column(name = "invoke_status")
+    private Integer invokeStatus;
+
+    /**
+     * 日志编号
+     */
+    @Column(name = "log_id")
+    private String logId;
+
+    /**
+     * 创建时间
+     */
+    @Column(name = "created_time")
+    private Date createdTime;
+
+
+    public String getGroupId() {
+        return groupId;
+    }
+
+    public void setGroupId(String groupId) {
+        this.groupId = groupId;
+    }
+
+    private static final long serialVersionUID = 1L;
+
+    public Integer getId() {
+        return id;
+    }
+
+    public void setId(Integer id) {
+        this.id = id;
+    }
+
+    public String getMsgType() {
+        return msgType;
+    }
+
+    public void setMsgType(String msgType) {
+        this.msgType = msgType;
+    }
+
+    public String getApplicationName() {
+        return applicationName;
+    }
+
+    public void setApplicationName(String applicationName) {
+        this.applicationName = applicationName;
+    }
+
+    public String getApplicationIp() {
+        return applicationIp;
+    }
+
+    public void setApplicationIp(String applicationIp) {
+        this.applicationIp = applicationIp;
+    }
+
+    public String getTopic() {
+        return topic;
+    }
+
+    public void setTopic(String topic) {
+        this.topic = topic;
+    }
+
+    public String getPartitionName() {
+        return partitionName;
+    }
+
+    public void setPartitionName(String partitionName) {
+        this.partitionName = partitionName;
+    }
+
+    public String getMsgKey() {
+        return msgKey;
+    }
+
+    public void setMsgKey(String msgKey) {
+        this.msgKey = msgKey;
+    }
+
+    public String getMsgId() {
+        return msgId;
+    }
+
+    public void setMsgId(String msgId) {
+        this.msgId = msgId;
+    }
+
+    public Integer getInvokeStatus() {
+        return invokeStatus;
+    }
+
+    public void setInvokeStatus(Integer invokeStatus) {
+        this.invokeStatus = invokeStatus;
+    }
+
+    public String getLogId() {
+        return logId;
+    }
+
+    public void setLogId(String logId) {
+        this.logId = logId;
+    }
+
+    public Date getCreatedTime() {
+        return createdTime;
+    }
+
+    public void setCreatedTime(Date createdTime) {
+        this.createdTime = createdTime;
+    }
+}

elab-mq/src/main/java/com/elab/mq/model/ProducerEntity.java

@@ -1,213 +1,213 @@
-package com.elab.mq.model;
-
-import io.swagger.annotations.ApiModel;
-import io.swagger.annotations.ApiModelProperty;
-
-import javax.persistence.Column;
-import javax.persistence.Id;
-import javax.persistence.Table;
-import java.util.Date;
-
-/**
- * @author liuhx
- * @create 2019/05/16 19:32
- * @email liuhx@elab-plus.com
- **/
-@Table(name = "mq_producer", catalog = "elab_db")
-@ApiModel(description = "消息队列生产者 实体")
-public class ProducerEntity {
-
-    @Id
-    @ApiModelProperty(name = "id", value = "主键")
-    private Integer id;
-
-    @Column(name = "module_name")
-    @ApiModelProperty(name = "moduleName", value = "微服务模块名")
-    private String moduleName;
-
-    @Column(name = "house_id")
-    @ApiModelProperty(name = "houseId", value = "项目id")
-    private Integer houseId;
-
-    @Column(name = "producer_status")
-    @ApiModelProperty(name = "producerStatus", value = "生产者状态 1成功-1失败")
-    private Integer producerStatus;
-
-    @Column(name = "group_id")
-    @ApiModelProperty(name = "groupId", value = "队列的groupId")
-    private String groupId;
-
-    @Column(name = "topic_id")
-    @ApiModelProperty(name = "topicId", value = "队列通道id")
-    private String topicId;
-
-    @Column(name = "content")
-    @ApiModelProperty(name = "content", value = "内容")
-    private String content;
-
-    @Column(name = "result_content")
-    @ApiModelProperty(name = "resultContent", value = "返回结果内容")
-    private String resultContent;
-
-    @Column(name = "cat_id")
-    @ApiModelProperty(name = "catId", value = "链路编号")
-    private String catId;
-    @Column(name = "tag")
-    @ApiModelProperty(name = "tag", value = "标签名称")
-    private String tag;
-
-    /**
-     * 状态：1  有效  -1  无效
-     */
-    @Column(name = "status")
-    @ApiModelProperty(name = "status", value = "状态：1  有效  -1  无效")
-    private Integer status;
-
-    /**
-     * 创建时间
-     */
-    @Column(name = "created")
-    @ApiModelProperty(name = "created", value = "创建时间")
-    private Date created;
-
-    /**
-     * 修改时间
-     */
-    @Column(name = "updated")
-    @ApiModelProperty(name = "updated", value = "修改时间")
-    private Date updated;
-
-    /**
-     * 创建者
-     */
-    @Column(name = "creator")
-    @ApiModelProperty(name = "creator", value = "创建者")
-    private String creator;
-
-    /**
-     * 修改者
-     */
-    @Column(name = "updator")
-    @ApiModelProperty(name = "updator", value = "修改者")
-    private String updator;
-
-    public String getCatId() {
-        return catId;
-    }
-
-    public void setCatId(String catId) {
-        this.catId = catId;
-    }
-
-    public String getTag() {
-        return tag;
-    }
-
-    public void setTag(String tag) {
-        this.tag = tag;
-    }
-
-    public String getResultContent() {
-        return resultContent;
-    }
-
-    public void setResultContent(String resultContent) {
-        this.resultContent = resultContent;
-    }
-
-    public Integer getId() {
-        return id;
-    }
-
-    public void setId(Integer id) {
-        this.id = id;
-    }
-
-    public String getModuleName() {
-        return moduleName;
-    }
-
-    public void setModuleName(String moduleName) {
-        this.moduleName = moduleName;
-    }
-
-    public Integer getHouseId() {
-        return houseId;
-    }
-
-    public void setHouseId(Integer houseId) {
-        this.houseId = houseId;
-    }
-
-    public Integer getProducerStatus() {
-        return producerStatus;
-    }
-
-    public void setProducerStatus(Integer producerStatus) {
-        this.producerStatus = producerStatus;
-    }
-
-    public String getGroupId() {
-        return groupId;
-    }
-
-    public void setGroupId(String groupId) {
-        this.groupId = groupId;
-    }
-
-    public String getTopicId() {
-        return topicId;
-    }
-
-    public void setTopicId(String topicId) {
-        this.topicId = topicId;
-    }
-
-    public String getContent() {
-        return content;
-    }
-
-    public void setContent(String content) {
-        this.content = content;
-    }
-
-    public Integer getStatus() {
-        return status;
-    }
-
-    public void setStatus(Integer status) {
-        this.status = status;
-    }
-
-    public Date getCreated() {
-        return created;
-    }
-
-    public void setCreated(Date created) {
-        this.created = created;
-    }
-
-    public Date getUpdated() {
-        return updated;
-    }
-
-    public void setUpdated(Date updated) {
-        this.updated = updated;
-    }
-
-    public String getCreator() {
-        return creator;
-    }
-
-    public void setCreator(String creator) {
-        this.creator = creator;
-    }
-
-    public String getUpdator() {
-        return updator;
-    }
-
-    public void setUpdator(String updator) {
-        this.updator = updator;
-    }
-}
+//package com.elab.mq.model;
+//
+//import io.swagger.annotations.ApiModel;
+//import io.swagger.annotations.ApiModelProperty;
+//
+//import javax.persistence.Column;
+//import javax.persistence.Id;
+//import javax.persistence.Table;
+//import java.util.Date;
+//
+///**
+// * @author liuhx
+// * @create 2019/05/16 19:32
+// * @email liuhx@elab-plus.com
+// **/
+//@Table(name = "mq_producer", catalog = "elab_db")
+//@ApiModel(description = "消息队列生产者 实体")
+//public class ProducerEntity {
+//
+//    @Id
+//    @ApiModelProperty(name = "id", value = "主键")
+//    private Integer id;
+//
+//    @Column(name = "module_name")
+//    @ApiModelProperty(name = "moduleName", value = "微服务模块名")
+//    private String moduleName;
+//
+//    @Column(name = "house_id")
+//    @ApiModelProperty(name = "houseId", value = "项目id")
+//    private Integer houseId;
+//
+//    @Column(name = "producer_status")
+//    @ApiModelProperty(name = "producerStatus", value = "生产者状态 1成功-1失败")
+//    private Integer producerStatus;
+//
+//    @Column(name = "group_id")
+//    @ApiModelProperty(name = "groupId", value = "队列的groupId")
+//    private String groupId;
+//
+//    @Column(name = "topic_id")
+//    @ApiModelProperty(name = "topicId", value = "队列通道id")
+//    private String topicId;
+//
+//    @Column(name = "content")
+//    @ApiModelProperty(name = "content", value = "内容")
+//    private String content;
+//
+//    @Column(name = "result_content")
+//    @ApiModelProperty(name = "resultContent", value = "返回结果内容")
+//    private String resultContent;
+//
+//    @Column(name = "cat_id")
+//    @ApiModelProperty(name = "catId", value = "链路编号")
+//    private String catId;
+//    @Column(name = "tag")
+//    @ApiModelProperty(name = "tag", value = "标签名称")
+//    private String tag;
+//
+//    /**
+//     * 状态：1  有效  -1  无效
+//     */
+//    @Column(name = "status")
+//    @ApiModelProperty(name = "status", value = "状态：1  有效  -1  无效")
+//    private Integer status;
+//
+//    /**
+//     * 创建时间
+//     */
+//    @Column(name = "created")
+//    @ApiModelProperty(name = "created", value = "创建时间")
+//    private Date created;
+//
+//    /**
+//     * 修改时间
+//     */
+//    @Column(name = "updated")
+//    @ApiModelProperty(name = "updated", value = "修改时间")
+//    private Date updated;
+//
+//    /**
+//     * 创建者
+//     */
+//    @Column(name = "creator")
+//    @ApiModelProperty(name = "creator", value = "创建者")
+//    private String creator;
+//
+//    /**
+//     * 修改者
+//     */
+//    @Column(name = "updator")
+//    @ApiModelProperty(name = "updator", value = "修改者")
+//    private String updator;
+//
+//    public String getCatId() {
+//        return catId;
+//    }
+//
+//    public void setCatId(String catId) {
+//        this.catId = catId;
+//    }
+//
+//    public String getTag() {
+//        return tag;
+//    }
+//
+//    public void setTag(String tag) {
+//        this.tag = tag;
+//    }
+//
+//    public String getResultContent() {
+//        return resultContent;
+//    }
+//
+//    public void setResultContent(String resultContent) {
+//        this.resultContent = resultContent;
+//    }
+//
+//    public Integer getId() {
+//        return id;
+//    }
+//
+//    public void setId(Integer id) {
+//        this.id = id;
+//    }
+//
+//    public String getModuleName() {
+//        return moduleName;
+//    }
+//
+//    public void setModuleName(String moduleName) {
+//        this.moduleName = moduleName;
+//    }
+//
+//    public Integer getHouseId() {
+//        return houseId;
+//    }
+//
+//    public void setHouseId(Integer houseId) {
+//        this.houseId = houseId;
+//    }
+//
+//    public Integer getProducerStatus() {
+//        return producerStatus;
+//    }
+//
+//    public void setProducerStatus(Integer producerStatus) {
+//        this.producerStatus = producerStatus;
+//    }
+//
+//    public String getGroupId() {
+//        return groupId;
+//    }
+//
+//    public void setGroupId(String groupId) {
+//        this.groupId = groupId;
+//    }
+//
+//    public String getTopicId() {
+//        return topicId;
+//    }
+//
+//    public void setTopicId(String topicId) {
+//        this.topicId = topicId;
+//    }
+//
+//    public String getContent() {
+//        return content;
+//    }
+//
+//    public void setContent(String content) {
+//        this.content = content;
+//    }
+//
+//    public Integer getStatus() {
+//        return status;
+//    }
+//
+//    public void setStatus(Integer status) {
+//        this.status = status;
+//    }
+//
+//    public Date getCreated() {
+//        return created;
+//    }
+//
+//    public void setCreated(Date created) {
+//        this.created = created;
+//    }
+//
+//    public Date getUpdated() {
+//        return updated;
+//    }
+//
+//    public void setUpdated(Date updated) {
+//        this.updated = updated;
+//    }
+//
+//    public String getCreator() {
+//        return creator;
+//    }
+//
+//    public void setCreator(String creator) {
+//        this.creator = creator;
+//    }
+//
+//    public String getUpdator() {
+//        return updator;
+//    }
+//
+//    public void setUpdator(String updator) {
+//        this.updator = updator;
+//    }
+//}

elab-mq/src/main/java/com/elab/mq/msg/ProducerInterceptor.java

@@ -0,0 +1,37 @@
+package com.elab.mq.msg;
+
+import com.aliyun.openservices.ons.api.SendResult;
+import com.elab.mq.model.MessageModel;
+
+/**
+ * @Module TODO
+ * @Description TODO
+ * @Author liukaixiong
+ * @Date 2020/12/30 15:30
+ */
+public interface ProducerInterceptor {
+
+    /**
+     * 检查数据是否符合
+     *
+     * @param messageModel
+     * @return
+     */
+    public boolean check(MessageModel messageModel);
+
+    /**
+     * 成功数据回调
+     *
+     * @param messageModel 消息内容
+     * @param result
+     */
+    public void success(MessageModel messageModel, SendResult result);
+
+    /**
+     * 失败数据回调
+     *
+     * @param messageModel
+     */
+    public void error(MessageModel messageModel, Throwable e);
+
+}

elab-mq/src/main/java/com/elab/mq/msg/impl/MsgProducerImpl.java

@@ -4,20 +4,16 @@ import com.aliyun.openservices.ons.api.SendResult;
 import com.aliyun.openservices.ons.api.bean.ProducerBean;
 import com.dianping.cat.Cat;
 import com.dianping.cat.message.Transaction;
-import com.elab.core.utils.DateUtils;
-import com.elab.core.utils.ObjectUtils;
-import com.elab.core.utils.StringUtils;
 import com.elab.log.utils.CatCrossProcess;
-import com.elab.mq.dao.IProducerDao;
+import com.elab.mq.listener.LoggerConsumerInterceptor;
 import com.elab.mq.model.MessageModel;
-import com.elab.mq.model.ProducerEntity;
 import com.elab.mq.model.SendResultModel;
 import com.elab.mq.msg.IMsgProducerFacade;
+import com.elab.mq.msg.ProducerInterceptor;
 import com.elab.mq.msg.adptor.SendCallbackAdaptor;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Value;
 
 import java.util.Date;
 import java.util.Map;
@@ -32,83 +28,13 @@ import java.util.Properties;
  **/
 public class MsgProducerImpl extends ProducerBean implements IMsgProducerFacade {
 
-
     private Logger logger = LoggerFactory.getLogger(MsgProducerImpl.class);
 
-    @Autowired
-    private IProducerDao producerDao;
-
-    @Value("${spring.application.name}")
-    private String moduleName;
-
-    @Value(value = "${elab.mq.GROUP_ID:}")
-    private String groupId;
+    @Autowired(required = false)
+    private ProducerInterceptor producerInterceptor = new LoggerConsumerInterceptor();
 
     private final String MQ_PRODUCER = "MQ_PRODUCER";
 
-
-    /**
-     * 往生产者表中写入数据
-     *
-     * @param message
-     * @return
-     */
-    private int insertRecordProducer(MessageModel message) {
-        String content = ObjectUtils.objectParseJsonStr(message.getObject(Object.class));
-        ProducerEntity producerEntity = new ProducerEntity();
-        String tag = message.getTag();
-        if (StringUtils.isEmpty(message.getModuleName())) {
-            producerEntity.setModuleName(moduleName);
-        } else {
-            producerEntity.setModuleName(message.getModuleName());
-        }
-        producerEntity.setTag(tag);
-        producerEntity.setCatId(Cat.getCurrentMessageId());
-        producerEntity.setGroupId(groupId);
-        producerEntity.setHouseId(message.getHouseId());
-        producerEntity.setProducerStatus(0);
-        producerEntity.setGroupId(message.getGroupId());
-        producerEntity.setTopicId(message.getTopic());
-        if (content != null) {
-            if (content.length() < 10000) {
-                producerEntity.setContent(content);
-            } else {
-                producerEntity.setContent(content.substring(0, 10000));
-            }
-        }
-        producerEntity.setStatus(1);
-        producerEntity.setCreated(DateUtils.getCurrentDateTime());
-        int id = 0;
-        try {
-            id = producerDao.insert(producerEntity);
-        } catch (Exception e) {
-            e.printStackTrace();
-            logger.error("往生产者表中插入记录失败：，入参：" + ObjectUtils.objectParseJsonStr(message), e);
-        }
-        return id;
-    }
-
-    /**
-     * 修改生产者数据
-     *
-     * @param id
-     * @param result
-     * @return
-     */
-    private void updateRecordProducer(int id, Integer status, String result) {
-        try {
-            ProducerEntity producerEntity = producerDao.selectById(id + "");
-            if (producerEntity != null) {
-                producerEntity.setProducerStatus(status);
-                producerEntity.setResultContent(result);
-                producerDao.updateById(producerEntity);
-            }
-        } catch (Exception e) {
-            e.printStackTrace();
-            logger.error("修改生产者表中记录失败：，入参：" + id, e);
-        }
-    }
-
     @Override
     public void sendAsync(MessageModel message, SendCallbackAdaptor sendCallback) {
         logger.debug(" 发送一条异步消息 " + message + " -> " + message.toString() + " 回调类 : " + sendCallback.getClass().getName());
@@ -121,33 +47,33 @@ public class MsgProducerImpl extends ProducerBean implements IMsgProducerFacade
         String tag = message.getTag();
         Transaction t = Cat.newTransaction(MQ_PRODUCER, topic + "_" + tag + "_" + getClass().getSimpleName());
         logger.debug(" 发送一条消息 " + message.getMsgID() + " -> " + message.toString());
-        int id = insertRecordProducer(message);
+
+        // 提前插入消息，为了让消息编号能够带过去，如果发送失败，则根据编号更改为-1状态。
+        // int id = insertRecordProducer(message);
         try {
+
             catTraceId(message);
-            message.setProducerId(id + "");
+
+            boolean check = producerInterceptor.check(message);
+            if (!check) {
+                return null;
+            }
+
             SendResult result = super.send(message);
-            //updateRecordProducer(id, 1, result.toString());
             logger.debug(" 消息发送结果 : " + result.toString());
             t.setSuccessStatus();
+            producerInterceptor.success(message, result);
             return new SendResultModel(result);
         } catch (Exception e) {
             logger.error("消息发送异常", e);
             t.setStatus(e);
-            //updateRecordProducer(id, -1, e.getMessage());
+            producerInterceptor.error(message, e);
         } finally {
             t.complete();
         }
         return null;
     }
 
-    private void catTraceId(MessageModel message) {
-        // 植入Cat链路编号
-        Map<String, String> msgContextMap = CatCrossProcess.getMsgContextMap();
-        msgContextMap.forEach((K, V) -> {
-            message.putUserProperties(K, V);
-        });
-    }
-
     @Override
     public void sendOneway(MessageModel message) {
         logger.debug(" 发送一条单向消息 " + message.getMsgID() + " -> " + message.toString());
@@ -174,5 +100,12 @@ public class MsgProducerImpl extends ProducerBean implements IMsgProducerFacade
         super.setProperties(properties);
     }
 
+    private void catTraceId(MessageModel message) {
+        // 植入Cat链路编号
+        Map<String, String> msgContextMap = CatCrossProcess.getMsgContextMap();
+        msgContextMap.forEach((K, V) -> {
+            message.putUserProperties(K, V);
+        });
+    }
 
 }

elab-redis/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>elab-parent</artifactId>
         <groupId>com.elab.core</groupId>
-        <version>2.0.5.1-SNAPSHOT</version>
+        <version>2.0.5.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
@@ -21,7 +21,6 @@
         <dependency>
             <groupId>javax.validation</groupId>
             <artifactId>validation-api</artifactId>
-            <version>2.0.1.Final</version>
         </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>

elab-redis/src/main/java/com/elab/redis/annotation/CacheLoopSubmit.java

@@ -15,7 +15,6 @@ public @interface CacheLoopSubmit {
      */
     String cacheName() default "";
 
-
     /**
      * 重复提交的标识key.
      * ALL如果是实体对象那么，对比所有属性。
@@ -27,18 +26,24 @@ public @interface CacheLoopSubmit {
     String[] unionKey() default {};
 
     /**
-     * 锁超时时间(秒)
+     * 锁超时时间(ms)
      *
      * @return
      */
-    int timeOut() default 5;
+    long timeOut() default 5000;
 
     /**
-     * 优先级
+     * 是否等待完成
      *
-     * @return
+     * @return true 尝试抢锁,未抢到等待timeOut的超时时间，超过到则释放， false:没抢到锁直接跳过
      */
-    int order() default Integer.MAX_VALUE;
+    boolean isWaitComplete() default false;
 
+    /**
+     * 错误提示
+     *
+     * @return
+     */
+    String errorMsg() default "系统认定为重复请求";
 
 }

elab-redis/src/main/java/com/elab/redis/config/CacheAutoConfiguration.java

@@ -1,5 +1,6 @@
 package com.elab.redis.config;
 
+import com.alibaba.fastjson.support.spring.GenericFastJsonRedisSerializer;
 import com.elab.redis.CacheTemplate;
 import com.elab.redis.annotation.CacheLoopSubmit;
 import com.elab.redis.annotation.CacheReadLock;
@@ -12,11 +13,16 @@ import com.elab.redis.spring.data.RedisTemplateDecorator;
 import org.redisson.api.RedissonClient;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cache.CacheManager;
-import org.springframework.context.annotation.*;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Role;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.core.RedisTemplate;
-import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
 import org.springframework.data.redis.serializer.StringRedisSerializer;
 
 /**
@@ -26,7 +32,7 @@ import org.springframework.data.redis.serializer.StringRedisSerializer;
  * @time ： 2020/7/8 - 19:12
  */
 @Configuration
-@Import({ElabRedisProperties.class})
+@EnableConfigurationProperties({RedisProperties.class, ElabRedisProperties.class})
 @ComponentScan(value = {"com.elab.redis.interceptor.impl"})
 public class CacheAutoConfiguration {
 
@@ -40,12 +46,17 @@ public class CacheAutoConfiguration {
     }
 
     @Bean
+    @ConditionalOnMissingBean(value = {RedisTemplate.class})
     public RedisTemplate<String, Object> redisTemplate(
             RedisConnectionFactory redisConnectionFactory) throws Exception {
         RedisTemplate<String, Object> template = new RedisTemplateDecorator<>();
         template.setConnectionFactory(redisConnectionFactory);
         template.setKeySerializer(new StringRedisSerializer());
-        template.setValueSerializer(new GenericJackson2JsonRedisSerializer());
+//        ObjectMapper objectMapper = new ObjectMapper();
+//        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
+//        objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
+//        GenericJackson2JsonRedisSerializer genericJackson2JsonRedisSerializer = new GenericJackson2JsonRedisSerializer(objectMapper);
+        template.setValueSerializer(new GenericFastJsonRedisSerializer());
         return template;
     }
 

elab-redis/src/main/java/com/elab/redis/exceptions/ReSubmitException.java

@@ -0,0 +1,20 @@
+package com.elab.redis.exceptions;
+
+import com.elab.core.exception.BusinessException;
+
+/**
+ * @Module 异常管理
+ * @Description 重复提交异常
+ * @Author liukaixiong
+ * @Date 2021/1/4 17:23
+ */
+public class ReSubmitException extends BusinessException {
+
+    public ReSubmitException(String message) {
+        super(message);
+    }
+
+    public ReSubmitException(String errorCode, String message) {
+        super(errorCode, message);
+    }
+}

elab-redis/src/main/java/com/elab/redis/interceptor/impl/CacheLoopProcessImpl.java

@@ -1,10 +1,10 @@
 package com.elab.redis.interceptor.impl;
 
 import com.elab.redis.annotation.CacheLoopSubmit;
+import com.elab.redis.exceptions.ReSubmitException;
 import com.elab.redis.interceptor.ICacheProcessService;
 import com.elab.redis.utils.CacheParseUtil;
 import org.redisson.api.RLock;
-import org.redisson.api.RSet;
 import org.redisson.api.RedissonClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -40,27 +40,56 @@ public class CacheLoopProcessImpl implements ICacheProcessService {
         Method method = invocation.getMethod();
         CacheLoopSubmit cacheLoopSubmit = AnnotationUtils.getAnnotation(method, CacheLoopSubmit.class);
         String[] unionKey = cacheLoopSubmit.unionKey();
-        int timeOut = cacheLoopSubmit.timeOut();
         String clazzName = method.getDeclaringClass().getName();
         Object[] arguments = invocation.getArguments();
 
         String cacheKey = CacheParseUtil.generateUnionKey(unionKey, arguments);
         String lockKey = clazzName + "." + method.getName();
 
-        RSet<Object> lockObject = client.getSet(lockKey);
-        RLock lock = lockObject.getLock(cacheKey);
+        RLock lockObject = client.getLock(lockKey);
+//        RSet<Object> lockObject = client.getSet(lockKey);
+//        RLock lock = lockObject.getLock(cacheKey);
+
+        String errorMsg = cacheLoopSubmit.errorMsg();
+
         try {
-            if (lock.tryLock(timeOut, TimeUnit.SECONDS)) {
+            if (tryLock(lockObject, cacheLoopSubmit)) {
                 Object proceed = invocation.proceed();
                 return proceed;
+            } else {
+                throw new ReSubmitException(errorMsg);
             }
         } catch (InterruptedException e) {
             logger.warn("尝试获取锁超时", e);
         } finally {
-            lock.unlock();
+            // 如果持锁线程是当前线程，则释放锁，非当前线程则不处理
+            if (lockObject.isLocked() && lockObject.isHeldByCurrentThread()) {
+                lockObject.unlock();
+            }
         }
         return null;
     }
 
+    /**
+     * 尝试获取锁
+     *
+     * @param lock            锁对象
+     * @param cacheLoopSubmit 注解
+     * @return
+     */
+    private boolean tryLock(RLock lock, CacheLoopSubmit cacheLoopSubmit) {
+        try {
+            if (true) {
+//            if (cacheLoopSubmit.isWaitComplete()) {
+                return lock.tryLock(cacheLoopSubmit.timeOut(), TimeUnit.MILLISECONDS);
+            } else {
+                return lock.tryLock();
+            }
+        } catch (Exception e) {
+            logger.warn("获取锁异常:" + e.getMessage());
+            return false;
+        }
+    }
+
 
 }

elab-redis/src/test/java/com/elab/redis/cache/CacheTest.java

@@ -1,12 +1,13 @@
 package com.elab.redis.cache;
 
 import com.alibaba.fastjson.JSONObject;
-import com.elab.core.utils.RandomUtils;
 import com.elab.redis.RedisSpringBoot;
 import com.elab.redis.redisson.doc.model.SomeObject;
 import com.elab.redis.service.IDemoService;
 import com.elab.redis.service.impl.DemoServiceImpl;
 import org.junit.Test;
+import org.redisson.api.RLock;
+import org.redisson.api.RedissonClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -24,12 +25,24 @@ public class CacheTest extends RedisSpringBoot {
     @Autowired
     private IDemoService demoService;
 
+    @Autowired
+    private RedissonClient client;
+
+    @Test
+    public void testWatchDog() throws InterruptedException {
+        RLock abc = client.getLock("abc");
+        // 获取锁
+        abc.lock();
+        // 模拟业务超长时间，然后去关注redis的key的过期时间。看是否发生了变化。
+        Thread.sleep(1000000);
+    }
+
     @Test
     public void testCache() throws Exception {
         logger.info(" 开始进入竞争场景 ...");
-        run(10, () -> {
+        run(1, () -> {
             try {
-                demoService.submit("abc_" + RandomUtils.randomString(5));
+                demoService.submit("abc_");
             } catch (Exception e) {
                 e.printStackTrace();
             }

elab-redis/src/test/resources/application.yml

@@ -1,6 +1,6 @@
 spring:
   redis:
-    database: 0
+    database: 4
     host: r-uf6e60u5cmlj0sx563pd.redis.rds.aliyuncs.com
     password: xcGm4kTks6
     port: 6379

elab-rocketMQ/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>elab-parent</artifactId>
         <groupId>com.elab.core</groupId>
-        <version>2.0.5.1-SNAPSHOT</version>
+        <version>2.0.5.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

elab-spring/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>elab-parent</artifactId>
         <groupId>com.elab.core</groupId>
-        <version>2.0.5.1-SNAPSHOT</version>
+        <version>2.0.5.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
@@ -25,17 +25,18 @@
     <!--<version>2.0.1-SNAPSHOT</version>-->
 
     <dependencies>
-        <dependency>
-            <groupId>com.elab.core</groupId>
-            <artifactId>elab-core</artifactId>
-            <version>${project.version}</version>
-        </dependency>
+<!--        <dependency>-->
+<!--            <groupId>com.elab.core</groupId>-->
+<!--            <artifactId>elab-core</artifactId>-->
+<!--            <version>${project.version}</version>-->
+<!--        </dependency>-->
 
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-autoconfigure</artifactId>
             <scope>provided</scope>
         </dependency>
+
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-configuration-processor</artifactId>
@@ -49,6 +50,26 @@
             <version>${project.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>com.elab.core</groupId>
+            <artifactId>elab-mq</artifactId>
+            <version>${project.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>com.elab.core</groupId>
+            <artifactId>elab-redis</artifactId>
+            <version>${project.version}</version>
+            <exclusions>
+                <exclusion>
+                    <artifactId>jackson-core</artifactId>
+                    <groupId>com.fasterxml.jackson.core</groupId>
+                </exclusion>
+            </exclusions>
+            <!--            <scope>provided</scope>-->
+        </dependency>
+
         <dependency>
             <groupId>com.elab.core</groupId>
             <artifactId>elab-db</artifactId>
@@ -123,6 +144,12 @@
             <groupId>io.springfox</groupId>
             <artifactId>springfox-swagger2</artifactId>
             <scope>compile</scope>
+            <exclusions>
+                <exclusion>
+                    <artifactId>swagger-models</artifactId>
+                    <groupId>io.swagger</groupId>
+                </exclusion>
+            </exclusions>
         </dependency>
 
         <dependency>
@@ -140,16 +167,12 @@
             <artifactId>hibernate-validator</artifactId>
             <scope>compile</scope>
         </dependency>
-        <dependency>
-            <groupId>org.junit.jupiter</groupId>
-            <artifactId>junit-jupiter-api</artifactId>
-            <version>RELEASE</version>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>junit</groupId>
-            <artifactId>junit</artifactId>
-        </dependency>
+<!--        <dependency>-->
+<!--            <groupId>org.junit.jupiter</groupId>-->
+<!--            <artifactId>junit-jupiter-api</artifactId>-->
+<!--            <version>RELEASE</version>-->
+<!--            <scope>test</scope>-->
+<!--        </dependency>-->
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>

elab-spring/src/main/java/com/elab/spring/config/RmqConfiguration.java

@@ -0,0 +1,57 @@
+package com.elab.spring.config;
+
+import com.elab.core.aop.annotations.EnableElabDB;
+import com.elab.mq.listener.AbstractMessageListener;
+import com.elab.mq.listener.ConsumerInterceptor;
+import com.elab.mq.msg.IMsgProducerFacade;
+import com.elab.mq.msg.ProducerInterceptor;
+import com.elab.spring.dao.IConsumerDao;
+import com.elab.spring.dao.IProducerDao;
+import com.elab.spring.intercept.RMQCacheConsumerInterceptor;
+import com.elab.spring.intercept.RMQCacheProducerInterceptor;
+import com.elab.spring.intercept.RMQMsgTableConsumerInterceptor;
+import com.elab.spring.intercept.RMQMsgTableProducerInterceptor;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @Module TODO
+ * @Description TODO
+ * @Author liukaixiong
+ * @Date 2020/12/30 15:15
+ */
+@Configuration
+@EnableElabDB(basePackageClasses = {IConsumerDao.class, IProducerDao.class})
+public class RmqConfiguration {
+
+    @Bean
+    @ConditionalOnMissingBean(value = {ConsumerInterceptor.class})
+//    @ConditionalOnBean(value = {CacheTemplate.class, AbstractMessageListener.class})
+    public ConsumerInterceptor cacheConsumerInterceptor() {
+        return new RMQCacheConsumerInterceptor();
+    }
+
+    @Bean
+    @ConditionalOnMissingBean(value = {ProducerInterceptor.class})
+//    @ConditionalOnBean(value = {CacheTemplate.class, IMsgProducerFacade.class})
+    public ProducerInterceptor cacProducerInterceptor() {
+        return new RMQCacheProducerInterceptor();
+    }
+
+    @Bean
+    @ConditionalOnMissingBean(value = {ConsumerInterceptor.class})
+    @ConditionalOnBean(value = {AbstractMessageListener.class})
+    public ConsumerInterceptor msgConsumerInterceptor() {
+        return new RMQMsgTableConsumerInterceptor();
+    }
+
+    @Bean
+    @ConditionalOnMissingBean(value = {ProducerInterceptor.class})
+    @ConditionalOnBean(value = {IMsgProducerFacade.class})
+    public ProducerInterceptor msgProducerInterceptor() {
+        return new RMQMsgTableProducerInterceptor();
+    }
+
+}

elab-spring/src/main/java/com/elab/spring/dao/IConsumerDao.java

@@ -0,0 +1,14 @@
+package com.elab.spring.dao;
+
+import com.elab.core.aop.annotations.XmlGroupName;
+import com.elab.core.dao.IBaseDaoSupport;
+import com.elab.mq.model.MqConsumerLogEntity;
+
+/**
+ * @author liuhx
+ * @create 2019/05/17 18:36
+ * @email liuhx@elab-plus.com
+ **/
+@XmlGroupName("consumer")
+public interface IConsumerDao extends IBaseDaoSupport<MqConsumerLogEntity> {
+}

elab-spring/src/main/java/com/elab/spring/dao/IProducerDao.java

@@ -0,0 +1,15 @@
+package com.elab.spring.dao;
+
+import com.elab.core.aop.annotations.XmlGroupName;
+import com.elab.core.dao.IBaseDaoSupport;
+import com.elab.mq.model.MqProducerLogEntity;
+
+/**
+ * @author liuhx
+ * @create 2019/05/16 19:42
+ * @email liuhx@elab-plus.com
+ **/
+@XmlGroupName("producer")
+public interface IProducerDao extends IBaseDaoSupport<MqProducerLogEntity> {
+
+}

elab-spring/src/main/java/com/elab/spring/intercept/RMQCacheConsumerInterceptor.java

@@ -0,0 +1,141 @@
+package com.elab.spring.intercept;
+
+import com.elab.mq.consts.MqConstants;
+import com.elab.mq.listener.ConsumerInterceptor;
+import com.elab.mq.model.MessageModel;
+import com.elab.redis.config.ElabRedisProperties;
+import com.jay.monitor.data.client.MonitorSendProducer;
+import com.jay.monitor.data.client.utils.MonitorUtils;
+import com.jay.monitor.data.core.model.serializable.MQDataDTO;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.data.redis.core.RedisTemplate;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @Module RMQ拦截器
+ * @Description 缓存去做拦截器
+ * @Author liukaixiong
+ * @Date 2020/12/30 15:27
+ */
+public class RMQCacheConsumerInterceptor implements ConsumerInterceptor {
+
+    private Logger logger = LoggerFactory.getLogger(getClass());
+
+    @Autowired
+    private RedisTemplate<String, Object> redisTemplate;
+
+    @Value(value = "${elab.mq.GROUP_ID:}")
+    private String groupId;
+
+    @Autowired
+    private ElabRedisProperties elabRedisProperties;
+
+    private String defaultPrefix = "RMQConsumer";
+
+    /**
+     * 消费中的时间，避免服务挂掉了，负载均衡到另一台
+     */
+    private Long maxWaitTime = TimeUnit.SECONDS.toMillis(100);
+    /**
+     * 消费完成之后，多久的过期时间
+     */
+    private Long successValidTime = TimeUnit.MINUTES.toMillis(30);
+
+    @Override
+    public boolean check(MessageModel messageModel) {
+        /**
+         * 0: 消费中
+         * 1: 消费成功
+         * > 1 : 失败次数
+         *
+         * 考虑到的情况:
+         * 1. 消费慢
+         * - 过了超时时间，但是没有超过RMQ的ConsumeTimeout确认超时时间。没关系，等提交确认状态就好了,因为同一条消息只会被同一个组的一个线程消费.
+         * - 过了超时时间，也超过了RMQ的ConsumeTimeout确认超时时间。
+         * --- RMQ会执行重试，这个时候就依赖业务这种需求做幂等。(比较极端)
+         * 2. 服务宕机: 这个时候肯定不会提交ACK状态,直接走重试逻辑就好.
+         * - 2.1 -> 等待key超时就行了
+         * 3. RMQ确认超时
+         * --- 这个没有关系,只要业务成功了,会改缓存的状态.
+         * 4. 消费失败 ， 重试
+         * ---
+         *
+         */
+        String key = getCacheKey(messageModel);
+        Object statusObject = redisTemplate.opsForValue().get(key);
+        if (statusObject == null) {
+            // 如果maxWaitTime还没消费完，只有两种情况：1.服务挂掉了， 2. 消费慢
+            /**
+             * 1. 挂掉了 : 等待key超时就行了
+             * 2. 消费慢 :
+             * - 2.1 : RMQ认为消费超时了,RMQ的消费确认时间肯定大于这个maxWaitTime时间就行.
+             * - 2.2 : 确实消费慢,那key超时了没关系,只要maxWaitTime小于RMQ超时确认时间
+             */
+            redisTemplate.opsForValue().set(key, MqConstants.MSG_WAIT, maxWaitTime);
+            logger.debug("缓存幂等校验通过:未发现值:" + key);
+            return true;
+        } else {
+            Integer status = Integer.valueOf(statusObject.toString());
+            if (MqConstants.MSG_OK.equals(status)) {
+                /**
+                 * 到达这里的话，只有可能是出现超时重试的情况。
+                 * 实际消费成功了，但是没有确认给RMQ。
+                 */
+                logger.info("该数据已经被消费过了 : " + key);
+                return false;
+            } else if (MqConstants.MSG_WAIT.equals(status)) {
+                // 几乎不可能触发.
+                throw new RuntimeException("等待中未消费完成的数据");
+            } else {
+                Long count = redisTemplate.opsForValue().increment(key, 1);
+                logger.debug("缓存幂等校验通过 -> 重试 : " + count);
+                return true;
+            }
+        }
+    }
+
+    private String getCacheKey(MessageModel messageModel) {
+        String producerId = messageModel.getProducerId();
+        String prefixWith = elabRedisProperties.getPrefixWith();
+        StringBuilder key = new StringBuilder();
+        key.append(prefixWith).append(":");
+        key.append(defaultPrefix).append(":");
+        key.append(groupId).append(":");
+        key.append(producerId);
+        return key.toString();
+    }
+
+    @Override
+    public void success(MessageModel messageModel) {
+        String key = getCacheKey(messageModel);
+        redisTemplate.opsForValue().set(key, MqConstants.MSG_OK, maxWaitTime, TimeUnit.SECONDS);
+        sendMonitorData(messageModel, 1, messageModel.getInvokeTime());
+    }
+
+    @Override
+    public void error(MessageModel messageModel, Throwable e) {
+        String key = getCacheKey(messageModel);
+        Integer currentStatus = (Integer) redisTemplate.opsForValue().get(key);
+        if (currentStatus <= 1) {
+            redisTemplate.opsForValue().set(key, 2);
+        }
+        sendMonitorData(messageModel, -1, messageModel.getInvokeTime());
+    }
+
+    /**
+     * 发送监控数据
+     *
+     * @param data        数据内容
+     * @param status      状态
+     * @param requestTime 请求时长
+     */
+    private void sendMonitorData(MessageModel data, Integer status, long requestTime) {
+        String content = (String) data.getObject(String.class);
+        MQDataDTO mqDataDTO = MonitorUtils.builderConsumerRMQDataDTO(data.getTopic(), data.getKey(), data.getTag(), data.getMsgID(), content, status, requestTime);
+        MonitorSendProducer.sendMsg(data.getTopic() + "-" + data.getTag(), mqDataDTO);
+    }
+}

elab-spring/src/main/java/com/elab/spring/intercept/RMQCacheProducerInterceptor.java

@@ -0,0 +1,61 @@
+package com.elab.spring.intercept;
+
+import com.aliyun.openservices.ons.api.SendResult;
+import com.elab.mq.model.MessageModel;
+import com.elab.mq.msg.ProducerInterceptor;
+import com.elab.redis.CacheTemplate;
+import com.jay.monitor.data.client.MonitorSendProducer;
+import com.jay.monitor.data.client.utils.MonitorUtils;
+import com.jay.monitor.data.core.model.serializable.MQDataDTO;
+import org.springframework.beans.factory.annotation.Autowired;
+
+/**
+ * @Module rmq拦截器
+ * @Description 基于缓存实现
+ * @Author liukaixiong
+ * @Date 2020/12/30 17:25
+ */
+public class RMQCacheProducerInterceptor implements ProducerInterceptor {
+
+    /**
+     * 所有客户端共用这一个唯一编号
+     */
+    private String defaultPrefix = "RMQProducer:index";
+
+    @Autowired
+    private CacheTemplate cacheTemplate;
+
+    @Override
+    public boolean check(MessageModel messageModel) {
+        // 由于多个客户端发送，如果区分开的话就得维护多个
+        String cacheKey = getCacheKey();
+        Long producerId = cacheTemplate.string().increment(cacheKey, 1);
+        messageModel.setProducerId(producerId.toString());
+        return true;
+    }
+
+    @Override
+    public void success(MessageModel messageModel, SendResult result) {
+        sendMonitorData(messageModel, 1);
+    }
+
+    @Override
+    public void error(MessageModel messageModel, Throwable e) {
+        sendMonitorData(messageModel, -1);
+    }
+
+    private void sendMonitorData(MessageModel data, Integer status) {
+        String content = (String) data.getObject(String.class);
+        MQDataDTO mqDataDTO = MonitorUtils.builderProducerRMQDataDTO(data.getTopic(), data.getKey(), data.getTag(), data.getMsgID(), content, status);
+        MonitorSendProducer.sendMsg(data.getTopic() + "-" + data.getTag(), mqDataDTO);
+    }
+
+
+    private String getCacheKey() {
+//        String prefixWith = elabRedisProperties.getPrefixWith() == null ? "unknown" : elabRedisProperties.getPrefixWith();
+//        StringBuilder key = new StringBuilder();
+//        key.append(prefixWith).append(":");
+//        key.append(defaultPrefix);
+        return defaultPrefix;
+    }
+}

elab-spring/src/main/java/com/elab/spring/intercept/RMQMsgTableConsumerInterceptor.java

@@ -0,0 +1,138 @@
+package com.elab.spring.intercept;
+
+import com.alibaba.fastjson.JSON;
+import com.dianping.cat.Cat;
+import com.elab.core.utils.DataUtils;
+import com.elab.mq.consts.MqConstants;
+import com.elab.mq.listener.ConsumerInterceptor;
+import com.elab.mq.model.MessageModel;
+import com.elab.mq.model.MqConsumerLogEntity;
+import com.elab.spring.dao.IConsumerDao;
+import com.jay.monitor.data.client.MonitorSendProducer;
+import com.jay.monitor.data.client.utils.MonitorUtils;
+import com.jay.monitor.data.core.model.serializable.MQDataDTO;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+
+import java.net.InetAddress;
+import java.util.Date;
+
+/**
+ * @Module 拦截器
+ * @Description 消费者表拦截器
+ * @Author liukaixiong
+ * @Date 2020/12/30 15:16
+ */
+public class RMQMsgTableConsumerInterceptor implements ConsumerInterceptor {
+
+    private Logger logger = LoggerFactory.getLogger(getClass());
+
+    @Value("${spring.application.name}")
+    private String applicationName;
+
+    @Value(value = "${elab.mq.GROUP_ID:}")
+    private String groupId;
+
+    @Autowired
+    private IConsumerDao consumerDao;
+
+    @Override
+    public boolean check(MessageModel messageModel) {
+        return checkRepeatMsg(messageModel);
+    }
+
+    @Override
+    public void success(MessageModel messageModel) {
+        saveConsumerStatus(messageModel, MqConstants.MSG_OK, messageModel.getInvokeTime());
+    }
+
+    @Override
+    public void error(MessageModel messageModel, Throwable e) {
+        saveConsumerStatus(messageModel, MqConstants.MSG_NO, messageModel.getInvokeTime());
+    }
+
+    /**
+     * 检查是否存在重复消费的消息
+     *
+     * @param message
+     * @return
+     * @throws Exception
+     */
+    protected boolean checkRepeatMsg(MessageModel message) {
+        Integer producerId = 0;
+
+        if (message.getProducerId() != null) {
+            producerId = Integer.valueOf(message.getProducerId());
+        }
+
+        // 查询消息是否存在
+        MqConsumerLogEntity consumerEntity = new MqConsumerLogEntity();
+        consumerEntity.setProducerLogId(producerId);
+        consumerEntity.setGroupId(groupId);
+        consumerEntity.setModuleMethod(getClass().getName());
+        consumerEntity.setInvokeStatus(MqConstants.MSG_OK);
+        // 这里其实可以优化成只查询id是否存在
+        Integer repeatCount = null;
+        try {
+            repeatCount = consumerDao.selectByObjectToCount(consumerEntity);
+            if (repeatCount != null && repeatCount > 0) {
+                logger.warn("消费者验证失败，已经消费成功过一次，不允许重复消费 已经成功的总数 : " + repeatCount + " 查询参数:" + JSON.toJSONString(consumerEntity));
+                // throw new BusinessException("请不要重复消费，key：" + message);
+                return false;
+            }
+            logger.debug("消费者验证通过，消费者消费的数据匹配上生产者数据，");
+        } catch (Exception e) {
+            logger.error("消息匹配异常", e);
+        }
+        return true;
+    }
+
+    /**
+     * 消费完成之后的结果保存
+     *
+     * @param message     消息内容
+     * @param status      状态
+     * @param requestTime 请求时长
+     */
+    protected void saveConsumerStatus(MessageModel message, Integer status, long requestTime) {
+        try {
+            Integer producerId = DataUtils.getStringByIntegerValue(message.getProducerId());
+            MqConsumerLogEntity consumerEntity = new MqConsumerLogEntity();
+            consumerEntity.setProducerLogId(producerId);
+            consumerEntity.setGroupId(groupId);
+            consumerEntity.setModuleMethod(getClass().getName());
+            consumerEntity.setInvokeStatus(status);
+            consumerEntity.setCreatedTime(new Date());
+            consumerEntity.setMsgId(message.getMsgID());
+            consumerEntity.setMsgKey(message.getKey());
+            consumerEntity.setMsgType(MqConstants.MQ_TYPE);
+            consumerEntity.setApplicationIp(InetAddress.getLocalHost().getHostAddress());
+            consumerEntity.setApplicationName(applicationName);
+            consumerEntity.setLogId(Cat.getCurrentMessageId());
+            consumerEntity.setPartitionName(message.getTag());
+            consumerEntity.setTopic(message.getTopic());
+            int insert = consumerDao.insert(consumerEntity);
+            //logger.debug("插入kafka去重数据:" + insert);
+        } catch (Exception e) {
+            logger.error("消费记录日志添加失败", e);
+        }
+        // 发送监控数据出去
+        sendMonitorData(message, status, requestTime);
+    }
+
+    /**
+     * 发送监控数据
+     *
+     * @param data        数据内容
+     * @param status      状态
+     * @param requestTime 请求时长
+     */
+    private void sendMonitorData(MessageModel data, Integer status, long requestTime) {
+        String content = (String) data.getObject(String.class);
+        MQDataDTO mqDataDTO = MonitorUtils.builderConsumerRMQDataDTO(data.getTopic(), data.getKey(), data.getTag(), data.getMsgID(), content, status, requestTime);
+        MonitorSendProducer.sendMsg(data.getTopic() + "-" + data.getTag(), mqDataDTO);
+    }
+
+}

elab-spring/src/main/java/com/elab/spring/intercept/RMQMsgTableProducerInterceptor.java

@@ -0,0 +1,122 @@
+package com.elab.spring.intercept;
+
+import com.aliyun.openservices.ons.api.SendResult;
+import com.dianping.cat.Cat;
+import com.elab.core.utils.DateUtils;
+import com.elab.core.utils.ObjectUtils;
+import com.elab.core.utils.StringUtils;
+import com.elab.mq.model.MessageModel;
+import com.elab.mq.model.MqProducerLogEntity;
+import com.elab.mq.msg.ProducerInterceptor;
+import com.elab.spring.dao.IProducerDao;
+import com.jay.monitor.data.client.MonitorSendProducer;
+import com.jay.monitor.data.client.utils.MonitorUtils;
+import com.jay.monitor.data.core.model.serializable.MQDataDTO;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+
+import java.net.InetAddress;
+
+/**
+ * @Module 拦截器
+ * @Description rmq生产者处理
+ * @Author liukaixiong
+ * @Date 2020/12/30 15:36
+ */
+public class RMQMsgTableProducerInterceptor implements ProducerInterceptor {
+    private Logger logger = LoggerFactory.getLogger(getClass());
+
+    @Autowired(required = false)
+    private IProducerDao producerDao;
+
+    @Value("${spring.application.name}")
+    private String moduleName;
+
+    @Value(value = "${elab.mq.GROUP_ID:}")
+    private String groupId;
+
+    @Override
+    public boolean check(MessageModel messageModel) {
+        int id = insertRecordProducer(messageModel);
+        messageModel.setProducerId(id + "");
+        return true;
+    }
+
+    @Override
+    public void success(MessageModel messageModel, SendResult result) {
+        sendMonitorData(messageModel, 1);
+    }
+
+    @Override
+    public void error(MessageModel messageModel, Throwable e) {
+        updateRecordProducer(Integer.valueOf(messageModel.getProducerId()), -1, e.getMessage());
+        sendMonitorData(messageModel, -1);
+    }
+
+    /**
+     * 往生产者表中写入数据
+     *
+     * @param message
+     * @return
+     */
+    private int insertRecordProducer(MessageModel message) {
+        int id = 0;
+        try {
+            MqProducerLogEntity producerEntity = new MqProducerLogEntity();
+            String tag = message.getTag();
+            if (StringUtils.isEmpty(message.getModuleName())) {
+                producerEntity.setApplicationName(moduleName);
+            } else {
+                producerEntity.setApplicationName(message.getModuleName());
+            }
+            producerEntity.setApplicationIp(InetAddress.getLocalHost().getHostAddress());
+            producerEntity.setPartitionName(tag);
+            producerEntity.setLogId(Cat.getCurrentMessageId());
+            producerEntity.setGroupId(groupId);
+            producerEntity.setInvokeStatus(1);
+            producerEntity.setMsgId(message.getMsgID());
+            producerEntity.setMsgKey(message.getKey());
+            producerEntity.setGroupId(message.getGroupId());
+            producerEntity.setTopic(message.getTopic());
+            producerEntity.setCreatedTime(DateUtils.getCurrentDateTime());
+            id = producerDao.insert(producerEntity);
+        } catch (Exception e) {
+            e.printStackTrace();
+            logger.error("往生产者表中插入记录失败：，入参：" + ObjectUtils.objectParseJsonStr(message), e);
+        }
+        return id;
+    }
+
+    /**
+     * 修改生产者数据
+     *
+     * @param id
+     * @param result
+     * @return
+     */
+    private void updateRecordProducer(int id, Integer status, String result) {
+        if (id <= 0) {
+            logger.warn("数据插入失败，无法修改生产消息日志表");
+            return;
+        }
+        try {
+            MqProducerLogEntity producerLogEntity = producerDao.selectById(id + "");
+            if (producerLogEntity != null) {
+                producerLogEntity.setInvokeStatus(status);
+                producerDao.updateById(producerLogEntity);
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+            logger.error("修改生产者表中记录失败：，入参：" + id, e);
+        }
+    }
+
+    private void sendMonitorData(MessageModel data, Integer status) {
+        String content = (String) data.getObject(String.class);
+        MQDataDTO mqDataDTO = MonitorUtils.builderProducerRMQDataDTO(data.getTopic(), data.getKey(), data.getTag(), data.getMsgID(), content, status);
+        MonitorSendProducer.sendMsg(data.getTopic() + "-" + data.getTag(), mqDataDTO);
+    }
+
+}

elab-spring/src/test/java/com/elab/spring/config/SwaggerConfigBeanTest.java

@@ -0,0 +1,37 @@
+package com.elab.spring.config;
+
+import junit.framework.TestCase;
+import org.junit.Test;
+import springfox.documentation.builders.ApiInfoBuilder;
+import springfox.documentation.builders.PathSelectors;
+import springfox.documentation.builders.RequestHandlerSelectors;
+import springfox.documentation.service.ApiInfo;
+import springfox.documentation.spi.DocumentationType;
+import springfox.documentation.spring.web.plugins.Docket;
+
+/**
+ * @Module TODO
+ * @Description TODO
+ * @Author liukaixiong
+ * @Date 2020/12/21 13:48
+ */
+public class SwaggerConfigBeanTest extends TestCase {
+
+
+    @Test
+    public void test1() {
+        new Docket(DocumentationType.SWAGGER_2)
+                .apiInfo(apiInfo())
+                .select()
+                // 具体需要扫描的包
+                .apis(RequestHandlerSelectors.basePackage("com.elab.spring.config"))
+                .paths(PathSelectors.any())
+                .build();
+    }
+
+    private ApiInfo apiInfo() {
+        return new ApiInfoBuilder()
+                .title("aaaa")
+                .build();
+    }
+}

elab-spring/src/test/java/com/elab/spring/intercept/RMQCacheConsumerInterceptorTest.java

@@ -0,0 +1,95 @@
+package com.elab.spring.intercept;
+
+import com.elab.mq.model.MessageModel;
+import com.elab.redis.annotation.EnableElabRedis;
+import com.elab.spring.config.RmqConfiguration;
+import org.json.JSONException;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.redisson.spring.starter.RedissonAutoConfiguration;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+
+/**
+ * @Module TODO
+ * @Description TODO
+ * @Author liukaixiong
+ * @Date 2020/12/31 14:28
+ */
+@RunWith(SpringRunner.class)
+@EnableConfigurationProperties(value = {RedisProperties.class})
+@SpringBootTest(classes = {RmqConfiguration.class, RedissonAutoConfiguration.class})
+@EnableElabRedis
+public class RMQCacheConsumerInterceptorTest {
+
+    @Autowired
+    private RMQCacheConsumerInterceptor rmqCacheConsumerInterceptor;
+
+    @Autowired
+    private RMQCacheProducerInterceptor producerInterceptor;
+
+    @Test
+    public void check() {
+
+    }
+
+    public MessageModel sendMsg() {
+        com.alibaba.fastjson.JSONObject jsonObject = new com.alibaba.fastjson.JSONObject();
+        jsonObject.put("key", "test");
+        MessageModel messageModel = new MessageModel("A", "B", "C", jsonObject.toString());
+        messageModel.setInvokeTime(1000L);
+        producerInterceptor.check(messageModel);
+        return messageModel;
+    }
+
+    @Test
+    public void success() throws JSONException {
+        successProcess();
+    }
+
+    private void successProcess() {
+        MessageModel messageModel = sendMsg();
+        messageModel.setInvokeTime(1000L);
+        boolean check = rmqCacheConsumerInterceptor.check(messageModel);
+        if (check) {
+            rmqCacheConsumerInterceptor.success(messageModel);
+        }
+    }
+
+    @Test
+    public void error() {
+        errorProcess(null);
+    }
+
+    /**
+     * 异常场景触发
+     * @param fixedProducerId   固定的producer编号
+     */
+    private void errorProcess(Integer fixedProducerId) {
+        MessageModel messageModel = sendMsg();
+        messageModel.setInvokeTime(1000L);
+        if (fixedProducerId != null) {
+            messageModel.setProducerId(fixedProducerId + "");
+        }
+        boolean check = rmqCacheConsumerInterceptor.check(messageModel);
+        if (check) {
+            rmqCacheConsumerInterceptor.error(messageModel, null);
+        }
+    }
+
+    @Test
+    public void retryTest() {
+        boolean isSuccess = false;
+        int retryCount = 2;
+        for (int i = 0; i < retryCount; i++) {
+            if (isSuccess) {
+                successProcess();
+            } else {
+                errorProcess(100);
+            }
+        }
+    }
+}

elab-spring/src/test/java/com/elab/spring/intercept/RMQCacheProducerInterceptorTest.java

@@ -0,0 +1,50 @@
+package com.elab.spring.intercept;
+
+import com.alibaba.fastjson.JSONObject;
+import com.elab.mq.model.MessageModel;
+import com.elab.redis.annotation.EnableElabRedis;
+import com.elab.spring.config.RmqConfiguration;
+import junit.framework.TestCase;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.redisson.spring.starter.RedissonAutoConfiguration;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+
+/**
+ * @Module TODO
+ * @Description TODO
+ * @Author liukaixiong
+ * @Date 2020/12/31 15:43
+ */
+@RunWith(SpringRunner.class)
+@EnableConfigurationProperties(value = {RedisProperties.class})
+@SpringBootTest(classes = {RmqConfiguration.class, RedissonAutoConfiguration.class})
+@EnableElabRedis
+public class RMQCacheProducerInterceptorTest extends TestCase {
+
+    @Autowired
+    private RMQCacheProducerInterceptor producerInterceptor;
+
+    @Test
+    public void check() {
+        JSONObject jsonObject = new JSONObject();
+        jsonObject.put("key", "test");
+        MessageModel messageModel = new MessageModel("A", "B", "C", jsonObject.toString());
+        messageModel.setInvokeTime(1000L);
+        boolean check = producerInterceptor.check(messageModel);
+        System.out.println(check);
+    }
+
+    @Test
+    public void success() {
+
+    }
+
+    @Test
+    public void error() {
+    }
+}

elab-spring/src/test/resources/application-dev.yml

@@ -1,5 +1,3 @@
-
-
 default:
   minIdle: 10
   validationQuery: SELECT 1
@@ -7,7 +5,7 @@ default:
   maxWait: 60000
   filters: wall,stat
   poolPreparedStatements: true
-#  url: jdbc:mysql://43.254.221.77/marketing_db?characterEncoding=UTF-8&connectTimeout=60000&socketTimeout=60000
+  #  url: jdbc:mysql://43.254.221.77/marketing_db?characterEncoding=UTF-8&connectTimeout=60000&socketTimeout=60000
   url: jdbc:mysql://192.168.0.13/marketing_db?characterEncoding=UTF-8&connectTimeout=60000&socketTimeout=60000&serverTimezone=Asia/Shanghai
   logAbandoned: true
   password: elab@123
@@ -38,4 +36,26 @@ elab:
     defaultTopic: dev-elab-marketing-user
     dataWorkTopic: dev-data-work-boot
   jinmao:
-      url: http://192.168.0.26:8088/sql/executor
+    url: http://192.168.0.26:8088/sql/executor
+
+spring:
+  redis:
+    database: 0
+    host: r-uf6e60u5cmlj0sx563pd.redis.rds.aliyuncs.com
+    password: xcGm4kTks6
+    port: 6379
+    timeout: 5s
+    pool:
+      # 连接池最大连接数（使用负值表示没有限制）
+      max-active: 64
+      # 连接池中的最大空闲连接
+      max-idle: 64
+      # 连接池最大阻塞等待时间（使用负值表示没有限制）
+      max-wait: -1
+      # 连接池中的最小空闲连接
+      min-idle: 1
+    elab:
+      # 注解的统一超时时间
+      ttl: 30
+      # 注解的统一前缀
+      prefix-with: user_

elab-spring/src/test/resources/logback.xml

@@ -12,7 +12,8 @@
 
     <appender name="cat" class="com.elab.log.log4j.CatLogbackLog"></appender>
 
-    <logger name="com.elab" level="INFO" additivity="false">
+    <logger name="com.elab" level="debug" additivity="false">
+        <appender-ref ref="stdout"/>
         <appender-ref ref="cat"/>
     </logger>
 

pom.xml

@@ -7,7 +7,7 @@
     <groupId>com.elab.core</groupId>
     <artifactId>elab-parent</artifactId>
     <packaging>pom</packaging>
-    <version>2.0.5.1-SNAPSHOT</version>
+    <version>2.0.5.2-SNAPSHOT</version>
     <modules>
         <module>elab-core</module>
         <module>elab-cache</module>
@@ -221,7 +221,7 @@
             <dependency>
                 <groupId>javax.validation</groupId>
                 <artifactId>validation-api</artifactId>
-                <version>1.1.0.Final</version>
+                <version>2.0.1.Final</version>
             </dependency>
             <dependency>
                 <groupId>org.hibernate</groupId>
@@ -274,7 +274,6 @@
             <!--<artifactId>cat-core</artifactId>-->
             <!--<version>2.0.0</version>-->
             <!--</dependency>-->
-
             <dependency>
                 <groupId>junit</groupId>
                 <artifactId>junit</artifactId>
@@ -335,7 +334,7 @@
             <dependency>
                 <groupId>com.jay.monitor.data</groupId>
                 <artifactId>jay-monitor-data-client</artifactId>
-                <version>1.0-SNAPSHOT</version>
+                <version>1.1-SNAPSHOT</version>
             </dependency>
         </dependencies>
     </dependencyManagement>