版本分支提交

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

@@ -37,5 +37,11 @@ public interface ITaskProducer {
      */
     public void sendSchedulingList(SchedulingTaskExecutor schedulingTaskExecutor);
 
+    /**
+     * 聚合运算
+     * @param calculateList
+     * @param function
+     * @return
+     */
     public Map<String, Object> reduce(List<?> calculateList, Function<List<?>, Map<String, Object>> function);
 }

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

@@ -0,0 +1,43 @@
+package com.elab.core.componts;
+
+import java.util.concurrent.CyclicBarrier;
+import java.util.function.Supplier;
+
+/**
+ * @Module 工具类
+ * @Description 并发工具类
+ * @Author liukaixiong
+ * @Date 2021/1/5 11:08
+ */
+public class ConcurrentTool {
+
+    public int count;
+
+    private CyclicBarrier cyclicBarrier;
+
+    public ConcurrentTool(int count) {
+        this.count = count;
+        this.cyclicBarrier = new CyclicBarrier(count);
+    }
+
+    /**
+     * 执行压测方式
+     *
+     * @param supplier
+     */
+    public void process(Supplier supplier) {
+        for (int i = 0; i < count; i++) {
+            new Thread(new Runnable() {
+                @Override
+                public void run() {
+                    try {
+                        cyclicBarrier.await();
+                        Object o = supplier.get();
+                    } catch (Exception e) {
+                        e.printStackTrace();
+                    }
+                }
+            }).start();
+        }
+    }
+}

elab-core/src/test/java/com/elab/core/async/pruducer/TaskProducerTest.java

@@ -1,6 +1,5 @@
 package com.elab.core.async.pruducer;
 
-import com.alibaba.fastjson.JSON;
 import com.elab.core.async.consumer.TaskConsumer;
 import com.elab.core.async.store.TaskExecutorQueue;
 import org.junit.Before;
@@ -9,7 +8,8 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
-import java.util.*;
+import java.util.Arrays;
+import java.util.List;
 
 public class TaskProducerTest {
     private Logger logger = LoggerFactory.getLogger(getClass());
@@ -67,26 +67,30 @@ public class TaskProducerTest {
     public void testReduce() throws Exception {
         List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9);
 
-        Map<String, Object> reduce = producer.reduce(list, (intList) -> {
-            System.out.println("====>>>>开始处理" + JSON.toJSONString(intList));
-            Map<String, Object> map = new HashMap<>();
-            List<Integer> aList = new ArrayList<>();
-            List<Integer> bList = new ArrayList<>();
-
-            for (int i = 0; i < intList.size(); i++) {
-                Integer index = list.get(i);
-                if (index % 2 == 0) {
-                    aList.add(index);
-                } else {
-                    bList.add(index);
-                }
-            }
-
-            map.put("A", aList);
-            map.put("B", bList);
-            return map;
+        list.parallelStream().forEach((data) -> {
+            System.out.println(Thread.currentThread().getName() + "\t" + data.toString());
         });
-        System.out.println("map : " + JSON.toJSONString(reduce));
+
+//        Map<String, Object> reduce = producer.reduce(list, (intList) -> {
+//            System.out.println("====>>>>开始处理" + JSON.toJSONString(intList));
+//            Map<String, Object> map = new HashMap<>();
+//            List<Integer> aList = new ArrayList<>();
+//            List<Integer> bList = new ArrayList<>();
+//
+//            for (int i = 0; i < intList.size(); i++) {
+//                Integer index = list.get(i);
+//                if (index % 2 == 0) {
+//                    aList.add(index);
+//                } else {
+//                    bList.add(index);
+//                }
+//            }
+//
+//            map.put("A", aList);
+//            map.put("B", bList);
+//            return map;
+//        });
+//        System.out.println("map : " + JSON.toJSONString(reduce));
         System.in.read();
 
     }

elab-db/src/main/java/com/elab/core/spring/common/utils/DataUtils.java

@@ -33,7 +33,7 @@ public class DataUtils {
      * mergePO 时支持的数据类型
      */
     private static Map<Class, String> supportTypeMap = new HashMap<Class, String>();
-
+    private static Pattern pattern = Pattern.compile("[0-9]*");
     static {
         //基本数据类型
         supportTypeMap.put(Integer.class, "");
@@ -512,7 +512,7 @@ public class DataUtils {
      */
     public static boolean isNumeric(String str) {
         if (str == null) return false;
-        Pattern pattern = Pattern.compile("[0-9]*");
+
         return pattern.matcher(str).matches();
     }
 

elab-db/src/test/java/com.db.service/ITestService.java

@@ -11,5 +11,5 @@ import com.elab.core.services.ICommonService;
 public interface ITestService extends ICommonService<TTest> {
 
 
-
+    int updateByExample(TTest test);
 }

elab-db/src/test/java/com.db.service/dao/ITestDao.java

@@ -37,4 +37,7 @@ public interface ITestDao extends IBaseDaoSupport<TTest> {
     PageModel<Map> getTestPageListMap(PageModel page, LinkedHashMap params) throws Exception;
 
 
+    public int updateByExample(TTest test);
+
+
 }

elab-db/src/test/java/com.db.service/impl/TestServiceImpl.java

@@ -7,6 +7,7 @@ import com.elab.core.dao.IBaseDaoSupport;
 import com.elab.core.services.impl.CommonServiceAdaptor;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 
 import java.util.List;
 
@@ -35,4 +36,16 @@ public class TestServiceImpl extends CommonServiceAdaptor<TTest> implements ITes
 
         return super.selectByList(obj);
     }
+
+    @Override
+    @Transactional
+    public int updateByExample(TTest test) {
+        int update = testDao.updateByExample(test);
+        try {
+            Thread.sleep(10000);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+        return update;
+    }
 }

elab-db/src/test/java/com.db.service/main/BasicServiceCase.java

@@ -2,21 +2,27 @@ package com.db.service.main;
 
 import com.db.service.ITestService;
 import com.db.service.model.TTest;
+import com.elab.core.componts.ConcurrentTool;
+import com.elab.core.utils.RandomUtils;
 import org.databene.contiperf.PerfTest;
 import org.databene.contiperf.Required;
 import org.databene.contiperf.junit.ContiPerfRule;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.transaction.annotation.EnableTransactionManagement;
 
 import java.util.List;
 
 /**
  * 公共service层公共服务测试用例
+ *
  * @author : liukx
  * @create : 2018/5/17 13:47
  * @email : liukx@elab-plus.com
@@ -24,8 +30,9 @@ import java.util.List;
 @RunWith(SpringJUnit4ClassRunner.class)  //使用junit4进行测试
 @ContextConfiguration
         ({"classpath:applicationContext-*.xml",})
+@EnableTransactionManagement
 public class BasicServiceCase {
-
+    private Logger logger = LoggerFactory.getLogger(getClass());
     @Autowired
     private ITestService testService;
 
@@ -35,9 +42,50 @@ public class BasicServiceCase {
     @Rule
     public ContiPerfRule rule = new ContiPerfRule();
 
+
     @Test
-    @PerfTest(invocations = 10000,threads = 10)
-    @Required(throughput = 1000,max = 1000)
+    public void testDeadLockUpdate() throws Exception {
+        ConcurrentTool concurrentTool = new ConcurrentTool(10);
+        updateTestById();
+        int index = 0;
+        concurrentTool.process(() -> {
+            updateTestById();
+            return null;
+        });
+
+//        concurrentTool.process(() -> {
+//            updateByExample1();
+//            return null;
+//        });
+        System.out.println("处理结束");
+        System.in.read();
+    }
+
+    private void updateTestById() {
+        TTest test = new TTest();
+        test.setLoveName("bbb" + RandomUtils.randomNum(5));
+        test.setId(1);
+        try {
+            testService.updateById(test);
+        } catch (Exception e) {
+            logger.error("修改异常", e);
+        }
+    }
+
+    private void updateByExample1() {
+        TTest test = new TTest();
+        test.setLoveName("bbb" + RandomUtils.randomNum(5));
+        test.setName("bbb26175");
+        try {
+            testService.updateByExample(test);
+        } catch (Exception e) {
+            logger.error("修改异常", e);
+        }
+    }
+
+    @Test
+    @PerfTest(invocations = 10000, threads = 10)
+    @Required(throughput = 1000, max = 1000)
     public void insert() throws Exception {
         TTest test = new TTest();
         test.setName("某某某");
@@ -78,5 +126,4 @@ public class BasicServiceCase {
     }
 
 
-
 }

elab-db/src/test/java/com.db.service/model/TTest.java

@@ -11,7 +11,7 @@ import java.util.List;
 
 //import javax.persistence.Table;
 //implements ColumnMapping
-@Table(name = "t_test",catalog = "test")
+@Table(name = "t_test" )
 public class TTest implements ColumnMapping {
     //
     // 表字段 : t_test.id

elab-db/src/test/resources/applicationContext-datasource.xml

@@ -1,11 +1,12 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <beans xmlns="http://www.springframework.org/schema/beans"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-       xmlns:context="http://www.springframework.org/schema/context"
+       xmlns:context="http://www.springframework.org/schema/context" xmlns:tx="http://www.springframework.org/schema/tx"
        xsi:schemaLocation="http://www.springframework.org/schema/beans
     http://www.springframework.org/schema/beans/spring-beans-4.1.xsd
     http://www.springframework.org/schema/context
-    http://www.springframework.org/schema/context/spring-context.xsd">
+    http://www.springframework.org/schema/context/spring-context.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd">
+    <tx:annotation-driven transaction-manager="dataSourceTransactionManager" />
 
     <context:component-scan base-package="com.db.service.*">
         <context:include-filter type="annotation" expression="org.springframework.stereotype.Service"/>
@@ -14,9 +15,9 @@
 
     <bean id="mysqlDataSource" class="com.alibaba.druid.pool.DruidDataSource">
         <property name="driverClassName" value="com.mysql.jdbc.Driver"/>
-        <property name="url" value="jdbc:mysql://localhost:3306/test?characterEncoding=utf-8"/>
+        <property name="url" value="jdbc:mysql://192.168.0.13:3306/elab_db?characterEncoding=utf-8"/>
         <property name="username" value="root"/>
-        <property name="password" value="1234"/>
+        <property name="password" value="elab@123"/>
         <property name="initialSize" value="50"/>
         <property name="maxActive" value="100"/>
         <property name="minIdle" value="1"/>
@@ -54,4 +55,8 @@
         <property name="configurableFactory" ref="configurableFactory"/>
     </bean>
 
+    <bean id="dataSourceTransactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
+        <property name="dataSource" ref="mysqlDataSource"/>
+    </bean>
+
 </beans>

elab-db/src/test/resources/sql/test-sql.xml

@@ -116,4 +116,8 @@
         and b.organize_id = :organizeId and (a.name like :keyword1 or a.mobile like :keyword2)
         order by ifnull(b.updated,b.created) desc
     </sql>
+
+    <sql id="updateByExample">
+        update t_test set love_name=:loveName  where name = :name
+    </sql>
 </sqlGroup>

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

@@ -10,6 +10,11 @@ import com.jay.monitor.data.core.model.serializable.base.TranceLogData;
 import java.util.Map;
 import java.util.function.BiConsumer;
 
+/**
+ * 消息监控规则
+ *
+ * @author liukaixong
+ */
 public class CatIdMonitorRule implements MonitorRuleCallback {
 
     @Override
@@ -18,11 +23,13 @@ public class CatIdMonitorRule implements MonitorRuleCallback {
             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());
+                if (catCrossIdModel.get_catRootMessageId() != null) {
+                    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/utils/CatCrossProcess.java

@@ -2,6 +2,7 @@ package com.elab.log.utils;
 
 import com.dianping.cat.Cat;
 import com.dianping.cat.message.Event;
+import com.dianping.cat.message.TaggedTransaction;
 import com.dianping.cat.message.Transaction;
 import com.elab.log.ext.FunctionProcess;
 import org.apache.commons.lang3.StringUtils;
@@ -43,17 +44,17 @@ public class CatCrossProcess {
      */
     public static Object createRemoteMsg(String type, String name, Function<Map<String, String>, Object> consumer) {
         // 构建一个事务日志
-        Transaction t = Cat.newTransaction(type, name);
+        Transaction transaction = Cat.newTransaction(type, name);
         try {
             // 构建一个上下文事务消息
             Map<String, String> msgContextMap = getMsgContextMap();
             Object apply = consumer.apply(msgContextMap);
-            t.setSuccessStatus();
+            transaction.setSuccessStatus();
             return apply;
         } catch (Exception e) {
-            t.setStatus(e);
+            transaction.setStatus(e);
         } finally {
-            t.complete();
+            transaction.complete();
         }
         return null;
     }
@@ -68,10 +69,12 @@ public class CatCrossProcess {
      */
     public static Object createRemoteMQMsg(String type, String name, Function<Map<String, String>, Object> consumer) {
         // 构建一个事务日志
-        Transaction t = Cat.newTransaction(type, name);
+        TaggedTransaction t = Cat.newTaggedTransaction(type, name, "MQ");
         try {
-            // 构建一个上下文事务消息
             Map<String, String> msgContextMap = getDomainMsgContextMap("MQ-CONSUMER");
+            t.bind("MQ",msgContextMap.get(Cat.Context.CHILD),"MQ消息请走特殊查询");
+            // 构建一个上下文事务消息
+
             Object apply = consumer.apply(msgContextMap);
             t.setSuccessStatus();
             return apply;
@@ -141,7 +144,8 @@ public class CatCrossProcess {
      */
     public static void createProviderCross(HttpServletRequest request, Transaction t) {
         Event crossAppEvent = Cat.newEvent(CatMsgConstants.PROVIDER_CALL_APP, request.getHeader(CatMsgConstants.APPLICATION_KEY));
-        Event crossServerEvent = Cat.newEvent(CatMsgConstants.PROVIDER_CALL_SERVER, request.getRemoteAddr());    //clientIp
+        //clientIp
+        Event crossServerEvent = Cat.newEvent(CatMsgConstants.PROVIDER_CALL_SERVER, request.getRemoteAddr());
         crossAppEvent.setStatus(Event.SUCCESS);
         crossServerEvent.setStatus(Event.SUCCESS);
         t.addChild(crossAppEvent);
@@ -229,8 +233,8 @@ public class CatCrossProcess {
             Cat.Context context = new CatMsgContext();
             context.addProperty(Cat.Context.ROOT, msgIdMap.get(Cat.Context.ROOT));
             context.addProperty(Cat.Context.PARENT, msgIdMap.get(Cat.Context.PARENT));
-            context.addProperty(Cat.Context.CHILD, Cat.getCurrentMessageId());
             // 这里不需要构建客户端带过来的编号，因为消费者是多个，避免LOG被覆盖的情况。
+            context.addProperty(Cat.Context.CHILD, Cat.getCurrentMessageId());
             Cat.logRemoteCallServer(context);
 
             Event crossAppEvent = Cat.newEvent(CatMsgConstants.PROVIDER_CALL_APP, msgIdMap.get(CatMsgConstants.APPLICATION_KEY));

elab-log/src/test/java/com/elab/log/utils/CatCrossProcessTest.java

@@ -2,6 +2,7 @@ package com.elab.log.utils;
 
 import com.alibaba.fastjson.JSON;
 import com.dianping.cat.Cat;
+import com.dianping.cat.message.TaggedTransaction;
 import junit.framework.TestCase;
 import org.junit.Test;
 import org.slf4j.Logger;
@@ -19,6 +20,47 @@ public class CatCrossProcessTest extends TestCase {
 
     private Logger logger = LoggerFactory.getLogger(getClass());
 
+    @Test
+    public void testCatMsg() throws Exception {
+        CatCrossProcess.createRemoteMQMsg("TEST", "LOG-MSG", (idMap) -> {
+            String logInfo = JSON.toJSONString(idMap);
+            String nodeLogId = idMap.get(Cat.Context.CHILD);
+            String logId = Cat.getCurrentMessageId();
+            logger.info("cat test");
+            logger.info("---------->>>>>" + JSON.toJSONString(idMap));
+            logger.info("---------->>>>>msg id : " + Cat.getCurrentMessageId());
+//            ForkedTransaction forkedTransaction = Cat.newForkedTransaction("fork-test", "remote-msg");
+//            String forkedMessageId = forkedTransaction.getForkedMessageId();
+//            forkedTransaction.setSuccessStatus();
+//            logger.info("----forkedMessageId----" + forkedMessageId);
+//            forkedTransaction.complete();
+
+            try {
+                buildRemoteMsg(idMap, "  1 收到消息");
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+
+
+            // 全新的远程消息
+//            CatCrossProcess.createRemoteMsg("TEST", "LOG-MSG", (idMaps) -> {
+//                try {
+//                    buildRemoteMsg(idMaps, "  4 收到消息");
+//                } catch (Exception e) {
+//                    e.printStackTrace();
+//                }
+//                return null;
+//            });
+
+            return null;
+        });
+
+        System.out.println("处理完成了");
+        Thread.sleep(50000);
+        System.in.read();
+    }
+
+
     @Test
     public void testCreateConsumerCross() throws Exception {
         CatCrossProcess.createRemoteMQMsg("TEST", "LOG-MSG", (idMap) -> {
@@ -56,6 +98,8 @@ public class CatCrossProcessTest extends TestCase {
         new Thread(() -> {
             CatCrossProcess.buildRemoteMQMsg("TEST", "REMOTE-LOG-MSG", idMap, () -> {
                 logger.info(" 远程接收到消息 :" + text + " \t 消息编号 : " + Cat.getCurrentMessageId());
+
+
                 return null;
             });
         }).start();
@@ -65,6 +109,12 @@ public class CatCrossProcessTest extends TestCase {
         new Thread(() -> {
             CatCrossProcess.buildRemoteMsg("TEST", "REMOTE-LOG-MSG", idMap, () -> {
                 logger.info(" 远程接收到消息 :" + text + " \t 消息编号 : " + Cat.getCurrentMessageId());
+                String childLog = idMap.get(Cat.Context.PARENT);
+                TaggedTransaction taggedTransaction = Cat.newTaggedTransaction("remote-test", "remote-name", Cat.getCurrentMessageId());
+                taggedTransaction.bind(childLog, childLog, "MQ远程测试啊");
+                taggedTransaction.setSuccessStatus();
+                taggedTransaction.complete();
+
                 return null;
             });
         }).start();

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

@@ -19,13 +19,14 @@
 
     </appender>
 
-    <root level="WARN">
+    <root level="INFO">
 <!--        <appender-ref ref="cat"/>-->
         <appender-ref ref="stdout"/>
     </root>
 
-    <logger name="com.elab.log.ext" level="INFO" additivity="false">
+    <logger name="com.elab.log" level="INFO" additivity="false">
         <appender-ref ref="cat"/>
+        <appender-ref ref="stdout"/>
     </logger>
 
 </configuration>

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

@@ -55,18 +55,18 @@ public abstract class AbstractMessageListener implements MessageListener {
     public Action consume(Message message, ConsumeContext consumeContext) {
         String topic = message.getTopic();
         String tag = message.getTag();
+        String groupName = this.getClass().getSimpleName();
 
         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 {
-            //boolean isConsumer = checkRepeatMsg(messageModel);
+            // 前置检查
+            boolean isConsumer = this.consumerInterceptor.check(messageModel, groupName);
 
-            boolean isConsumer = consumerInterceptor.check(messageModel);
             //如果已经成功过一次
             if (!isConsumer) {
                 t.setSuccessStatus();
@@ -74,17 +74,19 @@ public abstract class AbstractMessageListener implements MessageListener {
             }
 
             action = consume0(messageModel, consumeContext);
+
             long time = System.currentTimeMillis() - start;
+
             messageModel.setInvokeTime(time);
+
             t.setSuccessStatus();
-            //saveConsumerStatus(messageModel, MqConstants.MSG_OK, time);
-            this.consumerInterceptor.success(messageModel);
+
+            this.consumerInterceptor.success(messageModel, groupName);
         } catch (Exception e) {
             t.setStatus(e);
             logger.error("消息处理异常 : ", e);
             messageModel.setInvokeTime(System.currentTimeMillis() - start);
-            this.consumerInterceptor.error(messageModel, e);
-            //saveConsumerStatus(messageModel, MqConstants.MSG_NO, System.currentTimeMillis() - start);
+            this.consumerInterceptor.error(messageModel, groupName, e);
             return Action.ReconsumeLater;
         } finally {
             t.complete();
@@ -93,5 +95,4 @@ public abstract class AbstractMessageListener implements MessageListener {
         return action;
     }
 
-
 }

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

@@ -13,23 +13,26 @@ public interface ConsumerInterceptor {
     /**
      * 检查数据是否符合
      *
-     * @param messageModel
+     * @param messageModel 消息内容
+     * @param groupName    分组名称 拓展
      * @return
      */
-    public boolean check(MessageModel messageModel);
+    public boolean check(MessageModel messageModel, String groupName);
 
     /**
      * 成功数据回调
      *
      * @param messageModel 消息内容
      */
-    public void success(MessageModel messageModel);
+    public void success(MessageModel messageModel, String groupName);
 
     /**
      * 失败数据回调
      *
      * @param messageModel
+     * @param groupName
+     * @param e
      */
-    public void error(MessageModel messageModel,Throwable e);
+    public void error(MessageModel messageModel, String groupName, Throwable e);
 
 }

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

@@ -16,23 +16,23 @@ public class LoggerConsumerInterceptor implements ConsumerInterceptor, ProducerI
     private Logger logger = LoggerFactory.getLogger(getClass());
 
     @Override
-    public boolean check(MessageModel messageModel) {
+    public boolean check(MessageModel messageModel, String groupName) {
         logger.debug("检测数据 : " + messageModel.getObject(String.class));
         return true;
     }
 
     @Override
-    public void success(MessageModel messageModel, SendResult result) {
+    public void success(MessageModel messageModel, String groupName, SendResult result) {
         logger.debug("发送成功:" + result);
     }
 
     @Override
-    public void error(MessageModel messageModel, Throwable e) {
+    public void error(MessageModel messageModel, String groupName, Throwable e) {
         logger.debug("消费失败:" + messageModel.getMsgID(), e);
     }
 
     @Override
-    public void success(MessageModel messageModel) {
+    public void success(MessageModel messageModel, String groupName) {
         logger.debug("消费:" + messageModel.getMsgID());
     }
 }

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

@@ -4,8 +4,13 @@ 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.elab.log.utils.CatCrossProcess;
+import com.elab.mq.model.MessageModel;
 import com.google.common.base.Splitter;
+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;
@@ -38,8 +43,12 @@ public class MessageListenerWrapper implements MessageListener {
         if (messageListeners != null) {
             try {
                 Map<String, String> msgMap = conversionPropertyToMap(message.getUserProperties());
-
-                result = CatCrossProcess.buildRemoteMsg(defaultRmqGroup, message.getTopic(), msgMap, () -> {
+                // 构建上下文消息
+                result = CatCrossProcess.buildRemoteMQMsg(defaultRmqGroup, message.getTopic(), msgMap, () -> {
+                    MessageModel messageModel = new MessageModel(message);
+                    logger.info("消息体:" + message.toString());
+                    logger.info("消息内容:" + messageModel.getObject(String.class));
+                    long start = System.currentTimeMillis();
                     for (int i = 0; i < messageListeners.size(); i++) {
                         AbstractMessageListener abstractMessageListener = messageListeners.get(i);
                         String currentTopic = message.getTopic();
@@ -50,10 +59,13 @@ public class MessageListenerWrapper implements MessageListener {
                             Action actionResult = abstractMessageListener.consume(message, consumeContext);
                             if (actionResult == null || Action.ReconsumeLater == actionResult) {
                                 logger.warn(" MQ 消费失败 : " + abstractMessageListener);
+                                sendMonitorData(message, msgMap, -1, (System.currentTimeMillis() - start));
                                 return Action.ReconsumeLater;
                             }
                         }
                     }
+                    long time = System.currentTimeMillis() - start;
+                    sendMonitorData(message, msgMap, 1, time);
                     return Action.CommitMessage;
                 });
             } catch (Exception e) {
@@ -87,11 +99,50 @@ public class MessageListenerWrapper implements MessageListener {
         return flag;
     }
 
+    private void sendMonitorData(Message data, Map<String, String> msgMap, Integer status, long time) {
+        MessageModel messageModel = new MessageModel(data);
+        String content = (String) messageModel.getObject(String.class);
+        MQDataDTO mqDataDTO = MonitorUtils.builderConsumerRMQDataDTO(data.getTopic(), data.getKey(), data.getTag(), data.getMsgID(), content, status, time);
+
+        builderRemoteId(messageModel, mqDataDTO);
+
+        MonitorSendProducer.sendMsg(data.getTopic(), mqDataDTO);
+    }
+
+    /**
+     * 构建远程编号
+     *
+     * @param data
+     * @param mqDataDTO
+     */
+    private void builderRemoteId(MessageModel data, MQDataDTO mqDataDTO) {
+        String rootId = data.getUserProperties(Cat.Context.ROOT);
+        String parentId = data.getUserProperties(Cat.Context.PARENT);
+
+        mqDataDTO.setParentLogId(parentId);
+        mqDataDTO.setRootLogId(rootId);
+        mqDataDTO.setLogId(Cat.getCurrentMessageId());
+
+        /////////////////// 默认的关键索引线索 ///////////////////////
+
+        if (mqDataDTO.getGroupKeyName() == null) {
+            mqDataDTO.setGroupKeyName(data.getKey());
+        }
+
+        if (mqDataDTO.getGroupName() == null) {
+            mqDataDTO.setGroupName(data.getModuleName());
+        }
+
+        if (mqDataDTO.getDataId() == null) {
+            mqDataDTO.setDataId(data.getProducerId());
+        }
+
+    }
+
     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));

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

@@ -15,23 +15,25 @@ public interface ProducerInterceptor {
      * 检查数据是否符合
      *
      * @param messageModel
+     * @param groupName
      * @return
      */
-    public boolean check(MessageModel messageModel);
+    public boolean check(MessageModel messageModel,String groupName);
 
     /**
      * 成功数据回调
      *
      * @param messageModel 消息内容
+     * @param groupName
      * @param result
      */
-    public void success(MessageModel messageModel, SendResult result);
+    public void success(MessageModel messageModel,String groupName, SendResult result);
 
     /**
      * 失败数据回调
      *
      * @param messageModel
      */
-    public void error(MessageModel messageModel, Throwable e);
+    public void error(MessageModel messageModel,String groupName, Throwable e);
 
 }

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

@@ -2,8 +2,7 @@ package com.elab.mq.msg.impl;
 
 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.StringUtils;
 import com.elab.log.utils.CatCrossProcess;
 import com.elab.mq.listener.LoggerConsumerInterceptor;
 import com.elab.mq.model.MessageModel;
@@ -16,7 +15,6 @@ import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 
 import java.util.Date;
-import java.util.Map;
 import java.util.Properties;
 
 /**
@@ -45,32 +43,39 @@ public class MsgProducerImpl extends ProducerBean implements IMsgProducerFacade
     public SendResultModel send(MessageModel message) {
         String topic = message.getTopic();
         String tag = message.getTag();
-        Transaction t = Cat.newTransaction(MQ_PRODUCER, topic + "_" + tag + "_" + getClass().getSimpleName());
-        logger.debug(" 发送一条消息 " + message.getMsgID() + " -> " + message.toString());
 
-        // 提前插入消息，为了让消息编号能够带过去，如果发送失败，则根据编号更改为-1状态。
-        // int id = insertRecordProducer(message);
-        try {
+        String key = message.getKey();
+        if (StringUtils.isNotEmpty(key)) {
+            logger.warn("请填相应的key，方便定位该数据");
+        }
 
-            catTraceId(message);
+        CatCrossProcess.createRemoteMQMsg(MQ_PRODUCER, topic + "_" + tag + "_" + getClass().getSimpleName(), (idMap) -> {
+            logger.debug(" 发送一条消息 " + message.getMsgID() + " -> " + message.toString());
 
-            boolean check = producerInterceptor.check(message);
-            if (!check) {
-                return null;
-            }
+            // 提前插入消息，为了让消息编号能够带过去，如果发送失败，则根据编号更改为-1状态。
+            // int id = insertRecordProducer(message);
+            try {
 
-            SendResult result = super.send(message);
-            logger.debug(" 消息发送结果 : " + result.toString());
-            t.setSuccessStatus();
-            producerInterceptor.success(message, result);
-            return new SendResultModel(result);
-        } catch (Exception e) {
-            logger.error("消息发送异常", e);
-            t.setStatus(e);
-            producerInterceptor.error(message, e);
-        } finally {
-            t.complete();
-        }
+                idMap.forEach((K, V) -> {
+                    message.putUserProperties(K, V);
+                });
+
+                boolean check = producerInterceptor.check(message, message.getModuleName());
+
+                if (!check) {
+                    return null;
+                }
+
+                SendResult result = super.send(message);
+                logger.debug(" 消息发送结果 : " + result.toString());
+                producerInterceptor.success(message, message.getModuleName(), result);
+                return new SendResultModel(result);
+            } catch (Exception e) {
+                logger.error("消息发送异常", e);
+                producerInterceptor.error(message, message.getModuleName(), e);
+                throw e;
+            }
+        });
         return null;
     }
 
@@ -100,12 +105,4 @@ 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/README.md

@@ -107,6 +107,18 @@ public void testApp() throws Exception {
 ## 内置自定义注解
 #### CacheLoopSubmit 
 同一时刻，只能有一个相匹配的key被访问！
+注解字段描述:
+- cacheName : 自定义的缓存名字，如果为空默认方法名+参数名
+- unionKey  : 唯一的key,重复提交的关键线索,会根据填写的参数名称去获取对应的值作为缓存关键key,如果匹配则认为重复提交.
+    
+    -[0] : 代表第0个参数
+    
+    -{text}  : 代表实体对象参数的参数名称
+
+> 参考案例 [0]{text} 表示第0个参数的text属性
+- isWaitComplete : 这个代表如果遇到匹配重复提交的情况,是阻塞等待占用线程释放后继续,还是直接跳过.
+- errorMsg : 一旦匹配重复提交的规则,返回的错误信息
+
 ```java
 @CacheLoopSubmit(unionKey = "[0]{text}", cacheName = "demo", timeOut = 10)
 public String submit(SomeObject text) throws Exception {

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

@@ -5,9 +5,9 @@ import com.elab.redis.CacheTemplate;
 import com.elab.redis.annotation.CacheLoopSubmit;
 import com.elab.redis.annotation.CacheReadLock;
 import com.elab.redis.annotation.CacheWriteLock;
-import com.elab.redis.interceptor.BeanFactoryCacheAttributeSourceAdvisor;
-import com.elab.redis.interceptor.CacheAttributeSourcePointcut;
-import com.elab.redis.interceptor.CacheInterceptor;
+import com.elab.redis.interceptor.LockCacheAttributeSourceAdvisor;
+import com.elab.redis.interceptor.LockCacheAttributeSourcePointcut;
+import com.elab.redis.interceptor.LockCacheInterceptor;
 import com.elab.redis.redisson.DefaultRedissonSpringCacheManager;
 import com.elab.redis.spring.data.RedisTemplateDecorator;
 import org.redisson.api.RedissonClient;
@@ -52,36 +52,28 @@ public class CacheAutoConfiguration {
         RedisTemplate<String, Object> template = new RedisTemplateDecorator<>();
         template.setConnectionFactory(redisConnectionFactory);
         template.setKeySerializer(new StringRedisSerializer());
-//        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());
+        template.setHashKeySerializer(new StringRedisSerializer());
+        template.setHashValueSerializer(new StringRedisSerializer());
         return template;
     }
 
-//    private Jackson2JsonRedisSerializer jackson2JsonRedisSerialize(){
-//        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
-//        ObjectMapper om = new ObjectMapper();
-//        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
-//        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
-//        jackson2JsonRedisSerializer.setObjectMapper(om);
-//        return jackson2JsonRedisSerializer;
-//    }
-
     @Bean
     @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
-    public BeanFactoryCacheAttributeSourceAdvisor transactionAdvisor() {
-        BeanFactoryCacheAttributeSourceAdvisor advisor = new BeanFactoryCacheAttributeSourceAdvisor();
+    public LockCacheAttributeSourceAdvisor transactionAdvisor() {
+        LockCacheAttributeSourceAdvisor advisor = new LockCacheAttributeSourceAdvisor();
+        // 具体的拦截器
         advisor.setAdvice(redisCacheInterceptor());
+        // 需要被拦截方法的规则判断器，一旦符合，才会被代理给拦截器处理
         advisor.setPointcut(cacheAttributeSourcePointcut());
         return advisor;
     }
 
     @Bean
     @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
-    public CacheAttributeSourcePointcut cacheAttributeSourcePointcut() {
-        CacheAttributeSourcePointcut cacheAttributeSourcePointcut = new CacheAttributeSourcePointcut();
+    public LockCacheAttributeSourcePointcut cacheAttributeSourcePointcut() {
+        // 该类是用作比对方法的注解是否符合代理的条件
+        LockCacheAttributeSourcePointcut cacheAttributeSourcePointcut = new LockCacheAttributeSourcePointcut();
         cacheAttributeSourcePointcut.addAnnotations(CacheLoopSubmit.class);
         cacheAttributeSourcePointcut.addAnnotations(CacheReadLock.class);
         cacheAttributeSourcePointcut.addAnnotations(CacheWriteLock.class);
@@ -90,8 +82,8 @@ public class CacheAutoConfiguration {
 
     @Bean
     @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
-    public CacheInterceptor redisCacheInterceptor() {
-        CacheInterceptor interceptor = new CacheInterceptor();
+    public LockCacheInterceptor redisCacheInterceptor() {
+        LockCacheInterceptor interceptor = new LockCacheInterceptor();
         return interceptor;
     }
 
@@ -99,7 +91,6 @@ public class CacheAutoConfiguration {
     public CacheManager cacheManager(RedissonClient redissonClient, ElabRedisProperties elabRedisProperties) {
         // 统一的注解缓存失效配置
         DefaultRedissonSpringCacheManager defaultRedissonSpringCacheManager = new DefaultRedissonSpringCacheManager(redissonClient);
-//         defaultRedissonSpringCacheManager.setCacheConfig(elabRedisProperties.getCacheConfig());
 
         return defaultRedissonSpringCacheManager;
     }

elab-redis/src/main/java/com/elab/redis/consts/CacheConstants.java

@@ -11,4 +11,6 @@ public class CacheConstants {
 
     public final static String ALL = "ALL";
 
+    public final static String CACHE_SEPARATOR = ":";
+
 }

elab-redis/src/main/java/com/elab/redis/interceptor/CacheAttributeSourceAdvisor.java

@@ -1,32 +1,32 @@
-package com.elab.redis.interceptor;
-
-import org.aopalliance.aop.Advice;
-import org.springframework.aop.Pointcut;
-import org.springframework.aop.support.AbstractPointcutAdvisor;
-import org.springframework.util.Assert;
-
-/**
- * @author ： liukx
- * @time ： 2020/7/9 - 19:59
- */
-public class CacheAttributeSourceAdvisor extends AbstractPointcutAdvisor {
-
-    private CacheInterceptor cacheInterceptor;
-
-    private CacheAttributeSourcePointcut pointcut;
-
-    public void setCacheInterceptor(CacheInterceptor cacheInterceptor) {
-        this.cacheInterceptor = cacheInterceptor;
-    }
-
-    @Override
-    public Pointcut getPointcut() {
-        return pointcut;
-    }
-
-    @Override
-    public Advice getAdvice() {
-        Assert.state(this.cacheInterceptor != null, "No cacheInterceptor set");
-        return cacheInterceptor;
-    }
-}
+//package com.elab.redis.interceptor;
+//
+//import org.aopalliance.aop.Advice;
+//import org.springframework.aop.Pointcut;
+//import org.springframework.aop.support.AbstractPointcutAdvisor;
+//import org.springframework.util.Assert;
+//
+///**
+// * @author ： liukx
+// * @time ： 2020/7/9 - 19:59
+// */
+//public class CacheAttributeSourceAdvisor extends AbstractPointcutAdvisor {
+//
+//    private CacheInterceptor cacheInterceptor;
+//
+//    private CacheAttributeSourcePointcut pointcut;
+//
+//    public void setCacheInterceptor(CacheInterceptor cacheInterceptor) {
+//        this.cacheInterceptor = cacheInterceptor;
+//    }
+//
+//    @Override
+//    public Pointcut getPointcut() {
+//        return pointcut;
+//    }
+//
+//    @Override
+//    public Advice getAdvice() {
+//        Assert.state(this.cacheInterceptor != null, "No cacheInterceptor set");
+//        return cacheInterceptor;
+//    }
+//}

elab-redis/src/main/java/com/elab/redis/interceptor/BeanFactoryCacheAttributeSourceAdvisor.java

@@ -9,11 +9,11 @@ import org.springframework.aop.support.AbstractBeanFactoryPointcutAdvisor;
  * @author ： liukx
  * @time ： 2020/7/9 - 19:55
  */
-public class BeanFactoryCacheAttributeSourceAdvisor extends AbstractBeanFactoryPointcutAdvisor {
+public class LockCacheAttributeSourceAdvisor extends AbstractBeanFactoryPointcutAdvisor {
 
-    private CacheAttributeSourcePointcut pointcut;
+    private LockCacheAttributeSourcePointcut pointcut;
 
-    public void setPointcut(CacheAttributeSourcePointcut pointcut) {
+    public void setPointcut(LockCacheAttributeSourcePointcut pointcut) {
         this.pointcut = pointcut;
     }
 

elab-redis/src/main/java/com/elab/redis/interceptor/CacheAttributeSourcePointcut.java

@@ -13,17 +13,17 @@ import java.util.Set;
  * @author ： liukx
  * @time ： 2020/7/9 - 20:02
  */
-public class CacheAttributeSourcePointcut extends StaticMethodMatcherPointcut implements Serializable {
+public class LockCacheAttributeSourcePointcut extends StaticMethodMatcherPointcut implements Serializable {
 
-    private Set<Class<? extends Annotation>> CACHE_OPERATION_ANNOTATIONS = new LinkedHashSet<>(8);
+    private Set<Class<? extends Annotation>> cacheOperationAnnotations = new LinkedHashSet<>(8);
 
     public void addAnnotations(Class<? extends Annotation> annotation) {
-        CACHE_OPERATION_ANNOTATIONS.add(annotation);
+        cacheOperationAnnotations.add(annotation);
     }
 
     @Override
     public boolean matches(Method method, Class<?> targetClass) {
-        if (CacheParseUtil.isContainAnnotations(CACHE_OPERATION_ANNOTATIONS, method)) {
+        if (CacheParseUtil.isContainAnnotations(cacheOperationAnnotations, method)) {
             return true;
         }
         return false;

elab-redis/src/main/java/com/elab/redis/interceptor/CacheInterceptor.java

@@ -10,10 +10,11 @@ import java.lang.reflect.Method;
 import java.util.List;
 
 /**
+ * 实现缓存的拦截器
  * @author ： liukx
  * @time ： 2020/7/9 - 20:08
  */
-public class CacheInterceptor implements MethodInterceptor, Serializable {
+public class LockCacheInterceptor implements MethodInterceptor, Serializable {
     @Autowired
     private List<ICacheProcessService> cacheProcessServices;
 

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

@@ -1,34 +0,0 @@
-package com.elab.redis.interceptor.impl;
-
-import org.redisson.api.RLock;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.aop.framework.ReflectiveMethodInvocation;
-
-import java.util.concurrent.TimeUnit;
-
-/**
- * 抽象执行器
- *
- * @author ： liukx
- * @time ： 2020/7/28 - 15:51
- */
-public abstract class AbstractProcess {
-
-    private Logger logger = LoggerFactory.getLogger(AbstractProcess.class);
-
-    protected Object processBussiness(ReflectiveMethodInvocation invocation, int timeOut, RLock lock)
-            throws Throwable {
-        try {
-            if (lock.tryLock(timeOut, TimeUnit.SECONDS)) {
-                Object proceed = invocation.proceed();
-                return proceed;
-            }
-        } catch (InterruptedException e) {
-            logger.warn("尝试获取锁超时", e);
-        } finally {
-            lock.unlock();
-        }
-        return null;
-    }
-}

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

@@ -1,9 +1,13 @@
 package com.elab.redis.interceptor.impl;
 
+import com.elab.core.utils.StringUtils;
 import com.elab.redis.annotation.CacheLoopSubmit;
+import com.elab.redis.config.ElabRedisProperties;
+import com.elab.redis.consts.CacheConstants;
 import com.elab.redis.exceptions.ReSubmitException;
 import com.elab.redis.interceptor.ICacheProcessService;
 import com.elab.redis.utils.CacheParseUtil;
+import com.elab.redis.utils.RedisLockUtils;
 import org.redisson.api.RLock;
 import org.redisson.api.RedissonClient;
 import org.slf4j.Logger;
@@ -25,6 +29,11 @@ import java.util.concurrent.TimeUnit;
 @Component
 public class CacheLoopProcessImpl implements ICacheProcessService {
 
+    private String prefix = "loopSubmit";
+
+    @Autowired
+    private ElabRedisProperties redisProperties;
+
     @Autowired
     private RedissonClient client;
 
@@ -44,11 +53,18 @@ public class CacheLoopProcessImpl implements ICacheProcessService {
         Object[] arguments = invocation.getArguments();
 
         String cacheKey = CacheParseUtil.generateUnionKey(unionKey, arguments);
-        String lockKey = clazzName + "." + method.getName();
 
-        RLock lockObject = client.getLock(lockKey);
-//        RSet<Object> lockObject = client.getSet(lockKey);
-//        RLock lock = lockObject.getLock(cacheKey);
+        String cacheName = cacheLoopSubmit.cacheName();
+
+        if (StringUtils.isEmpty(cacheKey)) {
+            cacheName = getPrefix() + CacheConstants.CACHE_SEPARATOR + prefix + CacheConstants.CACHE_SEPARATOR + clazzName + CacheConstants.CACHE_SEPARATOR + method.getName();
+        } else {
+            cacheName = getPrefix() + CacheConstants.CACHE_SEPARATOR + cacheName;
+        }
+
+        cacheName = cacheName + CacheConstants.CACHE_SEPARATOR + cacheKey;
+
+        RLock lockObject = client.getLock(cacheName);
 
         String errorMsg = cacheLoopSubmit.errorMsg();
 
@@ -62,10 +78,7 @@ public class CacheLoopProcessImpl implements ICacheProcessService {
         } catch (InterruptedException e) {
             logger.warn("尝试获取锁超时", e);
         } finally {
-            // 如果持锁线程是当前线程，则释放锁，非当前线程则不处理
-            if (lockObject.isLocked() && lockObject.isHeldByCurrentThread()) {
-                lockObject.unlock();
-            }
+            RedisLockUtils.releaseLock(lockObject);
         }
         return null;
     }
@@ -79,8 +92,7 @@ public class CacheLoopProcessImpl implements ICacheProcessService {
      */
     private boolean tryLock(RLock lock, CacheLoopSubmit cacheLoopSubmit) {
         try {
-            if (true) {
-//            if (cacheLoopSubmit.isWaitComplete()) {
+            if (cacheLoopSubmit.isWaitComplete()) {
                 return lock.tryLock(cacheLoopSubmit.timeOut(), TimeUnit.MILLISECONDS);
             } else {
                 return lock.tryLock();
@@ -91,5 +103,12 @@ public class CacheLoopProcessImpl implements ICacheProcessService {
         }
     }
 
+    private String getPrefix() {
+        if (redisProperties != null && StringUtils.isNotEmpty(redisProperties.getPrefixWith())) {
+            return redisProperties.getPrefixWith();
+        }
+        return "";
+    }
+
 
 }

elab-redis/src/main/java/com/elab/redis/utils/RedisLockUtils.java

@@ -0,0 +1,73 @@
+package com.elab.redis.utils;
+
+import org.redisson.api.RLock;
+import org.redisson.api.RedissonClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+/**
+ * @Module 工具类
+ * @Description redisLock相关的工具类
+ * @Author liukaixiong
+ * @Date 2021/1/5 13:21
+ */
+public class RedisLockUtils {
+    private static Logger logger = LoggerFactory.getLogger(RedisLockUtils.class);
+
+
+    public static <T> T lock(RedissonClient redissonClient, String key, Supplier<T> supplier) {
+        return lock(redissonClient, key, false, 0L, supplier);
+    }
+
+    /**
+     * 获取分布式锁
+     *
+     * @param redissonClient redisson对象
+     * @param key            锁的key
+     * @param supplier       具体执行的业务逻辑代码
+     * @param isWait         没有抢到锁的情况是否会等待
+     * @param timeOut        等待的超时时间
+     * @param <T>            返回的结果
+     * @return
+     */
+    public static <T> T lock(RedissonClient redissonClient, String key, boolean isWait, Long timeOut, Supplier<T> supplier) {
+        RLock lock = redissonClient.getLock(key);
+        try {
+            if (isTryLock(lock, isWait, timeOut)) {
+                return supplier.get();
+            }
+        } finally {
+            releaseLock(lock);
+        }
+        return null;
+    }
+
+    private static boolean isTryLock(RLock lock, boolean isWait, Long timeOut) {
+        if (isWait) {
+            try {
+                return lock.tryLock(timeOut, TimeUnit.MILLISECONDS);
+            } catch (InterruptedException e) {
+                logger.warn("lock wait time out : " + e.getLocalizedMessage() + " time out : " + timeOut);
+                return false;
+            }
+        } else {
+            return lock.tryLock();
+        }
+    }
+
+
+    /**
+     * 锁释放
+     *
+     * @param lock
+     */
+    public static void releaseLock(RLock lock) {
+        if (lock.isLocked() && lock.isHeldByCurrentThread()) {
+            lock.unlock();
+        }
+    }
+
+}

elab-redis/src/test/java/com/elab/redis/RedisSpringBoot.java

@@ -7,7 +7,6 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
-import org.springframework.cache.annotation.EnableCaching;
 import org.springframework.test.context.junit4.SpringRunner;
 
 /**
@@ -18,7 +17,7 @@ import org.springframework.test.context.junit4.SpringRunner;
 @SpringBootTest(
         classes = {RedissonApplication.class, CacheAutoConfiguration.class}
 )
-@EnableCaching
+//@EnableCaching
 public class RedisSpringBoot {
     @Autowired
     protected RedissonClient client;
@@ -37,7 +36,6 @@ public class RedisSpringBoot {
         for (int i = 0; i < count; i++) {
             new Thread(r).start();
         }
-
     }
 
 }

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

@@ -1,6 +1,7 @@
 package com.elab.redis.cache;
 
 import com.alibaba.fastjson.JSONObject;
+import com.elab.core.componts.ConcurrentTool;
 import com.elab.redis.RedisSpringBoot;
 import com.elab.redis.redisson.doc.model.SomeObject;
 import com.elab.redis.service.IDemoService;
@@ -28,24 +29,29 @@ public class CacheTest extends RedisSpringBoot {
     @Autowired
     private RedissonClient client;
 
+
     @Test
     public void testWatchDog() throws InterruptedException {
         RLock abc = client.getLock("abc");
         // 获取锁
         abc.lock();
-        // 模拟业务超长时间，然后去关注redis的key的过期时间。看是否发生了变化。
+        // 模拟业务超长时间，然后去关注rediseeeeeee的key的过期时间。看是否发生了变化。
         Thread.sleep(1000000);
     }
 
     @Test
     public void testCache() throws Exception {
         logger.info(" 开始进入竞争场景 ...");
-        run(1, () -> {
+
+        ConcurrentTool concurrentTool = new ConcurrentTool(10);
+
+        concurrentTool.process(() -> {
             try {
                 demoService.submit("abc_");
             } catch (Exception e) {
                 e.printStackTrace();
             }
+            return null;
         });
         System.in.read();
     }
@@ -54,6 +60,7 @@ public class CacheTest extends RedisSpringBoot {
     public void testCacheObject() throws Exception {
         SomeObject someObject = new SomeObject();
         someObject.setText("asdfasdf");
+        someObject.setName("lkx");
         demoService.submit(someObject);
     }
 

elab-redis/src/test/java/com/elab/redis/redisson/doc/model/SomeObject.java

@@ -7,11 +7,21 @@ package com.elab.redis.redisson.doc.model;
 public class SomeObject {
 
 
+    private String name;
+
     private String text;
 
     public SomeObject() {
     }
 
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
     public SomeObject(String text) {
         this.text = text;
     }

elab-redis/src/test/java/com/elab/redis/service/impl/DemoServiceImpl.java

@@ -36,7 +36,7 @@ public class DemoServiceImpl implements IDemoService {
     }
 
     @Override
-    @CacheLoopSubmit(unionKey = "[0]{text}", cacheName = "demo", timeOut = 10)
+    @CacheLoopSubmit(unionKey = {"[0]{text}", "[0]{name}"}, cacheName = "demo", timeOut = 10)
     public String submit(SomeObject text) throws Exception {
         count--;
         Thread.sleep(500);

elab-redis/src/test/java/com/elab/redis/spring/SpringDataTest.java

@@ -1,16 +1,16 @@
 package com.elab.redis.spring;
 
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
 import com.elab.redis.CacheTemplate;
 import com.elab.redis.RedisSpringBoot;
 import org.junit.Test;
-import org.redisson.api.RLock;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.redis.core.StringRedisTemplate;
 
 import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
+import java.util.List;
 
 /**
  * springData
@@ -32,22 +32,50 @@ public class SpringDataTest extends RedisSpringBoot {
     @Test
     public void set() throws IOException {
 
-        cacheTemplate.set().add("aaa","a","b","c");
+//        cacheTemplate.set().add("aaa","a","b","c");
+//
+//
+//        RLock xxx = cacheTemplate.getRedissonClient().getLock("xxx");
+//        cacheTemplate.string().set("abc", "cccc");
+//        Object abc = cacheTemplate.string().get("abc");
+//        System.out.println("--->" + abc);
+//
+//        redisTemplate.opsForValue().set("kkx", "abc");
+//
+//        Map<String, String> map = new HashMap<>();
+//        map.put("lkx", "xxx");
+//        map.put("xxx", "sss");
+//        redisTemplate.opsForValue().set("obj", map);
+//        Map<String, String> xiong = (Map<String, String>) redisTemplate.opsForValue().get("obj");
+//        System.out.println(xiong);
 
+//        // 存储值
+//        cacheTemplate.map().put("aaa", "a-key", "v-value");
+//        cacheTemplate.map().put("aaa", "b-key", "b-value");
+//        cacheTemplate.map().put("aaa", "c-key", "c-value");
+//        cacheTemplate.map().put("aaa", "d-key", "d-value");
+//
+//        // 获取key的hashkey的值
+//        String val = (String) cacheTemplate.map().get("aaa", "d-key");
+//        String val = (String) cacheTemplate.map().get("aaa", "fff");
 
-        RLock xxx = cacheTemplate.getRedissonClient().getLock("xxx");
-        cacheTemplate.string().set("abc", "cccc");
-        Object abc = cacheTemplate.string().get("abc");
-        System.out.println("--->" + abc);
+        String key = "data_center:material_labels_key:1";
+        Object value = cacheTemplate.string().get(key);
 
-        redisTemplate.opsForValue().set("kkx", "abc");
+        List<JSONObject> jsonObjects = JSON.parseArray(value.toString(), JSONObject.class);
+        jsonObjects.forEach((str) -> {
+            String materialsId = str.getString("materialsId");
+            cacheTemplate.map().put("data_center:material_labels_key:1111", materialsId, str.toJSONString());
+        });
 
-        Map<String, String> map = new HashMap<>();
-        map.put("lkx", "xxx");
-        map.put("xxx", "sss");
-        redisTemplate.opsForValue().set("obj", map);
-        Map<String, String> xiong = (Map<String, String>) redisTemplate.opsForValue().get("obj");
-        System.out.println(xiong);
+        System.out.println(jsonObjects.toArray());
+
+        // 获取hash的key的所有值
+//        Map<String, Object> dataMap = cacheTemplate.map().entries("data_center:material_labels_key:1");
+//        System.out.println(dataMap.toString());
+//        System.out.println(val);
+
+        System.out.println("OK");
         System.in.read();
 
     }

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

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

elab-rocketMQ/src/main/java/com/elab/mq/rocket/msg/MemoryDataStore.java

@@ -18,7 +18,7 @@ import java.util.concurrent.TimeUnit;
  */
 public class MemoryDataStore implements ConsumerDataStore {
     private Map<String, RmqConsumerEntity> cacheMap = Maps.newConcurrentMap();
-    private LoadingCache<String, RmqConsumerEntity> cacheBuilder = CacheBuilder.newBuilder().expireAfterAccess(5l,
+    private LoadingCache<String, RmqConsumerEntity> cacheBuilder = CacheBuilder.newBuilder().expireAfterAccess(5L,
             TimeUnit.SECONDS)
             .build(new CacheLoader<String, RmqConsumerEntity>() {
                 @Override

elab-spring/README.md

@@ -39,7 +39,18 @@ httpClient.read.timeOut=120000
 # 读取property配置文件的路径
 spring.resources.path=classpath:*.properties
 ```
-
+### swagger 新增扫描多个包的情况
+`swagger2.basePackage` : 用以","号分割
+```yaml
+swagger2: 
+  basePackage: com.elab.marketing.brand.service.impl,com.elab.marketing.brand.controllers
+  title: brand
+  description: 品牌app微服务
+  termsOfServiceUrl: http://www.elab-plus.com
+  licenseUrl: http://127.0.0.1:5312/doc.html
+  version: 1.0.0
+  enable: true
+```
 
 ### DataSourceConfigBean : 数据源配置
 

elab-spring/pom.xml

@@ -25,11 +25,11 @@
     <!--<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>
@@ -61,13 +61,14 @@
             <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>-->
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+            <version>2.9.8</version>
+            <scope>provided</scope>
         </dependency>
 
         <dependency>
@@ -144,12 +145,12 @@
             <groupId>io.springfox</groupId>
             <artifactId>springfox-swagger2</artifactId>
             <scope>compile</scope>
-            <exclusions>
-                <exclusion>
-                    <artifactId>swagger-models</artifactId>
-                    <groupId>io.swagger</groupId>
-                </exclusion>
-            </exclusions>
+            <!--            <exclusions>-->
+            <!--                <exclusion>-->
+            <!--                    <artifactId>swagger-models</artifactId>-->
+            <!--                    <groupId>io.swagger</groupId>-->
+            <!--                </exclusion>-->
+            <!--            </exclusions>-->
         </dependency>
 
         <dependency>
@@ -167,12 +168,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>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/anno/EnableElabSpring.java

@@ -1,6 +1,7 @@
 package com.elab.spring.anno;
 
 import com.elab.log.asepct.LogResponseBodyAdvice;
+import com.elab.spring.config.RmqConfiguration;
 import com.elab.spring.config.SpringMvcConfig;
 import com.elab.spring.config.SwaggerConfigBean;
 import com.elab.spring.config.ThreadConfiguration;
@@ -19,7 +20,7 @@ import java.lang.annotation.*;
 @Target(ElementType.TYPE)
 @Documented
 //, SpringCommonConfig.class 有问题
-@Import({ThreadConfiguration.class, LogResponseBodyAdvice.class, CommonException.class, SwaggerConfigBean.class, SpringMvcConfig.class})
+@Import({RmqConfiguration.class,ThreadConfiguration.class, LogResponseBodyAdvice.class, CommonException.class, SwaggerConfigBean.class, SpringMvcConfig.class})
 public @interface EnableElabSpring {
 
 }

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

@@ -1,57 +1,70 @@
 package com.elab.spring.config;
 
 import com.elab.core.aop.annotations.EnableElabDB;
-import com.elab.mq.listener.AbstractMessageListener;
+import com.elab.mq.config.RocketMQConfiguration;
 import com.elab.mq.listener.ConsumerInterceptor;
-import com.elab.mq.msg.IMsgProducerFacade;
 import com.elab.mq.msg.ProducerInterceptor;
+import com.elab.redis.CacheTemplate;
 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.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
 
 /**
- * @Module TODO
- * @Description TODO
+ * @Module 配置类
+ * @Description 幂等配置
  * @Author liukaixiong
  * @Date 2020/12/30 15:15
  */
 @Configuration
+@ConditionalOnClass(value = {RocketMQConfiguration.class})
 @EnableElabDB(basePackageClasses = {IConsumerDao.class, IProducerDao.class})
+@Import(value = {RocketMQConfiguration.class})
 public class RmqConfiguration {
 
-    @Bean
-    @ConditionalOnMissingBean(value = {ConsumerInterceptor.class})
-//    @ConditionalOnBean(value = {CacheTemplate.class, AbstractMessageListener.class})
-    public ConsumerInterceptor cacheConsumerInterceptor() {
-        return new RMQCacheConsumerInterceptor();
-    }
+    @Configuration
+    @ConditionalOnClass(value = CacheTemplate.class)
+    @ConditionalOnProperty(prefix = "elab.mq", value = "store", havingValue = "cache", matchIfMissing = true)
+    protected static class CacheRocketMqConfiguration {
 
-    @Bean
-    @ConditionalOnMissingBean(value = {ProducerInterceptor.class})
-//    @ConditionalOnBean(value = {CacheTemplate.class, IMsgProducerFacade.class})
-    public ProducerInterceptor cacProducerInterceptor() {
-        return new RMQCacheProducerInterceptor();
-    }
+        @Bean
+        @ConditionalOnMissingBean(value = {ConsumerInterceptor.class})
+        public ConsumerInterceptor cacheConsumerInterceptor() {
+            return new RMQCacheConsumerInterceptor();
+        }
 
-    @Bean
-    @ConditionalOnMissingBean(value = {ConsumerInterceptor.class})
-    @ConditionalOnBean(value = {AbstractMessageListener.class})
-    public ConsumerInterceptor msgConsumerInterceptor() {
-        return new RMQMsgTableConsumerInterceptor();
+        @Bean
+        @ConditionalOnMissingBean(value = {ProducerInterceptor.class})
+        public ProducerInterceptor cacheProducerInterceptor() {
+            return new RMQCacheProducerInterceptor();
+        }
     }
 
-    @Bean
-    @ConditionalOnMissingBean(value = {ProducerInterceptor.class})
-    @ConditionalOnBean(value = {IMsgProducerFacade.class})
-    public ProducerInterceptor msgProducerInterceptor() {
-        return new RMQMsgTableProducerInterceptor();
+    @Configuration
+    @ConditionalOnMissingBean(value = CacheTemplate.class)
+    @ConditionalOnProperty(prefix = "elab.mq", value = "store", havingValue = "mysql", matchIfMissing = false)
+    protected static class MysqlRocketMqConfiguration {
+
+        @Bean
+        @ConditionalOnMissingBean(value = {ConsumerInterceptor.class})
+        public ConsumerInterceptor mysqlConsumerInterceptor() {
+            return new RMQMsgTableConsumerInterceptor();
+        }
+
+        @Bean
+        @ConditionalOnMissingBean(value = {ProducerInterceptor.class})
+        public ProducerInterceptor mysqlProducerInterceptor() {
+            return new RMQMsgTableProducerInterceptor();
+        }
     }
 
+
 }

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

@@ -1,10 +1,12 @@
 package com.elab.spring.config;
 
 import com.elab.spring.utils.RestTemplateUtils;
+import com.elab.spring.utils.SpringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.config.PropertyPlaceholderConfigurer;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
 import org.springframework.core.env.Environment;
 import org.springframework.core.io.Resource;
 import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
@@ -22,6 +24,7 @@ import java.io.IOException;
  * @email : liukx@elab-plus.com
  */
 @Configuration
+@Import(value = {SpringUtils.class})
 public class SpringCommonConfig {
 
     @Bean

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

@@ -1,18 +1,25 @@
 package com.elab.spring.config;
 
+import com.google.common.base.Function;
+import com.google.common.base.Optional;
+import com.google.common.base.Predicate;
+import com.google.common.base.Splitter;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.core.annotation.Order;
 import org.springframework.core.env.Environment;
+import org.springframework.util.ClassUtils;
+import springfox.documentation.RequestHandler;
 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;
 import springfox.documentation.swagger2.annotations.EnableSwagger2;
 
+import java.util.List;
+
 /**
  * Swagger配置
  *
@@ -28,13 +35,15 @@ public class SwaggerConfigBean {
     @Autowired
     private Environment environment;
 
+    private static String defaultDelimiter = ",";
+
     @Bean
     public Docket createRestApi() {
         return new Docket(DocumentationType.SWAGGER_2)
                 .apiInfo(apiInfo())
                 .select()
-                // 具体需要扫描的包
-                .apis(RequestHandlerSelectors.basePackage(environment.getProperty("swagger2.basePackage")))
+                // 具体需要扫描的包 RequestHandlerSelectors
+                .apis(baseMultiPackage(environment.getProperty("swagger2.basePackage")))
                 .paths(PathSelectors.any())
                 .build();
     }
@@ -48,4 +57,47 @@ public class SwaggerConfigBean {
                 .licenseUrl(environment.getProperty("swagger2.version"))
                 .build();
     }
+
+    /**
+     * 处理多个包的扫描情况
+     *
+     * @param basePackage
+     * @return
+     */
+    public static Predicate<RequestHandler> baseMultiPackage(final String basePackage) {
+        return new Predicate<RequestHandler>() {
+            @Override
+            public boolean apply(RequestHandler input) {
+                return Optional.fromNullable(input.declaringClass()).transform(handlerMultiPackage(basePackage)).or(true);
+            }
+        };
+    }
+
+    /**
+     * 匹配多个
+     *
+     * @param basePackage
+     * @return
+     */
+    private static Function<Class<?>, Boolean> handlerMultiPackage(final String basePackage) {
+        return new Function<Class<?>, Boolean>() {
+            @Override
+            public Boolean apply(Class<?> input) {
+
+                if (basePackage.indexOf(defaultDelimiter) > -1) {
+                    String packageName = ClassUtils.getPackageName(input);
+                    List<String> packageList = Splitter.on(defaultDelimiter).splitToList(basePackage);
+                    for (int i = 0; i < packageList.size(); i++) {
+                        String packages = packageList.get(i);
+                        if (packageName.startsWith(packages)) {
+                            return true;
+                        }
+                    }
+                    return false;
+                } else {
+                    return ClassUtils.getPackageName(input).startsWith(basePackage);
+                }
+            }
+        };
+    }
 }

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

@@ -29,13 +29,15 @@ public class CheckAnnotationProcess implements BeanPostProcessor, BeanFactoryPos
 
     private Environment environment;
     private boolean isException = false;
-    /*排除不包含的包名*/
+    /**
+     * 排除不需要扫描验证的包名
+     */
     private String excludePackage;
 
     public CheckAnnotationProcess(Environment environment) {
         this.environment = environment;
-        this.isException = ObjectUtils.isNotEmpty(environment.getProperty("elab.isException")) ? Boolean.parseBoolean(environment.getProperty("elab.isException")): false;
-        this.excludePackage = environment.getProperty("elab.excludePackage");
+        this.isException = ObjectUtils.isNotEmpty(environment.getProperty("elab.isException")) ? Boolean.parseBoolean(environment.getProperty("elab.isException")) : false;
+        this.excludePackage = environment.getProperty("elab.excludePackage", "");
     }
 
     public void setException(boolean exception) {
@@ -55,45 +57,53 @@ public class CheckAnnotationProcess implements BeanPostProcessor, BeanFactoryPos
     @Override
     public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
         String[] checkPackageList = excludePackage.split(",");
-        if (checkPackageList.length > 0) {
-            String exclude = "com.elab.marketing.auth.dao";
+        if (checkPackageList != null && checkPackageList.length > 0) {
             String proxyName = "com.sun.proxy";
             if (AopUtils.isAopProxy(bean)) {
                 Class<?> targetClass = AopUtils.getTargetClass(bean);
                 String name = targetClass.getName();
 
-
-                for(String singlePackage : checkPackageList) {
+                for (String singlePackage : checkPackageList) {
                     if (!name.startsWith(singlePackage) && !name.startsWith(proxyName)) {
-                        Method[] declaredMethods = targetClass.getDeclaredMethods();
-                        for (int i = 0; i < declaredMethods.length; i++) {
-                            Method declaredMethod = declaredMethods[i];
-                            String author = "";
-                            Transactional annotation = declaredMethod.getAnnotation(Transactional.class);
-                            ApiOperation apiOperation = declaredMethod.getAnnotation(ApiOperation.class);
-                            if (apiOperation != null && StringUtils.isNotEmpty(apiOperation.nickname())) {
-                                author = apiOperation.nickname();
-                            }
-                            if (annotation != null || apiOperation != null) {
-                                ExceptionHandle exceptionHandle = declaredMethod.getAnnotation(ExceptionHandle.class);
-                                if (exceptionHandle != null) {
-                                    if (StringUtils.isEmpty(exceptionHandle.username())) {
-                                        String message = "请在" + name + " \t " + declaredMethod.getName() + " --> " + author + " 方法头的@ExceptionHandle 加上username属性";
-                                        notifyWarning(message);
-                                    }
-                                } else {
-                                    String message = "请在" + name + " \t " + declaredMethod.getName() + " --> " + author + " 方法头加上 @ExceptionHandle(username = '')";
-                                    notifyWarning(message);
-                                }
-                            }
-                        }
+                        checkAnnotationValid(targetClass, name);
                     }
                 }
             }
+        } else {
+            if (AopUtils.isAopProxy(bean)) {
+                Class<?> targetClass = AopUtils.getTargetClass(bean);
+                String name = targetClass.getName();
+                checkAnnotationValid(targetClass, name);
+            }
         }
         return bean;
     }
 
+    private void checkAnnotationValid(Class<?> targetClass, String name) {
+        Method[] declaredMethods = targetClass.getDeclaredMethods();
+        for (int i = 0; i < declaredMethods.length; i++) {
+            Method declaredMethod = declaredMethods[i];
+            String author = "";
+            Transactional annotation = declaredMethod.getAnnotation(Transactional.class);
+            ApiOperation apiOperation = declaredMethod.getAnnotation(ApiOperation.class);
+            if (apiOperation != null && StringUtils.isNotEmpty(apiOperation.nickname())) {
+                author = apiOperation.nickname();
+            }
+            if (annotation != null || apiOperation != null) {
+                ExceptionHandle exceptionHandle = declaredMethod.getAnnotation(ExceptionHandle.class);
+                if (exceptionHandle != null) {
+                    if (StringUtils.isEmpty(exceptionHandle.username())) {
+                        String message = "请在" + name + " \t " + declaredMethod.getName() + " --> " + author + " 方法头的@ExceptionHandle 加上username属性";
+                        notifyWarning(message);
+                    }
+                } else {
+                    String message = "请在" + name + " \t " + declaredMethod.getName() + " --> " + author + " 方法头加上 @ExceptionHandle(username = '')";
+                    notifyWarning(message);
+                }
+            }
+        }
+    }
+
     private void notifyWarning(String message) {
         if (isException) {
             throw new CheckException(message);

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

@@ -4,9 +4,6 @@ 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;
@@ -39,14 +36,16 @@ public class RMQCacheConsumerInterceptor implements ConsumerInterceptor {
     /**
      * 消费中的时间，避免服务挂掉了，负载均衡到另一台
      */
-    private Long maxWaitTime = TimeUnit.SECONDS.toMillis(100);
+    private Long maxWaitTime = TimeUnit.SECONDS.toMillis(20);
+
+    private Long retryWaitTime = TimeUnit.DAYS.toMillis(1);
     /**
      * 消费完成之后，多久的过期时间
      */
     private Long successValidTime = TimeUnit.MINUTES.toMillis(30);
 
     @Override
-    public boolean check(MessageModel messageModel) {
+    public boolean check(MessageModel messageModel, String groupName) {
         /**
          * 0: 消费中
          * 1: 消费成功
@@ -54,8 +53,8 @@ public class RMQCacheConsumerInterceptor implements ConsumerInterceptor {
          *
          * 考虑到的情况:
          * 1. 消费慢
-         * - 过了超时时间，但是没有超过RMQ的ConsumeTimeout确认超时时间。没关系，等提交确认状态就好了,因为同一条消息只会被同一个组的一个线程消费.
-         * - 过了超时时间，也超过了RMQ的ConsumeTimeout确认超时时间。
+         * - 过了key超时时间，但是没有超过RMQ的ConsumeTimeout确认超时时间。没关系，等提交确认状态就好了,因为同一条消息只会被同一个组的一个线程消费.
+         * - 过了key超时时间，也超过了RMQ的ConsumeTimeout确认超时时间。
          * --- RMQ会执行重试，这个时候就依赖业务这种需求做幂等。(比较极端)
          * 2. 服务宕机: 这个时候肯定不会提交ACK状态,直接走重试逻辑就好.
          * - 2.1 -> 等待key超时就行了
@@ -65,7 +64,7 @@ public class RMQCacheConsumerInterceptor implements ConsumerInterceptor {
          * ---
          *
          */
-        String key = getCacheKey(messageModel);
+        String key = getCacheKey(messageModel, groupName);
         Object statusObject = redisTemplate.opsForValue().get(key);
         if (statusObject == null) {
             // 如果maxWaitTime还没消费完，只有两种情况：1.服务挂掉了， 2. 消费慢
@@ -75,7 +74,7 @@ public class RMQCacheConsumerInterceptor implements ConsumerInterceptor {
              * - 2.1 : RMQ认为消费超时了,RMQ的消费确认时间肯定大于这个maxWaitTime时间就行.
              * - 2.2 : 确实消费慢,那key超时了没关系,只要maxWaitTime小于RMQ超时确认时间
              */
-            redisTemplate.opsForValue().set(key, MqConstants.MSG_WAIT, maxWaitTime);
+            redisTemplate.opsForValue().set(key, MqConstants.MSG_WAIT, maxWaitTime, TimeUnit.MILLISECONDS);
             logger.debug("缓存幂等校验通过:未发现值:" + key);
             return true;
         } else {
@@ -84,58 +83,57 @@ public class RMQCacheConsumerInterceptor implements ConsumerInterceptor {
                 /**
                  * 到达这里的话，只有可能是出现超时重试的情况。
                  * 实际消费成功了，但是没有确认给RMQ。
+                 *
+                 * 一旦出现重试，则说明出现了故障问题，则将有效时间进行延长
                  */
                 logger.info("该数据已经被消费过了 : " + key);
+                redisTemplate.opsForValue().set(key, MqConstants.MSG_OK, retryWaitTime, TimeUnit.MILLISECONDS);
                 return false;
             } else if (MqConstants.MSG_WAIT.equals(status)) {
-                // 几乎不可能触发.
+                // 几乎不可能触发.因为同一时刻只可能有一个消费者消费，不会出现并发情况。
                 throw new RuntimeException("等待中未消费完成的数据");
             } else {
                 Long count = redisTemplate.opsForValue().increment(key, 1);
-                logger.debug("缓存幂等校验通过 -> 重试 : " + count);
+                Boolean expire = redisTemplate.expire(key, retryWaitTime, TimeUnit.MILLISECONDS);
+                logger.debug("缓存幂等校验通过 -> 重试 " + expire + ": " + count);
                 return true;
             }
         }
     }
 
-    private String getCacheKey(MessageModel messageModel) {
+    /**
+     * 目前是根据前缀+组id+redis的自增键去实现
+     * 如果有问题可以使用RMQ里面的MSGID去做唯一键.
+     *
+     * @param messageModel
+     * @param groupName
+     * @return
+     */
+    private String getCacheKey(MessageModel messageModel, String groupName) {
         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(groupName).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());
+    public void success(MessageModel messageModel, String groupName) {
+        String key = getCacheKey(messageModel, groupName);
+        redisTemplate.opsForValue().set(key, MqConstants.MSG_OK, maxWaitTime, TimeUnit.MILLISECONDS);
     }
 
     @Override
-    public void error(MessageModel messageModel, Throwable e) {
-        String key = getCacheKey(messageModel);
+    public void error(MessageModel messageModel, String groupName, Throwable e) {
+        String key = getCacheKey(messageModel, groupName);
         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

@@ -1,12 +1,15 @@
 package com.elab.spring.intercept;
 
 import com.aliyun.openservices.ons.api.SendResult;
+import com.dianping.cat.Cat;
 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 
 /**
@@ -22,11 +25,13 @@ public class RMQCacheProducerInterceptor implements ProducerInterceptor {
      */
     private String defaultPrefix = "RMQProducer:index";
 
+    private Logger logger = LoggerFactory.getLogger(getClass());
+
     @Autowired
     private CacheTemplate cacheTemplate;
 
     @Override
-    public boolean check(MessageModel messageModel) {
+    public boolean check(MessageModel messageModel, String groupName) {
         // 由于多个客户端发送，如果区分开的话就得维护多个
         String cacheKey = getCacheKey();
         Long producerId = cacheTemplate.string().increment(cacheKey, 1);
@@ -35,27 +40,57 @@ public class RMQCacheProducerInterceptor implements ProducerInterceptor {
     }
 
     @Override
-    public void success(MessageModel messageModel, SendResult result) {
-        sendMonitorData(messageModel, 1);
+    public void success(MessageModel messageModel, String groupName, SendResult result) {
+         sendMonitorData(messageModel, 1);
+        logger.debug("业务处理成功,携带的消息唯一编号:" + messageModel.getProducerId());
     }
 
     @Override
-    public void error(MessageModel messageModel, Throwable e) {
-        sendMonitorData(messageModel, -1);
+    public void error(MessageModel messageModel, String groupName, Throwable e) {
+         sendMonitorData(messageModel, -1);
+        logger.debug("业务处理失败,携带的消息唯一编号:" + messageModel.getProducerId());
     }
 
     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);
+
+        builderRemoteId(data, mqDataDTO);
+
+        MonitorSendProducer.sendMsg(data.getTopic(), mqDataDTO);
     }
 
+    /**
+     * 构建远程编号,留下相应线索
+     *
+     * @param data
+     * @param mqDataDTO
+     */
+    private void builderRemoteId(MessageModel data, MQDataDTO mqDataDTO) {
+        ///////////////// 上下文链路编号 ////////////////////
+        String rootId = data.getUserProperties(Cat.Context.ROOT);
+        String parentId = data.getUserProperties(Cat.Context.PARENT);
+
+        mqDataDTO.setParentLogId(parentId);
+        mqDataDTO.setRootLogId(rootId);
+        mqDataDTO.setLogId(parentId);
+
+        /////////////////// 默认的关键索引线索 ///////////////////////
+        if (mqDataDTO.getGroupKeyName() == null) {
+            mqDataDTO.setGroupKeyName(data.getKey());
+        }
+
+        if (mqDataDTO.getGroupName() == null) {
+            mqDataDTO.setGroupName(data.getModuleName());
+        }
+
+        if (mqDataDTO.getDataId() == null) {
+            mqDataDTO.setDataId(data.getProducerId());
+        }
+
+    }
 
     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

@@ -8,9 +8,6 @@ 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;
@@ -39,17 +36,17 @@ public class RMQMsgTableConsumerInterceptor implements ConsumerInterceptor {
     private IConsumerDao consumerDao;
 
     @Override
-    public boolean check(MessageModel messageModel) {
+    public boolean check(MessageModel messageModel,String groupName) {
         return checkRepeatMsg(messageModel);
     }
 
     @Override
-    public void success(MessageModel messageModel) {
+    public void success(MessageModel messageModel,String groupName) {
         saveConsumerStatus(messageModel, MqConstants.MSG_OK, messageModel.getInvokeTime());
     }
 
     @Override
-    public void error(MessageModel messageModel, Throwable e) {
+    public void error(MessageModel messageModel,String groupName, Throwable e) {
         saveConsumerStatus(messageModel, MqConstants.MSG_NO, messageModel.getInvokeTime());
     }
 
@@ -118,21 +115,6 @@ public class RMQMsgTableConsumerInterceptor implements ConsumerInterceptor {
         } 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

@@ -38,19 +38,19 @@ public class RMQMsgTableProducerInterceptor implements ProducerInterceptor {
     private String groupId;
 
     @Override
-    public boolean check(MessageModel messageModel) {
+    public boolean check(MessageModel messageModel,String groupName) {
         int id = insertRecordProducer(messageModel);
         messageModel.setProducerId(id + "");
         return true;
     }
 
     @Override
-    public void success(MessageModel messageModel, SendResult result) {
+    public void success(MessageModel messageModel,String groupName, SendResult result) {
         sendMonitorData(messageModel, 1);
     }
 
     @Override
-    public void error(MessageModel messageModel, Throwable e) {
+    public void error(MessageModel messageModel,String groupName, Throwable e) {
         updateRecordProducer(Integer.valueOf(messageModel.getProducerId()), -1, e.getMessage());
         sendMonitorData(messageModel, -1);
     }
@@ -116,7 +116,7 @@ public class RMQMsgTableProducerInterceptor implements ProducerInterceptor {
     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);
+        MonitorSendProducer.sendMsg(data.getTopic(), mqDataDTO);
     }
 
 }

elab-spring/src/main/java/com/elab/spring/utils/SpringUtils.java

@@ -8,6 +8,9 @@ import org.springframework.stereotype.Component;
 
 import java.util.Map;
 
+/**
+ *
+ */
 @Component
 public class SpringUtils implements ApplicationContextAware {
 

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

@@ -20,13 +20,14 @@ public class SwaggerConfigBeanTest extends TestCase {
 
     @Test
     public void test1() {
-        new Docket(DocumentationType.SWAGGER_2)
+        Docket build = new Docket(DocumentationType.SWAGGER_2)
                 .apiInfo(apiInfo())
                 .select()
                 // 具体需要扫描的包
                 .apis(RequestHandlerSelectors.basePackage("com.elab.spring.config"))
                 .paths(PathSelectors.any())
                 .build();
+
     }
 
     private ApiInfo apiInfo() {
@@ -34,4 +35,5 @@ public class SwaggerConfigBeanTest extends TestCase {
                 .title("aaaa")
                 .build();
     }
+
 }

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

@@ -41,7 +41,7 @@ public class RMQCacheConsumerInterceptorTest {
         jsonObject.put("key", "test");
         MessageModel messageModel = new MessageModel("A", "B", "C", jsonObject.toString());
         messageModel.setInvokeTime(1000L);
-        producerInterceptor.check(messageModel);
+        producerInterceptor.check(messageModel,"");
         return messageModel;
     }
 
@@ -53,9 +53,9 @@ public class RMQCacheConsumerInterceptorTest {
     private void successProcess() {
         MessageModel messageModel = sendMsg();
         messageModel.setInvokeTime(1000L);
-        boolean check = rmqCacheConsumerInterceptor.check(messageModel);
+        boolean check = rmqCacheConsumerInterceptor.check(messageModel,"");
         if (check) {
-            rmqCacheConsumerInterceptor.success(messageModel);
+            rmqCacheConsumerInterceptor.success(messageModel,"");
         }
     }
 
@@ -74,9 +74,9 @@ public class RMQCacheConsumerInterceptorTest {
         if (fixedProducerId != null) {
             messageModel.setProducerId(fixedProducerId + "");
         }
-        boolean check = rmqCacheConsumerInterceptor.check(messageModel);
+        boolean check = rmqCacheConsumerInterceptor.check(messageModel,"");
         if (check) {
-            rmqCacheConsumerInterceptor.error(messageModel, null);
+            rmqCacheConsumerInterceptor.error(messageModel, "",null);
         }
     }
 

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

@@ -2,6 +2,7 @@ package com.elab.spring.intercept;
 
 import com.alibaba.fastjson.JSONObject;
 import com.elab.mq.model.MessageModel;
+import com.elab.mq.msg.ProducerInterceptor;
 import com.elab.redis.annotation.EnableElabRedis;
 import com.elab.spring.config.RmqConfiguration;
 import junit.framework.TestCase;
@@ -27,7 +28,7 @@ import org.springframework.test.context.junit4.SpringRunner;
 public class RMQCacheProducerInterceptorTest extends TestCase {
 
     @Autowired
-    private RMQCacheProducerInterceptor producerInterceptor;
+    private ProducerInterceptor producerInterceptor;
 
     @Test
     public void check() {
@@ -35,7 +36,7 @@ public class RMQCacheProducerInterceptorTest extends TestCase {
         jsonObject.put("key", "test");
         MessageModel messageModel = new MessageModel("A", "B", "C", jsonObject.toString());
         messageModel.setInvokeTime(1000L);
-        boolean check = producerInterceptor.check(messageModel);
+        boolean check = producerInterceptor.check(messageModel, "");
         System.out.println(check);
     }
 

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

@@ -0,0 +1,53 @@
+package com.elab.spring.intercept;
+
+import com.elab.core.aop.annotations.EnableElabDB;
+import com.elab.mq.config.RocketMQConfiguration;
+import com.elab.mq.model.MessageModel;
+import com.elab.mq.msg.ProducerInterceptor;
+import com.elab.spring.config.RmqConfiguration;
+import com.elab.spring.dao.IConsumerDao;
+import com.elab.spring.dao.IProducerDao;
+import junit.framework.TestCase;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+
+/**
+ * @Module rmq模块测试
+ * @Description 测试mysql拦截
+ * @Author liukaixiong
+ * @Date 2021/1/11 16:33
+ */
+@RunWith(SpringRunner.class)
+@SpringBootTest(classes = {RocketMQConfiguration.class, RmqConfiguration.class})
+@EnableElabDB(basePackageClasses = {IConsumerDao.class, IProducerDao.class})
+public class RMQMsgTableProducerInterceptorTest extends TestCase {
+
+
+    @Autowired
+    private ProducerInterceptor producerInterceptor;
+
+    @Test
+    public void check() {
+        sendMsg();
+    }
+
+    @Test
+    public void success() {
+    }
+
+    @Test
+    public void error() {
+    }
+
+    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;
+    }
+}

elab-spring/src/test/java/com/elab/spring/utils/ThreadProcessUtilsTest.java

@@ -44,4 +44,5 @@ public class ThreadProcessUtilsTest extends TestCase {
         });
         System.in.read();
     }
+
 }