新增消息队列功能

elab-core/pom.xml

@@ -46,25 +46,25 @@
             </plugin>
 
             <!-- 打包javadoc插件 -->
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-javadoc-plugin</artifactId>
-                <!--<version>2.9</version>-->
-                <executions>
-                    <execution>
-                        <id>attach-javadocs</id>
-                        <goals>
-                            <goal>jar</goal>
-                        </goals>
-                        <configuration>
-                            <additionalparam>-Xdoclint:none</additionalparam>
-                            <charset>UTF-8</charset>
-                            <encoding>UTF-8</encoding>
-                            <docencoding>UTF-8</docencoding>
-                        </configuration>
-                    </execution>
-                </executions>
-            </plugin>
+            <!--<plugin>-->
+                <!--<groupId>org.apache.maven.plugins</groupId>-->
+                <!--<artifactId>maven-javadoc-plugin</artifactId>-->
+                <!--&lt;!&ndash;<version>2.9</version>&ndash;&gt;-->
+                <!--<executions>-->
+                    <!--<execution>-->
+                        <!--<id>attach-javadocs</id>-->
+                        <!--<goals>-->
+                            <!--<goal>jar</goal>-->
+                        <!--</goals>-->
+                        <!--<configuration>-->
+                            <!--<additionalparam>-Xdoclint:none</additionalparam>-->
+                            <!--<charset>UTF-8</charset>-->
+                            <!--<encoding>UTF-8</encoding>-->
+                            <!--<docencoding>UTF-8</docencoding>-->
+                        <!--</configuration>-->
+                    <!--</execution>-->
+                <!--</executions>-->
+            <!--</plugin>-->
         </plugins>
     </build>
 

elab-db/src/main/java/com/elab/core/aop/annotations/EnableElabDB.java

@@ -1,7 +1,8 @@
 package com.elab.core.aop.annotations;
 
+import com.elab.core.spring.DaoScannerRegister;
+import com.elab.core.spring.config.BasicDaoConfig;
 import com.elab.core.spring.config.DataSourceConfigBean;
-import com.elab.core.spring.config.JdbcBeanConfig;
 import com.elab.core.spring.config.TransactionConfigBean;
 import org.springframework.context.annotation.Import;
 
@@ -10,7 +11,45 @@ import java.lang.annotation.*;
 @Retention(RetentionPolicy.RUNTIME)
 @Target(ElementType.TYPE)
 @Documented
-@Import({DataSourceConfigBean.class, JdbcBeanConfig.class, TransactionConfigBean.class})
+@Import({DaoScannerRegister.class, DataSourceConfigBean.class, BasicDaoConfig.class, TransactionConfigBean.class})
 public @interface EnableElabDB {
+    /**
+     * Alias for the {@link #basePackages()} attribute. Allows for more concise annotation
+     * declarations e.g.: {@code @ComponentScan("org.my.pkg")} instead of
+     * {@code @ComponentScan(basePackages="org.my.pkg")}.
+     *
+     * @return the array of 'basePackages'.
+     */
+    String[] value() default {};
+
+    /**
+     * Base packages to scan for annotated components.
+     * <p>
+     * {@link #value()} is an alias for (and mutually exclusive with) this attribute.
+     * <p>
+     * Use {@link #basePackageClasses()} for a type-safe alternative to String-based
+     * package names.
+     *
+     * @return the array of 'basePackages'.
+     */
+    String[] basePackages() default {};
+
+    /**
+     * Type-safe alternative to {@link #basePackages()} for specifying the packages to
+     * scan for annotated components. The package of each class specified will be scanned.
+     * <p>
+     * Consider creating a special no-op marker class or interface in each package that
+     * serves no purpose other than being referenced by this attribute.
+     *
+     * @return the array of 'basePackageClasses'.
+     */
+    Class<?>[] basePackageClasses() default {};
+
+    /**
+     * SQL文件的存放路径
+     *
+     * @return
+     */
+    String sqlConfigurableLocations() default "sql";
 
 }

elab-db/src/main/java/com/elab/core/spring/DaoScannerRegister.java

@@ -0,0 +1,87 @@
+package com.elab.core.spring;
+
+import com.elab.core.aop.annotations.EnableElabDB;
+import com.elab.core.sql.ConfigurableFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.support.BeanDefinitionRegistry;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.context.annotation.ImportBeanDefinitionRegistrar;
+import org.springframework.core.type.AnnotationMetadata;
+import org.springframework.stereotype.Component;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.StringUtils;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * 需要代理类的注册
+ *
+ * @author ： liukx
+ * @time ： 2020/4/1 - 19:17
+ */
+@Component
+public class DaoScannerRegister implements ImportBeanDefinitionRegistrar, ApplicationContextAware {
+    private Logger logger = LoggerFactory.getLogger(DaoScannerRegister.class);
+    private ApplicationContext applicationContext;
+
+    @Override
+    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+        this.applicationContext = applicationContext;
+    }
+
+    public ConfigurableFactory getConfigurableFactory(String configPath) {
+        ConfigurableFactory configurableFactory = new ConfigurableFactory();
+        configurableFactory.setSqlConfigurableLocations(configPath);
+        configurableFactory.buildSqlConfigurableFactory();
+        return configurableFactory;
+    }
+
+    @Override
+    public void registerBeanDefinitions(AnnotationMetadata metadata, BeanDefinitionRegistry beanDefinitionRegistry) {
+        Map<String, Object> defaultAttrs = metadata
+                .getAnnotationAttributes(EnableElabDB.class.getName());
+        if (defaultAttrs != null) {
+            Set<String> basePackageList = getBasePackages(defaultAttrs);
+            if (basePackageList.size() > 0) {
+                String[] basePackages = basePackageList.stream().toArray(String[]::new);
+                logger.debug(" 开启ElabDB , 注解扫描指定的包 : {} ", Arrays.toString(basePackages));
+                ConfigurableFactory sqlConfigurableLocations = getConfigurableFactory(defaultAttrs.get("sqlConfigurableLocations").toString
+                        ());
+                ClassPathDaoScanner scanner = new ClassPathDaoScanner(beanDefinitionRegistry);
+                scanner.setBasicBaseDaoName("basicBaseDao");
+                scanner.setResourceLoader(this.applicationContext);
+                scanner.setConfigurableFactory(sqlConfigurableLocations);
+                //
+                scanner.registerFilters();
+                //注册对应的daoInterface
+                scanner.scan(basePackages);
+            }
+        }
+    }
+
+    protected Set<String> getBasePackages(Map<String, Object> attributes) {
+
+        Set<String> basePackages = new HashSet<>();
+        for (String pkg : (String[]) attributes.get("value")) {
+            if (StringUtils.hasText(pkg)) {
+                basePackages.add(pkg);
+            }
+        }
+        for (String pkg : (String[]) attributes.get("basePackages")) {
+            if (StringUtils.hasText(pkg)) {
+                basePackages.add(pkg);
+            }
+        }
+        for (Class<?> clazz : (Class[]) attributes.get("basePackageClasses")) {
+            basePackages.add(ClassUtils.getPackageName(clazz));
+        }
+
+        return basePackages;
+    }
+}

elab-db/src/main/java/com/elab/core/spring/conditional/EnableConditional.java

@@ -0,0 +1,24 @@
+package com.elab.core.spring.conditional;
+
+import com.elab.core.spring.DaoScannerRegister;
+import org.springframework.context.annotation.Condition;
+import org.springframework.context.annotation.ConditionContext;
+import org.springframework.core.type.AnnotatedTypeMetadata;
+
+/**
+ * 判断是否是注解开启
+ *
+ * @author ： liukx
+ * @time ： 2020/4/1 - 19:53
+ */
+public class EnableConditional implements Condition {
+
+    @Override
+    public boolean matches(ConditionContext conditionContext, AnnotatedTypeMetadata annotatedTypeMetadata) {
+        DaoScannerRegister bean = conditionContext.getBeanFactory().getBean(DaoScannerRegister.class);
+        if (bean == null) {
+            return true;
+        }
+        return false;
+    }
+}

elab-db/src/main/java/com/elab/core/spring/config/BasicDaoConfig.java

@@ -0,0 +1,34 @@
+package com.elab.core.spring.config;
+
+import com.elab.core.dao.BasicBaseDao;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.jdbc.core.JdbcTemplate;
+
+import javax.sql.DataSource;
+
+/**
+ * 基础dao层配置
+ * <p>
+ * author ： liukx
+ *
+ * @time ： 2020/4/2 - 10:58
+ */
+@Configuration
+public class BasicDaoConfig {
+
+    @Bean
+    public JdbcTemplate jdbcTemplate(@Autowired DataSource dataSource) {
+        JdbcTemplate jdbcTemplate = new JdbcTemplate();
+        jdbcTemplate.setDataSource(dataSource);
+        return jdbcTemplate;
+    }
+
+    @Bean
+    public BasicBaseDao basicBaseDao(@Autowired JdbcTemplate jdbcTemplate) {
+        BasicBaseDao basicBaseDao = new BasicBaseDao();
+        basicBaseDao.setJdbcTemplate(jdbcTemplate);
+        return basicBaseDao;
+    }
+}

elab-db/src/main/java/com/elab/core/spring/config/JdbcBeanConfig.java

@@ -1,15 +1,12 @@
 package com.elab.core.spring.config;
 
-import com.elab.core.dao.BasicBaseDao;
 import com.elab.core.spring.DaoScannerConfigurer;
 import com.elab.core.sql.ConfigurableFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 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.jdbc.core.JdbcTemplate;
-
-import javax.sql.DataSource;
 
 /**
  * 持久层配置
@@ -22,22 +19,9 @@ import javax.sql.DataSource;
  * @email liukx@elab-plus.com
  **/
 @Configuration
+@Import({BasicDaoConfig.class})
 public class JdbcBeanConfig {
 
-    @Bean
-    public JdbcTemplate jdbcTemplate(@Autowired DataSource dataSource) {
-        JdbcTemplate jdbcTemplate = new JdbcTemplate();
-        jdbcTemplate.setDataSource(dataSource);
-        return jdbcTemplate;
-    }
-
-    @Bean
-    public BasicBaseDao basicBaseDao(@Autowired JdbcTemplate jdbcTemplate) {
-        BasicBaseDao basicBaseDao = new BasicBaseDao();
-        basicBaseDao.setJdbcTemplate(jdbcTemplate);
-        return basicBaseDao;
-    }
-
     @Bean
     public ConfigurableFactory getConfigurableFactory(@Autowired Environment env) {
         ConfigurableFactory configurableFactory = new ConfigurableFactory();
@@ -46,6 +30,7 @@ public class JdbcBeanConfig {
     }
 
     @Bean
+//    @Conditional(value = {EnableConditional.class})
     public DaoScannerConfigurer daoScannerConfigurer(@Autowired ConfigurableFactory configurableFactory, @Autowired Environment env) {
         DaoScannerConfigurer daoScannerConfigurer = new DaoScannerConfigurer();
         daoScannerConfigurer.setBasePackage(env.getProperty("jdbc.config.scan", "com.elab.**"));
@@ -54,5 +39,4 @@ public class JdbcBeanConfig {
         return daoScannerConfigurer;
     }
 
-
 }

elab-es/src/main/java/com/elab/es/client/annotation/EnableElastaticSearch.java

@@ -0,0 +1,21 @@
+package com.elab.es.client.annotation;
+
+import com.elab.es.client.configuration.ESConfiguration;
+import org.springframework.context.annotation.Import;
+
+import java.lang.annotation.*;
+
+@Target({ElementType.TYPE})
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+@Inherited
+@Import({ESConfiguration.class})
+public @interface EnableElastaticSearch {
+    String[] basePackages() default {};
+
+    String[] value() default {};
+
+    String[] entityPath() default {};
+
+    boolean printregmsg() default false;
+}

elab-es/src/main/java/com/elab/es/client/configuration/ElabESProperties.java

@@ -0,0 +1,135 @@
+package com.elab.es.client.configuration;
+
+import org.springframework.boot.autoconfigure.elasticsearch.jest.JestProperties;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * ES配置属性
+ *
+ * @author ： liukx
+ * @time ： 2020/3/31 - 10:38
+ */
+public class ElabESProperties {
+    /**
+     * Comma-separated list of the Elasticsearch instances to use.
+     */
+    private List<String> uris = new ArrayList<String>(
+            Collections.singletonList("http://localhost:9200"));
+
+    /**
+     * Login username.
+     */
+    private String username;
+
+    /**
+     * Login password.
+     */
+    private String password;
+
+    /**
+     * Whether to enable connection requests from multiple execution threads.
+     */
+    private boolean multiThreaded = true;
+
+    /**
+     * Connection timeout.
+     */
+    private Duration connectionTimeout = Duration.ofSeconds(3);
+
+    /**
+     * Read timeout.
+     */
+    private Duration readTimeout = Duration.ofSeconds(3);
+
+    /**
+     * Proxy settings.
+     */
+    private final JestProperties.Proxy proxy = new JestProperties.Proxy();
+
+    public List<String> getUris() {
+        return this.uris;
+    }
+
+    public void setUris(List<String> uris) {
+        this.uris = uris;
+    }
+
+    public String getUsername() {
+        return this.username;
+    }
+
+    public void setUsername(String username) {
+        this.username = username;
+    }
+
+    public String getPassword() {
+        return this.password;
+    }
+
+    public void setPassword(String password) {
+        this.password = password;
+    }
+
+    public boolean isMultiThreaded() {
+        return this.multiThreaded;
+    }
+
+    public void setMultiThreaded(boolean multiThreaded) {
+        this.multiThreaded = multiThreaded;
+    }
+
+    public Duration getConnectionTimeout() {
+        return this.connectionTimeout;
+    }
+
+    public void setConnectionTimeout(Duration connectionTimeout) {
+        this.connectionTimeout = connectionTimeout;
+    }
+
+    public Duration getReadTimeout() {
+        return this.readTimeout;
+    }
+
+    public void setReadTimeout(Duration readTimeout) {
+        this.readTimeout = readTimeout;
+    }
+
+    public JestProperties.Proxy getProxy() {
+        return this.proxy;
+    }
+
+    public static class Proxy {
+
+        /**
+         * Proxy host the HTTP client should use.
+         */
+        private String host;
+
+        /**
+         * Proxy port the HTTP client should use.
+         */
+        private Integer port;
+
+        public String getHost() {
+            return this.host;
+        }
+
+        public void setHost(String host) {
+            this.host = host;
+        }
+
+        public Integer getPort() {
+            return this.port;
+        }
+
+        public void setPort(Integer port) {
+            this.port = port;
+        }
+
+    }
+
+}

elab-log/pom.xml

@@ -87,25 +87,25 @@
             </plugin>
 
             <!-- 打包javadoc插件 -->
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-javadoc-plugin</artifactId>
-                <!--<version>2.9</version>-->
-                <executions>
-                    <execution>
-                        <id>attach-javadocs</id>
-                        <goals>
-                            <goal>jar</goal>
-                        </goals>
-                        <configuration>
-                            <additionalparam>-Xdoclint:none</additionalparam>
-                            <charset>UTF-8</charset>
-                            <encoding>UTF-8</encoding>
-                            <docencoding>UTF-8</docencoding>
-                        </configuration>
-                    </execution>
-                </executions>
-            </plugin>
+            <!--<plugin>-->
+                <!--<groupId>org.apache.maven.plugins</groupId>-->
+                <!--<artifactId>maven-javadoc-plugin</artifactId>-->
+                <!--&lt;!&ndash;<version>2.9</version>&ndash;&gt;-->
+                <!--<executions>-->
+                    <!--<execution>-->
+                        <!--<id>attach-javadocs</id>-->
+                        <!--<goals>-->
+                            <!--<goal>jar</goal>-->
+                        <!--</goals>-->
+                        <!--<configuration>-->
+                            <!--<additionalparam>-Xdoclint:none</additionalparam>-->
+                            <!--<charset>UTF-8</charset>-->
+                            <!--<encoding>UTF-8</encoding>-->
+                            <!--<docencoding>UTF-8</docencoding>-->
+                        <!--</configuration>-->
+                    <!--</execution>-->
+                <!--</executions>-->
+            <!--</plugin>-->
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-compiler-plugin</artifactId>

elab-mq/pom.xml

@@ -12,6 +12,11 @@
     <artifactId>elab-mq</artifactId>
 
     <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-autoconfigure</artifactId>
+            <scope>provided</scope>
+        </dependency>
         <dependency>
             <groupId>com.elab.core</groupId>
             <artifactId>elab-core</artifactId>

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

@@ -7,7 +7,10 @@ import com.elab.mq.listener.AbstractMessageListener;
 import com.elab.mq.listener.MessageListenerWrapper;
 import com.elab.mq.msg.IMsgProducerFacade;
 import com.elab.mq.msg.impl.MsgProducerImpl;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
@@ -31,6 +34,8 @@ public class RocketMQConfiguration {
 
     private final String prefix = "elab.mq";
 
+    private Logger logger = LoggerFactory.getLogger(RocketMQConfiguration.class);
+
     @Bean
     public IMsgProducerFacade producerBean(@Autowired Environment env) {
         IMsgProducerFacade producerBean = new MsgProducerImpl();
@@ -40,20 +45,18 @@ public class RocketMQConfiguration {
     }
 
     @Bean
+    @ConditionalOnBean(value = {AbstractMessageListener.class})
     public Consumer consumerBean(@Autowired Environment env, @Autowired MessageListenerWrapper
             messageListenerWrapper, @Autowired List<AbstractMessageListener> messageListeners) {
         Consumer consumer = ONSFactory.createConsumer(mqProperties(env));
-
-        Set<String> existTopic = new HashSet<>();
         if (messageListeners.size() > 0) {
+            Set<String> existTopic = new HashSet<>();
             for (int i = 0; i < messageListeners.size(); i++) {
                 AbstractMessageListener abstractMessageListener = messageListeners.get(i);
                 String topic = abstractMessageListener.topic();
-                boolean exist = existTopic.add(topic);
-                if (!exist) {
-                    throw new RuntimeException(" mq topic : " + topic + " Can not repeat !");
-                }
-                consumer.subscribe(topic, "*", messageListenerWrapper);
+                String tag = abstractMessageListener.tag();
+                consumer.subscribe(topic, tag, messageListenerWrapper);
+                logger.debug(" MQ 消费者注册 -> topic : {} , tag : {} ,class : {}", topic, tag, abstractMessageListener.toString());
             }
             consumer.start();
         }

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

@@ -4,17 +4,17 @@ import com.aliyun.openservices.ons.api.Action;
 import com.aliyun.openservices.ons.api.ConsumeContext;
 import com.aliyun.openservices.ons.api.Message;
 import com.aliyun.openservices.ons.api.MessageListener;
-import com.elab.core.exception.BusinessException;
 import com.elab.core.utils.DateUtils;
 import com.elab.core.utils.ObjectUtils;
 import com.elab.mq.dao.IConsumerDao;
-import com.elab.mq.dao.IProducerDao;
 import com.elab.mq.model.ConsumerEntity;
 import com.elab.mq.model.MessageModel;
-import com.elab.mq.model.ProducerEntity;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+
+import java.util.Date;
 
 /**
  * 抽象成客户端能够使用的方法
@@ -26,12 +26,12 @@ public abstract class AbstractMessageListener implements MessageListener {
 
     private Logger logger = LoggerFactory.getLogger(getClass());
 
-    @Autowired
-    private IProducerDao producerDao;
-
     @Autowired
     private IConsumerDao consumerDao;
 
+    @Value("${spring.application.name}")
+    private String moduleName;
+
 
     /**
      * 客户端关注的topic消息,实现了该方法,消息监听被触发时会与之匹配。
@@ -50,47 +50,57 @@ public abstract class AbstractMessageListener implements MessageListener {
     }
 
 
-    private ConsumerEntity init(String producerId) throws Exception {
+    private ConsumerEntity init(MessageModel message) throws Exception {
+        String msgID = message.getMsgID();
+        Integer producerId = 0;
+
+        if (message.getProducerId() != null) {
+            producerId = Integer.valueOf(message.getProducerId());
+        }
+
+        String topic = message.getTopic();
+        String tag = message.getTag();
         ConsumerEntity oldConsumerEntity = new ConsumerEntity();
-        ProducerEntity producerEntity = producerDao.selectById(producerId);
-        logger.info("获取生产者的信息：" + ObjectUtils.objectParseJsonStr(producerEntity));
-        if (producerEntity != null) {
-            logger.info("进入消费者验证...");
-            ConsumerEntity consumerEntity = new ConsumerEntity();
-            consumerEntity.setProducerId(producerEntity.getId());
-            consumerEntity.setTopicId(producerEntity.getTopicId());
-            consumerEntity.setStatus(1);
-            oldConsumerEntity = consumerDao.selectByObject(consumerEntity);
-            if (oldConsumerEntity != null && oldConsumerEntity.getConsumerStatus().equals(1)) {
-                logger.info("消费者验证失败，已经消费成功过一次，不允许重复消费");
-                throw new BusinessException("请不要重复消费，key：" + producerId);
-            }
-            logger.info("消费者验证通过，消费者消费的数据匹配上生产者数据，");
-            if (oldConsumerEntity != null) {
-                oldConsumerEntity.setRetryCount(oldConsumerEntity.getRetryCount() + 1);
-                oldConsumerEntity.setConsumerStatus(-1);
-                oldConsumerEntity.setUpdated(DateUtils.getCurrentDateTime());
-                consumerDao.updateById(oldConsumerEntity);
-            } else {
-                oldConsumerEntity = new ConsumerEntity();
-                oldConsumerEntity.setProducerId(producerEntity.getId());
-                oldConsumerEntity.setContent(producerEntity.getContent());
-                oldConsumerEntity.setHouseId(producerEntity.getHouseId());
-                oldConsumerEntity.setModuleName(producerEntity.getModuleName());
-                oldConsumerEntity.setStatus(1);
-                oldConsumerEntity.setCreated(DateUtils.getCurrentDateTime());
-                oldConsumerEntity.setRetryCount(0);
-                oldConsumerEntity.setTopicId(producerEntity.getTopicId());
-                oldConsumerEntity.setConsumerStatus(0);
-                int id = consumerDao.insert(oldConsumerEntity);
-                oldConsumerEntity.setId(id);
-            }
+        logger.info("进入消费者验证...");
+        ConsumerEntity consumerEntity = new ConsumerEntity();
+
+        consumerEntity.setProducerId(producerId);
+        consumerEntity.setTopicId(topic);
+        consumerEntity.setModuleName(moduleName);
+        consumerEntity.setStatus(1);
+        consumerEntity.setMsgId(msgID);
+        consumerEntity.setTag(tag);
+        consumerEntity.setModuleMethod(getClass().getName());
+        oldConsumerEntity = consumerDao.selectByObject(consumerEntity);
+        if (oldConsumerEntity != null && oldConsumerEntity.getConsumerStatus().equals(1)) {
+            logger.warn("消费者验证失败，已经消费成功过一次，不允许重复消费 消费编号 : " + oldConsumerEntity.getId());
+            // throw new BusinessException("请不要重复消费，key：" + message);
+            return oldConsumerEntity;
+        }
+        logger.info("消费者验证通过，消费者消费的数据匹配上生产者数据，");
+        if (oldConsumerEntity != null) {
+            oldConsumerEntity.setRetryCount(oldConsumerEntity.getRetryCount() + 1);
+            oldConsumerEntity.setConsumerStatus(-1);
+            oldConsumerEntity.setUpdated(DateUtils.getCurrentDateTime());
+            consumerDao.updateById(oldConsumerEntity);
         } else {
-            logger.warn("生产者为空，无法匹配消费者");
+            oldConsumerEntity = new ConsumerEntity();
+            oldConsumerEntity.setProducerId(producerId);
+            oldConsumerEntity.setContent(new String(message.getBody()));
+            oldConsumerEntity.setHouseId(message.getHouseId());
+            oldConsumerEntity.setModuleName(moduleName);
+            oldConsumerEntity.setStatus(1);
+            oldConsumerEntity.setCreated(DateUtils.getCurrentDateTime());
+            oldConsumerEntity.setRetryCount(0);
+            oldConsumerEntity.setTopicId(topic);
+            oldConsumerEntity.setConsumerStatus(0);
+            oldConsumerEntity.setMsgId(msgID);
+            oldConsumerEntity.setTag(tag);
+            oldConsumerEntity.setModuleMethod(getClass().getName());
+            int id = consumerDao.insert(oldConsumerEntity);
+            oldConsumerEntity.setId(id);
         }
         return oldConsumerEntity;
-//        ConsumerEntity consumerEntity = new ConsumerEntity();
-
     }
 
 
@@ -109,15 +119,23 @@ public abstract class AbstractMessageListener implements MessageListener {
         logger.debug("消息处理被触发 : " + message.toString());
         Action action = null;
         try {
-            ConsumerEntity oldConsumerEntity = init(message.getKey());
+            ConsumerEntity oldConsumerEntity = init(messageModel);
+            //如果已经成功过一次
+            if (oldConsumerEntity.getConsumerStatus() == 1) {
+                logger.warn(" 重复消费 过滤掉. 消费编号 : " + oldConsumerEntity.getId());
+                return Action.CommitMessage;
+            }
+
             logger.info("更新消费者数据： " + ObjectUtils.objectParseJsonStr(oldConsumerEntity));
             action = consume0(messageModel, consumeContext);
 
-            if (Action.ReconsumeLater.equals(action)) {
+            if (action == null || Action.ReconsumeLater.equals(action)) {
                 oldConsumerEntity.setConsumerStatus(-1);
+                oldConsumerEntity.setUpdated(new Date());
                 consumerDao.updateById(oldConsumerEntity);
             } else {
                 oldConsumerEntity.setConsumerStatus(1);
+                oldConsumerEntity.setUpdated(new Date());
                 consumerDao.updateById(oldConsumerEntity);
             }
         } catch (Exception e) {

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

@@ -4,6 +4,8 @@ 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 org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
@@ -21,8 +23,11 @@ public class MessageListenerWrapper implements MessageListener {
     @Autowired(required = false)
     private List<AbstractMessageListener> messageListeners;
 
+    private Logger logger = LoggerFactory.getLogger(MessageListenerWrapper.class);
+
     @Override
     public Action consume(Message message, ConsumeContext consumeContext) {
+        Action result = Action.CommitMessage;
         if (messageListeners != null) {
             try {
                 for (int i = 0; i < messageListeners.size(); i++) {
@@ -30,7 +35,11 @@ public class MessageListenerWrapper implements MessageListener {
                     String currentTopic = message.getTopic();
                     String topic = abstractMessageListener.topic();
                     if (topic.equals(currentTopic)) {
-                        abstractMessageListener.consume(message, consumeContext);
+                        Action actionResult = abstractMessageListener.consume(message, consumeContext);
+                        if (actionResult == null || Action.ReconsumeLater == actionResult) {
+                            logger.warn(" MQ 消费失败 : " + abstractMessageListener);
+                            result = Action.ReconsumeLater;
+                        }
                     }
                 }
             } catch (Exception e) {
@@ -38,6 +47,6 @@ public class MessageListenerWrapper implements MessageListener {
                 return Action.ReconsumeLater;
             }
         }
-        return Action.CommitMessage;
+        return result;
     }
 }

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

@@ -28,6 +28,16 @@ public class ConsumerEntity {
     @Column(name = "module_name")
     @ApiModelProperty(name = "moduleName", value = "微服务模块名")
     private String moduleName;
+    @Column(name = "msg_id")
+    @ApiModelProperty(name = "msgId", value = "MQ的消息编号")
+    private String msgId;
+
+    @Column(name = "module_method")
+    @ApiModelProperty(name = "moduleMethod", value = "模块方法")
+    private String moduleMethod;
+    @Column(name = "tag")
+    @ApiModelProperty(name = "tag", value = "标签 业务区分标识")
+    private String tag;
 
     @Column(name = "house_id")
     @ApiModelProperty(name = "houseId", value = "项目id")
@@ -51,7 +61,6 @@ public class ConsumerEntity {
 
     /**
      * 状态：1  有效  -1  无效
-     *
      */
     @Column(name = "status")
     @ApiModelProperty(name = "status", value = "状态：1  有效  -1  无效")
@@ -59,7 +68,6 @@ public class ConsumerEntity {
 
     /**
      * 创建时间
-     *
      */
     @Column(name = "created")
     @ApiModelProperty(name = "created", value = "创建时间")
@@ -67,7 +75,6 @@ public class ConsumerEntity {
 
     /**
      * 修改时间
-     *
      */
     @Column(name = "updated")
     @ApiModelProperty(name = "updated", value = "修改时间")
@@ -75,7 +82,6 @@ public class ConsumerEntity {
 
     /**
      * 创建者
-     *
      */
     @Column(name = "creator")
     @ApiModelProperty(name = "creator", value = "创建者")
@@ -83,12 +89,35 @@ public class ConsumerEntity {
 
     /**
      * 修改者
-     *
      */
     @Column(name = "updator")
     @ApiModelProperty(name = "updator", value = "修改者")
     private String updator;
 
+    public String getTag() {
+        return tag;
+    }
+
+    public void setTag(String tag) {
+        this.tag = tag;
+    }
+
+    public String getMsgId() {
+        return msgId;
+    }
+
+    public void setMsgId(String msgId) {
+        this.msgId = msgId;
+    }
+
+    public String getModuleMethod() {
+        return moduleMethod;
+    }
+
+    public void setModuleMethod(String moduleMethod) {
+        this.moduleMethod = moduleMethod;
+    }
+
     public Integer getId() {
         return id;
     }

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

@@ -4,6 +4,8 @@ import com.alibaba.fastjson.JSON;
 import com.aliyun.openservices.ons.api.Message;
 import com.elab.core.utils.StringUtils;
 
+import java.util.Properties;
+
 /**
  * 消息容器承装实体
  *
@@ -15,8 +17,18 @@ public class MessageModel<T> extends Message {
 
     private T object;
     private String groupId;
+    /**
+     * 房源编号
+     */
     private Integer houseId;
+    /**
+     * 模块名称
+     */
     private String moduleName;
+    /**
+     * 生产者表的主键编号
+     */
+    private String producerId;
 
 
     /**
@@ -32,6 +44,14 @@ public class MessageModel<T> extends Message {
         setKey(key);
         setTopic(topic);
         setTag(tags);
+        addDefaultData();
+    }
+
+    private void addDefaultData() {
+        Properties userProperties = getUserProperties();
+        if (producerId != null) {
+            userProperties.setProperty("producerId", producerId + "");
+        }
     }
 
     /**
@@ -46,6 +66,7 @@ public class MessageModel<T> extends Message {
         setKey(key);
         setTopic(topic);
         setTag("*");
+        addDefaultData();
     }
 
     public MessageModel(Message message) {
@@ -116,4 +137,21 @@ public class MessageModel<T> extends Message {
     public void setModuleName(String moduleName) {
         this.moduleName = moduleName;
     }
+
+    public String getProducerId() {
+        return getDefault("producerId", null);
+    }
+
+    public void setProducerId(String producerId) {
+        putUserProperties("producerId", producerId);
+    }
+
+    public String getDefault(String key, String defaultValue) {
+        String userProperties = getUserProperties(key);
+        if (userProperties == null || "".equals(userProperties.trim())) {
+            return defaultValue;
+        }
+        return userProperties;
+    }
+
 }

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

@@ -45,10 +45,12 @@ public class ProducerEntity {
     @ApiModelProperty(name = "content", value = "内容")
     private String content;
 
+    @Column(name = "result_content")
+    @ApiModelProperty(name = "resultContent", value = "返回结果内容")
+    private String resultContent;
 
     /**
      * 状态：1  有效  -1  无效
-     *
      */
     @Column(name = "status")
     @ApiModelProperty(name = "status", value = "状态：1  有效  -1  无效")
@@ -56,7 +58,6 @@ public class ProducerEntity {
 
     /**
      * 创建时间
-     *
      */
     @Column(name = "created")
     @ApiModelProperty(name = "created", value = "创建时间")
@@ -64,7 +65,6 @@ public class ProducerEntity {
 
     /**
      * 修改时间
-     *
      */
     @Column(name = "updated")
     @ApiModelProperty(name = "updated", value = "修改时间")
@@ -72,7 +72,6 @@ public class ProducerEntity {
 
     /**
      * 创建者
-     *
      */
     @Column(name = "creator")
     @ApiModelProperty(name = "creator", value = "创建者")
@@ -80,12 +79,19 @@ public class ProducerEntity {
 
     /**
      * 修改者
-     *
      */
     @Column(name = "updator")
     @ApiModelProperty(name = "updator", value = "修改者")
     private String updator;
 
+    public String getResultContent() {
+        return resultContent;
+    }
+
+    public void setResultContent(String resultContent) {
+        this.resultContent = resultContent;
+    }
+
     public Integer getId() {
         return id;
     }

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

@@ -4,6 +4,7 @@ import com.aliyun.openservices.ons.api.SendResult;
 import com.aliyun.openservices.ons.api.bean.ProducerBean;
 import com.elab.core.utils.DateUtils;
 import com.elab.core.utils.ObjectUtils;
+import com.elab.core.utils.StringUtils;
 import com.elab.mq.dao.IProducerDao;
 import com.elab.mq.model.MessageModel;
 import com.elab.mq.model.ProducerEntity;
@@ -13,6 +14,7 @@ import com.elab.mq.msg.adptor.SendCallbackAdaptor;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 
 import java.util.Date;
 import java.util.Properties;
@@ -32,16 +34,26 @@ public class MsgProducerImpl extends ProducerBean implements IMsgProducerFacade
     @Autowired
     private IProducerDao producerDao;
 
+    @Value("${spring.application.name}")
+    private String moduleName;
+
 
     /**
      * 往生产者表中写入数据
+     *
      * @param message
      * @return
      */
     private int insertRecordProducer(MessageModel message) {
         String content = ObjectUtils.objectParseJsonStr(message.getObject(Object.class));
         ProducerEntity producerEntity = new ProducerEntity();
-        producerEntity.setModuleName(message.getModuleName());
+
+        if (StringUtils.isEmpty(message.getModuleName())) {
+            producerEntity.setModuleName(moduleName);
+        } else {
+            producerEntity.setModuleName(message.getModuleName());
+        }
+
         producerEntity.setHouseId(message.getHouseId());
         producerEntity.setProducerStatus(0);
         producerEntity.setGroupId(message.getGroupId());
@@ -52,7 +64,6 @@ public class MsgProducerImpl extends ProducerBean implements IMsgProducerFacade
         int id = 0;
         try {
             id = producerDao.insert(producerEntity);
-            message.setKey(id + "");
         } catch (Exception e) {
             e.printStackTrace();
             logger.error("往生产者表中插入记录失败：，入参：" + ObjectUtils.objectParseJsonStr(message), e);
@@ -62,15 +73,18 @@ public class MsgProducerImpl extends ProducerBean implements IMsgProducerFacade
 
     /**
      * 修改生产者数据
+     *
      * @param id
+     * @param result
      * @return
      */
-    private void updateRecordProducer(int id) {
+    private void updateRecordProducer(int id, Integer status, String result) {
 
         try {
             ProducerEntity producerEntity = producerDao.selectById(id + "");
             if (producerEntity != null) {
-                producerEntity.setProducerStatus(1);
+                producerEntity.setProducerStatus(status);
+                producerEntity.setResultContent(result);
                 producerDao.updateById(producerEntity);
             }
         } catch (Exception e) {
@@ -83,18 +97,23 @@ public class MsgProducerImpl extends ProducerBean implements IMsgProducerFacade
     public void sendAsync(MessageModel message, SendCallbackAdaptor sendCallback) {
         logger.debug(" 发送一条异步消息 " + message + " -> " + message.toString() + " 回调类 : " + sendCallback.getClass().getName());
         super.sendAsync(message, sendCallback);
-
-
     }
 
     @Override
     public SendResultModel send(MessageModel message) {
         logger.debug(" 发送一条消息 " + message.getMsgID() + " -> " + message.toString());
         int id = insertRecordProducer(message);
-        SendResult result = super.send(message);
-        updateRecordProducer(id);
-        logger.debug(" 消息发送结果 : " + result.toString());
-        return new SendResultModel(result);
+        try {
+            message.setProducerId(id + "");
+            SendResult result = super.send(message);
+            updateRecordProducer(id, 1, result.toString());
+            logger.debug(" 消息发送结果 : " + result.toString());
+            return new SendResultModel(result);
+        } catch (Exception e) {
+            logger.error("消息发送异常", e);
+            updateRecordProducer(id, -1, e.getMessage());
+        }
+        return null;
     }
 
     @Override

elab-rocketMQ/pom.xml

@@ -12,7 +12,7 @@
     <artifactId>elab-rocketMQ</artifactId>
 
     <properties>
-        <springframework.version>5.0.9.RELEASE</springframework.version>
+        <springframework.version>5.1.8.RELEASE</springframework.version>
         <rocketmq-spring-boot-starter-version>2.0.3</rocketmq-spring-boot-starter-version>
         <springboot.version></springboot.version>
     </properties>
@@ -48,5 +48,11 @@
             <artifactId>guava</artifactId>
         </dependency>
     </dependencies>
-
+    <distributionManagement>
+        <repository>
+            <id>releases</id>
+            <name>Nexus Release Repository</name>
+            <url>http://192.168.0.11:8081/nexus/content/repositories/elab/</url>
+        </repository>
+    </distributionManagement>
 </project>

elab-rocketMQ/src/main/java/com/elab/mq/rocket/anno/EnableRocketMQ.java

@@ -1,9 +1,12 @@
 package com.elab.mq.rocket.anno;
 
 import com.elab.core.aop.annotations.EnableElabDB;
+import com.elab.mq.rocket.configuration.RocketMQAutoConfiguration2;
 import com.elab.mq.rocket.configuration.RocketMQConfiguration;
+import org.apache.rocketmq.spring.autoconfigure.ListenerContainerConfiguration;
+import org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration;
+import org.springframework.boot.autoconfigure.ImportAutoConfiguration;
 import org.springframework.context.annotation.ComponentScan;
-import org.springframework.context.annotation.Import;
 
 import java.lang.annotation.*;
 
@@ -11,9 +14,10 @@ import java.lang.annotation.*;
 @Retention(RetentionPolicy.RUNTIME)
 @Documented
 @Inherited
-@Import({RocketMQConfiguration.class})
+@ImportAutoConfiguration(value = {RocketMQConfiguration.class, RocketMQAutoConfiguration2.class}, exclude =
+        {RocketMQAutoConfiguration.class, ListenerContainerConfiguration.class})
 @EnableElabDB
-@ComponentScan({"com.elab.mq.rocket.msg"})
+@ComponentScan(value = {"com.elab.mq.rocket.msg"})
 public @interface EnableRocketMQ {
     String[] value() default {};
 }

elab-rocketMQ/src/main/java/com/elab/mq/rocket/configuration/DefaultListenerContainerConfiguration.java

@@ -0,0 +1,128 @@
+package com.elab.mq.rocket.configuration;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.rocketmq.client.AccessChannel;
+import org.apache.rocketmq.spring.annotation.ConsumeMode;
+import org.apache.rocketmq.spring.annotation.MessageModel;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.aop.framework.AopProxyUtils;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.SmartInitializingSingleton;
+import org.springframework.beans.factory.support.BeanDefinitionValidationException;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.core.env.StandardEnvironment;
+import org.springframework.util.StringUtils;
+
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicLong;
+
+
+public class DefaultListenerContainerConfiguration implements ApplicationContextAware, SmartInitializingSingleton {
+    private Logger log = LoggerFactory.getLogger(DefaultListenerContainerConfiguration.class);
+
+    private ConfigurableApplicationContext applicationContext;
+
+    private AtomicLong counter = new AtomicLong(0);
+
+    private StandardEnvironment environment;
+
+    private RocketMQProperties rocketMQProperties;
+
+    private ObjectMapper objectMapper;
+
+    public DefaultListenerContainerConfiguration(ObjectMapper rocketMQMessageObjectMapper,
+                                                 StandardEnvironment environment,
+                                                 RocketMQProperties rocketMQProperties) {
+        this.objectMapper = rocketMQMessageObjectMapper;
+        this.environment = environment;
+        this.rocketMQProperties = rocketMQProperties;
+    }
+
+    @Override
+    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+        this.applicationContext = (ConfigurableApplicationContext) applicationContext;
+    }
+
+    @Override
+    public void afterSingletonsInstantiated() {
+        Map<String, Object> beans = this.applicationContext.getBeansWithAnnotation(RocketMQMessageListener.class);
+
+        if (Objects.nonNull(beans)) {
+            beans.forEach(this::registerContainer);
+        }
+    }
+
+    private void registerContainer(String beanName, Object bean) {
+        Class<?> clazz = AopProxyUtils.ultimateTargetClass(bean);
+
+        if (!RocketMQListener.class.isAssignableFrom(bean.getClass())) {
+            throw new IllegalStateException(clazz + " is not instance of " + RocketMQListener.class.getName());
+        }
+
+        RocketMQMessageListener annotation = clazz.getAnnotation(RocketMQMessageListener.class);
+        validate(annotation);
+
+        String containerBeanName = String.format("%s_%s", DefaultRocketMQListenerContainer.class.getName(),
+                counter.incrementAndGet());
+//        GenericApplicationContext genericApplicationContext = (GenericApplicationContext) applicationContext;
+
+//        genericApplicationContext.registerBean(containerBeanName, DefaultRocketMQListenerContainer.class,
+//            () -> createRocketMQListenerContainer(containerBeanName, bean, annotation));
+        DefaultRocketMQListenerContainer container = createRocketMQListenerContainer(containerBeanName, bean,
+                annotation);
+        if (!container.isRunning()) {
+            try {
+                container.start();
+            } catch (Exception e) {
+                log.error("Started container failed. {}", container, e);
+                throw new RuntimeException(e);
+            }
+        }
+
+        log.info("Register the listener to container, listenerBeanName:{}, containerBeanName:{}", beanName, containerBeanName);
+    }
+
+    private DefaultRocketMQListenerContainer createRocketMQListenerContainer(String name, Object bean, RocketMQMessageListener annotation) {
+        DefaultRocketMQListenerContainer container = new DefaultRocketMQListenerContainer();
+
+        String nameServer = environment.resolvePlaceholders(annotation.nameServer());
+        nameServer = StringUtils.isEmpty(nameServer) ? rocketMQProperties.getNameServer() : nameServer;
+        String accessChannel = environment.resolvePlaceholders(annotation.accessChannel());
+        container.setNameServer(nameServer);
+        if (!StringUtils.isEmpty(accessChannel)) {
+            container.setAccessChannel(AccessChannel.valueOf(accessChannel));
+        }
+        container.setTopic(environment.resolvePlaceholders(annotation.topic()));
+        container.setConsumerGroup(environment.resolvePlaceholders(annotation.consumerGroup()));
+        container.setRocketMQMessageListener(annotation);
+        container.setRocketMQListener((RocketMQListener) bean);
+        container.setObjectMapper(objectMapper);
+        container.setName(name);  // REVIEW ME, use the same clientId or multiple?
+        container.setApplicationContext(this.applicationContext);
+        log.info(" 创建 RocketMQ 信息 :nameServer -> {}, name -> {}, topic -> {} , group -> {}", nameServer,name,
+                container.getTopic(),
+                container.getConsumerGroup());
+        try {
+            container.afterPropertiesSet();
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+        return container;
+    }
+
+    private void validate(RocketMQMessageListener annotation) {
+        if (annotation.consumeMode() == ConsumeMode.ORDERLY &&
+                annotation.messageModel() == MessageModel.BROADCASTING) {
+            throw new BeanDefinitionValidationException(
+                    "Bad annotation definition in @RocketMQMessageListener, messageModel BROADCASTING does not support ORDERLY message!");
+        }
+    }
+}

elab-rocketMQ/src/main/java/com/elab/mq/rocket/configuration/RocketMQAutoConfiguration2.java

@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.elab.mq.rocket.configuration;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.rocketmq.acl.common.AclClientRPCHook;
+import org.apache.rocketmq.acl.common.SessionCredentials;
+import org.apache.rocketmq.client.AccessChannel;
+import org.apache.rocketmq.client.MQAdmin;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.spring.autoconfigure.ExtProducerResetConfiguration;
+import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
+import org.apache.rocketmq.spring.config.RocketMQConfigUtils;
+import org.apache.rocketmq.spring.config.RocketMQTransactionAnnotationProcessor;
+import org.apache.rocketmq.spring.config.TransactionHandlerRegistry;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+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.boot.autoconfigure.jackson.JacksonAutoConfiguration;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.context.annotation.Role;
+import org.springframework.core.env.Environment;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+import javax.annotation.PostConstruct;
+
+@Configuration
+@EnableConfigurationProperties(RocketMQProperties.class)
+@ConditionalOnClass({MQAdmin.class, ObjectMapper.class})
+@ConditionalOnProperty(prefix = "rocketmq", value = "name-server", matchIfMissing = true)
+@Import({DefaultListenerContainerConfiguration.class, ExtProducerResetConfiguration.class})
+@AutoConfigureAfter(JacksonAutoConfiguration.class)
+public class RocketMQAutoConfiguration2 {
+    private static final Logger log = LoggerFactory.getLogger(RocketMQAutoConfiguration2.class);
+
+    @Autowired
+    private Environment environment;
+
+    @PostConstruct
+    public void checkProperties() {
+        String nameServer = environment.getProperty("rocketmq.name-server", String.class);
+        log.debug("rocketmq.nameServer = {}", nameServer);
+        if (nameServer == null) {
+            log.warn("The necessary spring property 'rocketmq.name-server' is not defined, all rockertmq beans creation are skipped!");
+        }
+    }
+
+
+    @Bean
+    @ConditionalOnMissingBean(DefaultMQProducer.class)
+    @ConditionalOnProperty(prefix = "rocketmq", value = {"name-server", "producer.group"})
+    public DefaultMQProducer defaultMQProducer(RocketMQProperties rocketMQProperties) {
+        RocketMQProperties.Producer producerConfig = rocketMQProperties.getProducer();
+        String nameServer = rocketMQProperties.getNameServer();
+        String groupName = producerConfig.getGroup();
+        Assert.hasText(nameServer, "[rocketmq.name-server] must not be null");
+        Assert.hasText(groupName, "[rocketmq.producer.group] must not be null");
+
+        String accessChannel = rocketMQProperties.getAccessChannel();
+
+        DefaultMQProducer producer;
+        String ak = rocketMQProperties.getProducer().getAccessKey();
+        String sk = rocketMQProperties.getProducer().getSecretKey();
+        if (!StringUtils.isEmpty(ak) && !StringUtils.isEmpty(sk)) {
+            producer = new DefaultMQProducer(groupName, new AclClientRPCHook(new SessionCredentials(ak, sk)),
+                    rocketMQProperties.getProducer().isEnableMsgTrace(),
+                    rocketMQProperties.getProducer().getCustomizedTraceTopic());
+            producer.setVipChannelEnabled(false);
+        } else {
+            producer = new DefaultMQProducer(groupName, rocketMQProperties.getProducer().isEnableMsgTrace(),
+                    rocketMQProperties.getProducer().getCustomizedTraceTopic());
+        }
+
+        producer.setNamesrvAddr(nameServer);
+        if (!StringUtils.isEmpty(accessChannel)) {
+            producer.setAccessChannel(AccessChannel.valueOf(accessChannel));
+        }
+        producer.setSendMsgTimeout(producerConfig.getSendMessageTimeout());
+        producer.setRetryTimesWhenSendFailed(producerConfig.getRetryTimesWhenSendFailed());
+        producer.setRetryTimesWhenSendAsyncFailed(producerConfig.getRetryTimesWhenSendAsyncFailed());
+        producer.setMaxMessageSize(producerConfig.getMaxMessageSize());
+        producer.setCompressMsgBodyOverHowmuch(producerConfig.getCompressMessageBodyThreshold());
+        producer.setRetryAnotherBrokerWhenNotStoreOK(producerConfig.isRetryNextServer());
+
+        return producer;
+    }
+
+    @Bean(destroyMethod = "destroy")
+    @ConditionalOnBean(DefaultMQProducer.class)
+    @ConditionalOnMissingBean(name = RocketMQConfigUtils.ROCKETMQ_TEMPLATE_DEFAULT_GLOBAL_NAME)
+    public RocketMQTemplate rocketMQTemplate(DefaultMQProducer mqProducer, ObjectMapper rocketMQMessageObjectMapper) {
+        RocketMQTemplate rocketMQTemplate = new RocketMQTemplate();
+        rocketMQTemplate.setProducer(mqProducer);
+        rocketMQTemplate.setObjectMapper(rocketMQMessageObjectMapper);
+        return rocketMQTemplate;
+    }
+
+    @Bean
+    @ConditionalOnBean(name = RocketMQConfigUtils.ROCKETMQ_TEMPLATE_DEFAULT_GLOBAL_NAME)
+    @ConditionalOnMissingBean(TransactionHandlerRegistry.class)
+    public TransactionHandlerRegistry transactionHandlerRegistry(@Qualifier(RocketMQConfigUtils.ROCKETMQ_TEMPLATE_DEFAULT_GLOBAL_NAME)
+                                                                         RocketMQTemplate template) {
+        return new TransactionHandlerRegistry(template);
+    }
+
+    @Bean(name = RocketMQConfigUtils.ROCKETMQ_TRANSACTION_ANNOTATION_PROCESSOR_BEAN_NAME)
+    @ConditionalOnBean(TransactionHandlerRegistry.class)
+    @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
+    public static RocketMQTransactionAnnotationProcessor transactionAnnotationProcessor(
+            TransactionHandlerRegistry transactionHandlerRegistry) {
+        return new RocketMQTransactionAnnotationProcessor(transactionHandlerRegistry);
+    }
+
+}

elab-rocketMQ/src/main/java/com/elab/mq/rocket/listener/DefaultRocketMQListener.java

@@ -91,15 +91,13 @@ public class DefaultRocketMQListener implements RocketMQListener<MessageExt> {
             saveData(entity, isInsert);
         } catch (Exception e) {
             logger.error(" 业务消息消费失败 topic : " + topic + " key : " + keys, e);
-            // 消息执行失败
-            e.printStackTrace();
             try {
                 saveExceptionData(body, entity, isInsert);
             } catch (Exception ex) {
                 logger.error(" 异常消费信息 存储数据库失败 ", ex);
                 ex.printStackTrace();
-
             }
+            throw new MsgProcessException(e.getMessage());
         }
     }
     private void saveData(RmqConsumerEntity entity, boolean isInsert) throws Exception {

elab-spring/pom.xml

@@ -31,6 +31,12 @@
             <version>${project.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-autoconfigure</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
         <dependency>
             <groupId>com.elab.log</groupId>
             <artifactId>elab-log</artifactId>
@@ -90,9 +96,9 @@
             <artifactId>spring-web</artifactId>
         </dependency>
         <!--<dependency>-->
-            <!--<groupId>com.elab.core</groupId>-->
-            <!--<artifactId>elab-db</artifactId>-->
-            <!--<version>${project.version}</version>-->
+        <!--<groupId>com.elab.core</groupId>-->
+        <!--<artifactId>elab-db</artifactId>-->
+        <!--<version>${project.version}</version>-->
         <!--</dependency>-->
         <dependency>
             <groupId>javax.validation</groupId>
@@ -138,10 +144,10 @@
         </dependency>
 
         <!--<dependency>-->
-            <!--<groupId>ch.qos.logback</groupId>-->
-            <!--<artifactId>logback-classic</artifactId>-->
-            <!--<version>1.2.3</version>-->
-            <!--<scope>provided</scope>-->
+        <!--<groupId>ch.qos.logback</groupId>-->
+        <!--<artifactId>logback-classic</artifactId>-->
+        <!--<version>1.2.3</version>-->
+        <!--<scope>provided</scope>-->
         <!--</dependency>-->
     </dependencies>
 

pom.xml

@@ -32,12 +32,17 @@
         <springboot.version>2.0.5.RELEASE</springboot.version>
         <mysql.connector.version>5.1.38</mysql.connector.version>
         <druid.version>1.1.6</druid.version>
-        <fastjson.version>1.2.47</fastjson.version>
+        <fastjson.version>1.2.60</fastjson.version>
         <swagger_version>2.8.0</swagger_version>
     </properties>
 
     <dependencyManagement>
         <dependencies>
+            <dependency>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-autoconfigure</artifactId>
+                <version>${springboot.version}</version>
+            </dependency>
             <dependency>
                 <groupId>com.elab.core</groupId>
                 <artifactId>elab-core</artifactId>
@@ -186,9 +191,9 @@
             -->
             <!-- log4 日志 mongodb数据库存储 -->
             <!--<dependency>-->
-                <!--<groupId>org.log4mongo</groupId>-->
-                <!--<artifactId>log4mongo-java</artifactId>-->
-                <!--<version>0.9.0</version>-->
+            <!--<groupId>org.log4mongo</groupId>-->
+            <!--<artifactId>log4mongo-java</artifactId>-->
+            <!--<version>0.9.0</version>-->
             <!--</dependency>-->
 
             <dependency>
@@ -346,10 +351,11 @@
                             <goal>jar</goal>
                         </goals>
                         <configuration>
-                            <additionalparam>-Xdoclint:none</additionalparam>
+                            <!--<additionalparam>-Xdoclint:none</additionalparam>-->
                             <charset>UTF-8</charset>
                             <encoding>UTF-8</encoding>
                             <docencoding>UTF-8</docencoding>
+                            <doclint>none</doclint>
                         </configuration>
                     </execution>
                 </executions>
@@ -361,9 +367,9 @@
     <!--mvn -DnewVersion=1.0.0-SNAPSHOT -DgenerateBackupPoms=false versions:set-->
     <distributionManagement>
         <!--<repository>-->
-            <!--<id>releases</id>-->
-            <!--<name>Nexus Release Repository</name>-->
-            <!--<url>http://192.168.0.11:8081/nexus/content/repositories/elab/</url>-->
+        <!--<id>releases</id>-->
+        <!--<name>Nexus Release Repository</name>-->
+        <!--<url>http://192.168.0.11:8081/nexus/content/repositories/elab/</url>-->
         <!--</repository>-->
         <snapshotRepository>
             <id>snapshots</id>