被动rabbitmq模板封装

collect-sdry/src/main/java/com/docus/server/collect/basic/mq/RabbitMqBasicConsumer.java

@@ -0,0 +1,56 @@
+package com.docus.server.collect.basic.mq;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.rabbit.annotation.Exchange;
+import org.springframework.amqp.rabbit.annotation.Queue;
+import org.springframework.amqp.rabbit.annotation.QueueBinding;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+/**
+ * 基于注解的绑定交换器、队列、路由键设置
+ * 1. Queue配置：value=队列名称、durable=是否持久化(默认true)、exclusive=排他队列只在当前connection可用(默认false)、autoDelete=如无消息是否自动删除(默认false)
+ * 2. Exchange配置：value=交换器名称、type=类型(默认direct)、durable=是否持久化(默认true)、autoDelete=如无消息是否自动删除(默认false)
+ * 3. QueueBinding配置：key=路由键(string数组，支持* # 匹配)，*必须匹配一个单词，#匹配0个或N个单词，用.分隔
+ * 4. RabbitListener配置： bindings=Queue配置+Exchange配置+QueueBinding配置
+ * 注：如果代码创建交换器等且配置绑定关系，注解只需监听队列即可，如：@RabbitListener(queues = "direct.queue")
+ * <p>
+ * <p>
+ * rabbitmq basic consumer
+ *
+ * @author linruifeng
+ * @see com.docus.server.collect.basic.mq.impl.BasicRabbitMqCollectServiceImpl#addTBasic(String)
+ * @see com.docus.server.collect.user.mq.test.RabbitMQProducer#topicMQ03()
+ */
+@Component
+@Slf4j
+public class RabbitMqBasicConsumer {
+    @Resource(name = "basicRabbitMqCollectServiceImpl")
+    private IBasicMqCollectService basicMqCollectService;
+
+    /**
+     * topic n:1 类型 交换器队列 消费(普通会员注册提醒)
+     */
+    @RabbitListener(
+        bindings = {@QueueBinding(value = @Queue(value = "topic.queue.01"),
+            exchange = @Exchange(value = "topic.exchange", type = "topic"),
+            key = {"*.reg.msg"})})
+    public void getTopicMessage01(String message) {
+        log.info("rabbitmq message:{}", message);
+        basicMqCollectService.addTBasic(message);
+    }
+
+    /**
+     * topic n:1 类型 交换器队列 消费(超级会员注册提醒)
+     */
+    @RabbitListener(bindings = {
+        @QueueBinding(value = @Queue(value = "topic.queue.02"),
+            exchange = @Exchange(value = "topic.exchange", type = "topic"),
+            key = {"*.*.reg.msg.#"})})
+    public void getTopicMessage02(String message) {
+        log.info("rabbitmq message:{}", message);
+        basicMqCollectService.updateTBasic(message);
+    }
+}

collect-sdry/src/main/java/com/docus/server/collect/basic/mq/RocketMqBasicConsumer.java

@@ -1,15 +1,21 @@
 package com.docus.server.collect.basic.mq;
 
 import lombok.extern.slf4j.Slf4j;
-import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
 import org.apache.rocketmq.spring.core.RocketMQListener;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.Resource;
 
+/**
+ * rocketmq 基础数据消费者
+ *
+ * @see com.docus.server.collect.user.mq.test.RocketMQProvdier
+ * @see com.docus.server.collect.basic.mq.impl.BasicRocketMqCollectServiceImpl#addTBasic(String)
+ */
 @Component
 @Slf4j
-@RocketMQMessageListener(topic = "ta-cipher-persist", consumerGroup = "ta-cipher-encode")
+//启用rockertmq consumer，打开下面的注解监听
+//@RocketMQMessageListener(topic = "ta-cipher-persist", consumerGroup = "ta-cipher-encode")
 public class RocketMqBasicConsumer implements RocketMQListener<String> {
 
     @Resource(name = "basicRocketMqCollectServiceImpl")

collect-sdry/src/main/java/com/docus/server/collect/dept/mq/RabbitMqDeptConsumer.java

@@ -0,0 +1,55 @@
+package com.docus.server.collect.dept.mq;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.rabbit.annotation.Exchange;
+import org.springframework.amqp.rabbit.annotation.Queue;
+import org.springframework.amqp.rabbit.annotation.QueueBinding;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+/**
+ * 基于注解的绑定交换器、队列、路由键设置
+ * 1. Queue配置：value=队列名称、durable=是否持久化(默认true)、exclusive=排他队列只在当前connection可用(默认false)、autoDelete=如无消息是否自动删除(默认false)
+ * 2. Exchange配置：value=交换器名称、type=类型(默认direct)、durable=是否持久化(默认true)、autoDelete=如无消息是否自动删除(默认false)
+ * 3. QueueBinding配置：key=路由键(string数组，支持* # 匹配)，*必须匹配一个单词，#匹配0个或N个单词，用.分隔
+ * 4. RabbitListener配置： bindings=Queue配置+Exchange配置+QueueBinding配置
+ * 注：如果代码创建交换器等且配置绑定关系，注解只需监听队列即可，如：@RabbitListener(queues = "direct.queue")
+ * <p>
+ * rabbitmq 被动接收用户信息
+ *
+ * @author linruifeng
+ * @see com.docus.server.collect.user.mq.impl.UserRabbitMqCollectServiceImpl#addUser(String)
+ * @see com.docus.server.collect.user.mq.test.RabbitMQProducer#topicMQ03()
+ */
+@Component
+@Slf4j
+public class RabbitMqDeptConsumer {
+    @Resource(name = "deptRabbitMqCollectServiceImpl")
+    private IDeptMqCollectService deptMqCollectService;
+
+    /**
+     * topic n:1 类型 交换器队列 消费(普通会员注册提醒)
+     */
+    @RabbitListener(
+        bindings = {@QueueBinding(value = @Queue(value = "topic.queue.01"),
+            exchange = @Exchange(value = "topic.exchange", type = "topic"),
+            key = {"*.reg.msg"})})
+    public void getTopicMessage01(String message) {
+        log.info("rabbitmq message:{}", message);
+        deptMqCollectService.addDept(message);
+    }
+
+    /**
+     * topic n:1 类型 交换器队列 消费(超级会员注册提醒)
+     */
+    @RabbitListener(bindings = {
+        @QueueBinding(value = @Queue(value = "topic.queue.02"),
+            exchange = @Exchange(value = "topic.exchange", type = "topic"),
+            key = {"*.*.reg.msg.#"})})
+    public void getTopicMessage02(String message) {
+        log.info("rabbitmq message:{}", message);
+        deptMqCollectService.updateDept(message);
+    }
+}

collect-sdry/src/main/java/com/docus/server/collect/dept/mq/RocketMqDeptConsumer.java

@@ -1,15 +1,21 @@
 package com.docus.server.collect.dept.mq;
 
 import lombok.extern.slf4j.Slf4j;
-import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
 import org.apache.rocketmq.spring.core.RocketMQListener;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.Resource;
 
+/**
+ * rocketmq 科室数据消费者
+ *
+ * @see com.docus.server.collect.user.mq.RocketMqUserConsumer#send(String)
+ * @see com.docus.server.collect.user.mq.test.MQTestRestController
+ */
 @Component
 @Slf4j
-@RocketMQMessageListener(topic = "ta-cipher-persist", consumerGroup = "ta-cipher-encode")
+//启用rockertmq consumer，打开下面的注解监听
+//@RocketMQMessageListener(topic = "ta-cipher-persist", consumerGroup = "ta-cipher-encode")
 public class RocketMqDeptConsumer implements RocketMQListener<String> {
 
     @Resource(name = "deptJmsMqCollectServiceImpl")

collect-sdry/src/main/java/com/docus/server/collect/user/mq/RabbitMqUserConsumer.java

@@ -0,0 +1,55 @@
+package com.docus.server.collect.user.mq;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.rabbit.annotation.Exchange;
+import org.springframework.amqp.rabbit.annotation.Queue;
+import org.springframework.amqp.rabbit.annotation.QueueBinding;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+/**
+ * 基于注解的绑定交换器、队列、路由键设置
+ * 1. Queue配置：value=队列名称、durable=是否持久化(默认true)、exclusive=排他队列只在当前connection可用(默认false)、autoDelete=如无消息是否自动删除(默认false)
+ * 2. Exchange配置：value=交换器名称、type=类型(默认direct)、durable=是否持久化(默认true)、autoDelete=如无消息是否自动删除(默认false)
+ * 3. QueueBinding配置：key=路由键(string数组，支持* # 匹配)，*必须匹配一个单词，#匹配0个或N个单词，用.分隔
+ * 4. RabbitListener配置： bindings=Queue配置+Exchange配置+QueueBinding配置
+ * 注：如果代码创建交换器等且配置绑定关系，注解只需监听队列即可，如：@RabbitListener(queues = "direct.queue")
+ * <p>
+ * rabbitmq 被动接收用户信息
+ *
+ * @author linruifeng
+ * @see com.docus.server.collect.user.mq.impl.UserRabbitMqCollectServiceImpl#addUser(String)
+ * @see com.docus.server.collect.user.mq.test.RabbitMQProducer#topicMQ03()
+ */
+@Component
+@Slf4j
+public class RabbitMqUserConsumer {
+    @Resource(name = "userRabbitMqCollectServiceImpl")
+    private IUserMqCollectService userMqCollectService;
+
+    /**
+     * topic n:1 类型 交换器队列 消费(普通会员注册提醒)
+     */
+    @RabbitListener(
+        bindings = {@QueueBinding(value = @Queue(value = "topic.queue.01"),
+            exchange = @Exchange(value = "topic.exchange", type = "topic"),
+            key = {"*.reg.msg"})})
+    public void getTopicMessage01(String message) {
+        log.info("rabbitmq message:{}", message);
+        userMqCollectService.addUser(message);
+    }
+
+    /**
+     * topic n:1 类型 交换器队列 消费(超级会员注册提醒)
+     */
+    @RabbitListener(bindings = {
+        @QueueBinding(value = @Queue(value = "topic.queue.02"),
+            exchange = @Exchange(value = "topic.exchange", type = "topic"),
+            key = {"*.*.reg.msg.#"})})
+    public void getTopicMessage02(String message) {
+        log.info("rabbitmq message:{}", message);
+        userMqCollectService.updateUser(message);
+    }
+}

collect-sdry/src/main/java/com/docus/server/collect/user/mq/RocketMqUserConsumer.java

@@ -1,15 +1,18 @@
 package com.docus.server.collect.user.mq;
 
 import lombok.extern.slf4j.Slf4j;
-import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
 import org.apache.rocketmq.spring.core.RocketMQListener;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.Resource;
 
+/**
+ * rocketmq 被动接收用户信息
+ */
 @Component
 @Slf4j
-@RocketMQMessageListener(topic = "ta-cipher-persist", consumerGroup = "ta-cipher-encode")
+//启用rockertmq consumer，打开下面的注解监听
+//@RocketMQMessageListener(topic = "ta-cipher-persist", consumerGroup = "ta-cipher-encode")
 public class RocketMqUserConsumer implements RocketMQListener<String> {
 
     @Resource(name = "userRocketMqCollectServiceImpl")
@@ -18,7 +21,7 @@ public class RocketMqUserConsumer implements RocketMQListener<String> {
     @Override
     public void onMessage(String message) {
         try {
-            log.info("RocketMQ message:{}", message);
+            log.info("rocketmq message:{}", message);
             userMqCollectService.addUser(message);
         } catch (Exception e) {
             log.error("errorMessage:{}", message);

collect-sdry/src/main/java/com/docus/server/collect/user/mq/test/MQTestRestController.java

@@ -0,0 +1,37 @@
+package com.docus.server.collect.user.mq.test;
+
+import com.docus.server.collect.IConverter;
+import com.docus.server.collect.IHttpResult;
+import com.docus.server.collect.web.enums.CollectTypeEnum;
+import com.docus.server.collect.web.processor.VisitorProcessor;
+import io.swagger.annotations.Api;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+
+/**
+ * @author wen yongbin
+ * @date 2023年2月25日21:56:33
+ * @see CollectTypeEnum 枚举
+ * @see IConverter 通用转化器
+ * @see IHttpResult 通用返回结果
+ * @see VisitorProcessor 通用处理器
+ */
+
+@Api(value = "用户管理接口", tags = "用户管理接口")
+@Slf4j
+@RestController
+@RequestMapping("/api/rest/user")
+public class MQTestRestController {
+//    @Inject
+//    private RmqProvdier rmqProvdier;
+
+//    /**
+//     * 测试rocket mq
+//     */
+//    @GetMapping("/test")
+//    public void test() {
+//        rmqProvdier.send("hello world");
+//    }
+}

collect-sdry/src/main/java/com/docus/server/collect/user/mq/test/RabbitMQProducer.java

@@ -0,0 +1,60 @@
+package com.docus.server.collect.user.mq.test;
+
+import com.docus.server.collect.web.domain.TaskConfig;
+import com.docus.server.collect.web.service.impl.TaskConfigServiceImpl;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestMethod;
+import org.springframework.web.bind.annotation.RestController;
+
+import javax.inject.Inject;
+import java.util.List;
+
+/**
+ * @Description 生产者
+ * @Author jxb
+ * @Date 2019-03-09 09:43:47
+ */
+@RestController
+@RequestMapping("mqProducer")
+public class RabbitMQProducer {
+
+    @Autowired
+    public RabbitTemplate rabbitTemplate;
+    @Inject
+    private TaskConfigServiceImpl userService;
+
+    /**
+     * @Description topic n:1 类型 交换器队列 生产(3个)
+     * @Author jxb
+     * @Date 2019-03-09 09:56:45
+     */
+    @RequestMapping(value = "/topicMQ01", method = {RequestMethod.GET})
+    public List<TaskConfig> topicMQ01() {
+        List<TaskConfig> users = userService.list(null);
+        for (TaskConfig user : users) {
+            rabbitTemplate.convertAndSend("topic.exchange", "jd.reg.msg", user.getName());
+        }
+        return users;
+    }
+
+    @RequestMapping(value = "/topicMQ02", method = {RequestMethod.GET})
+    public List<TaskConfig> topicMQ02() {
+        List<TaskConfig> users = userService.list(null);
+        for (TaskConfig user : users) {
+            rabbitTemplate.convertAndSend("topic.exchange", "tm.reg.msg", user.getName());
+        }
+        return users;
+    }
+
+    @RequestMapping(value = "/topicMQ03", method = {RequestMethod.GET})
+    public List<TaskConfig> topicMQ03() {
+        List<TaskConfig> users = userService.list(null);
+        for (TaskConfig user : users) {
+            rabbitTemplate.convertAndSend("topic.exchange", "super.fzb.reg.msg", user.getName());
+        }
+        return users;
+    }
+
+}

collect-sdry/src/main/java/com/docus/server/collect/user/mq/test/RocketMQProvdier.java

@@ -0,0 +1,28 @@
+package com.docus.server.collect.user.mq.test;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+//
+//import lombok.extern.slf4j.Slf4j;
+//import org.apache.rocketmq.spring.core.RocketMQTemplate;
+//import org.springframework.beans.factory.annotation.Autowired;
+//import org.springframework.stereotype.Component;
+//
+@Component
+@Slf4j
+public class RocketMQProvdier {
+//    @Autowired
+//    private RocketMQTemplate rocketMQTemplate;
+//
+//    public boolean send(String message) {
+//        try {
+//            // 发送消息
+//            rocketMQTemplate.convertAndSend("ta-cipher-persist", message);
+//        } catch (Exception e) {
+//            log.error("send message:{}", message, e);
+//            return false;
+//        }
+//        return true;
+//    }
+}

collect-sdry/src/main/resources/bootstrap.yml

@@ -12,21 +12,24 @@ spring:
       datasource:
         master:
           url: jdbc:log4jdbc:mysql://db.docus.cn:3306/docus_archivefile?autoReconnect=true&allowMultiQueries=true&useSSL=false&useUnicode=true&characterEncoding=utf-8&serverTimezone=Asia/Shanghai
-          username: docus
+          username: root
-          password: docus702
+          password: root
           driver-class-name: net.sf.log4jdbc.sql.jdbcapi.DriverSpy
           type: com.alibaba.druid.pool.DruidDataSource
         his:
-          url: jdbc:log4jdbc:mysql://db.docus.cn:3306/his?autoReconnect=true&allowMultiQueries=true&useSSL=false&useUnicode=true&characterEncoding=utf-8&serverTimezone=Asia/Shanghai
+          url: jdbc:log4jdbc:mysql://127.0.0.1:3306/his?autoReconnect=true&allowMultiQueries=true&useSSL=false&useUnicode=true&characterEncoding=utf-8&serverTimezone=Asia/Shanghai
-          username: docus
+          username: root
-          password: docus702
+          password: root
           driver-class-name: net.sf.log4jdbc.sql.jdbcapi.DriverSpy
           type: com.alibaba.druid.pool.DruidDataSource
-
+  #  rabbitmq:
-
+  #    host: 127.0.0.1
+  #    port: 5672
+  #    username: guest
+  #    password: guest
   redis:
     host: redis.docus.cn
-    password: JSdocus@702
+  #    password: JSdocus@702
   cloud:
     nacos:
       discovery:
@@ -41,20 +44,21 @@ spring:
 
 docus:
   dbtype: mysql
-  user:
-    # 用户默认密码
-    defpwd: fd29cd53ec12616e5f36b77d4afffbff
 
 mybatis-plus:
   configuration:
     map-underscore-to-camel-case: true
     call-setters-on-nulls: true
+    jdbc-type-for-null: null
     log-impl: org.apache.ibatis.logging.stdout.StdOutImpl
   global-config:
     db-config:
+      update-strategy: ignored
       field-strategy: NOT_EMPTY
       db-type: MYSQL
+  mapper-locations: classpath*:/mapper/*Mapper.xml
   type-enums-package: com.docus.server.collect.web.enums
+
 xxl:
   job:
     accessToken:
@@ -68,21 +72,16 @@ xxl:
       logretentiondays: 30
       logpath: D:/xxl-job/inspection
 
-#
+api:
-#ibm:
+  soft-delete-column-name=state
-#  mq:
+  soft-delete-column-value=1
-#    queueManager: MqManager01
+logging:
-#    channel: chl_server01
+  level:
-#    connName: 10.222.23.130(1414)
+    com:
-#    ackUrl: 10.222.23.159:7800/ack
+      baomidou:
-#    domain: CDR
+        mybatisplus: DEBUG
-#    key: cdr-0012-adef
-#    replica:
-#      queueManager: MqManager02
-#      channel: chl_server02
-#      connName: 10.222.23.131(1414)
 
-rocketmq:
+#rocketmq:
-  name-server: 127.0.01:9876
+#  name-server: 127.0.01:9876
-  producer:
+#  producer:
-    group: ta-cipher-encode
+#    group: ta-cipher-encode

common-collect/src/main/java/com/docus/server/collect/basic/mq/IBasicMqCollectService.java

@@ -1,5 +1,8 @@
 package com.docus.server.collect.basic.mq;
 
+/**
+ * @author linruifeng
+ */
 public interface IBasicMqCollectService {
 
     /**

common-collect/src/main/java/com/docus/server/collect/basic/mq/impl/BasicRabbitMqCollectServiceImpl.java

@@ -0,0 +1,57 @@
+package com.docus.server.collect.basic.mq.impl;
+
+import com.docus.core.util.json.JSON;
+import com.docus.log.annotation.TrackGroup;
+import com.docus.log.context.TrackHelper;
+import com.docus.server.collect.basic.mq.IBasicMqCollectService;
+import com.docus.server.collect.web.processor.VisitorProcessor;
+import com.docus.server.collect.web.service.CollectService;
+import com.docus.server.sys.common.pojo.dto.DeptDTO;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+
+/**
+ * @author linruifeng
+ */
+@Service("basicRabbitMqCollectServiceImpl")
+@Slf4j
+public class BasicRabbitMqCollectServiceImpl implements IBasicMqCollectService {
+    @Resource
+    private CollectService collectService;
+
+    /**
+     * 订阅队列（基础数据）
+     *
+     * @param message 基础数据
+     */
+    @Override
+    @TrackGroup(
+        group = "RABBIT_MQ_PASSIVE_INSERT_BASIC",
+        beanNames = {"mqBasicConverter", "mqBasicResultImpl"},
+        processor = VisitorProcessor.class)
+    public void addTBasic(String message) {
+        log.info("注册基础数据 队列接收信息：{}", message);
+        collectService.insertOrUpdateDept(getValue());
+    }
+
+    /**
+     * 订阅队列（变更基础数据）
+     *
+     * @param message 基础数据
+     */
+    @Override
+    @TrackGroup(
+        group = "RABBIT_MQ_PASSIVE_UPDATE_BASIC",
+        beanNames = {"mqBasicConverter", "mqBasicResultImpl"},
+        processor = VisitorProcessor.class)
+    public void updateTBasic(String message) {
+        log.info("变更基础数据 队列接收信息：{}", message);
+        collectService.insertOrUpdateDept(getValue());
+    }
+
+    public DeptDTO getValue() {
+        return JSON.fromJSON((String) TrackHelper.getValue("jsonStr"), DeptDTO.class);
+    }
+}

common-collect/src/main/java/com/docus/server/collect/dept/mq/impl/DeptRabbitMqCollectServiceImpl.java

@@ -0,0 +1,55 @@
+package com.docus.server.collect.dept.mq.impl;
+
+import com.docus.core.util.json.JSON;
+import com.docus.log.annotation.TrackGroup;
+import com.docus.log.context.TrackHelper;
+import com.docus.server.collect.dept.mq.IDeptMqCollectService;
+import com.docus.server.collect.web.processor.VisitorProcessor;
+import com.docus.server.collect.web.service.CollectService;
+import com.docus.server.sys.common.pojo.dto.DeptDTO;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+
+@Service("deptRabbitMqCollectServiceImpl")
+@Slf4j
+public class DeptRabbitMqCollectServiceImpl implements IDeptMqCollectService {
+    @Resource
+    private CollectService collectService;
+
+    /**
+     * 订阅队列（注册科室）
+     *
+     * @param message 科室信息
+     */
+    @Override
+    @TrackGroup(
+        group = "RABBIT_MQ_PASSIVE_INSERT_DEPT",
+        beanNames = {"mqDeptConverter", "mqDeptResultImpl"},
+        processor = VisitorProcessor.class)
+    public void addDept(String message) {
+        log.info("注册科室 队列接收信息：{}", message);
+        collectService.insertOrUpdateDept(getValue());
+    }
+
+
+    /**
+     * 订阅队列（变更科室）
+     *
+     * @param message 科室信息
+     */
+    @Override
+    @TrackGroup(
+        group = "RABBIT_MQ_PASSIVE_UPDATE_DEPT",
+        beanNames = {"mqDeptConverter", "mqDeptResultImpl"},
+        processor = VisitorProcessor.class)
+    public void updateDept(String message) {
+        log.info("变更科室 队列接收信息：{}", message);
+        collectService.insertOrUpdateDept(getValue());
+    }
+
+    public DeptDTO getValue() {
+        return JSON.fromJSON((String) TrackHelper.getValue("jsonStr"), DeptDTO.class);
+    }
+}

common-collect/src/main/java/com/docus/server/collect/user/http/UserRestController.java

@@ -6,7 +6,6 @@ import com.docus.log.annotation.TrackGroup;
 import com.docus.log.context.TrackHelper;
 import com.docus.server.collect.IConverter;
 import com.docus.server.collect.IHttpResult;
-import com.docus.server.collect.user.mq.RmqProvdier;
 import com.docus.server.collect.web.enums.CollectTypeEnum;
 import com.docus.server.collect.web.processor.VisitorProcessor;
 import com.docus.server.collect.web.service.CollectService;
@@ -14,14 +13,12 @@ import com.docus.server.sys.common.pojo.dto.UserDTO;
 import io.swagger.annotations.Api;
 import io.swagger.annotations.ApiOperation;
 import lombok.extern.slf4j.Slf4j;
-import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
 
 import javax.annotation.Resource;
-import javax.inject.Inject;
 
 
 /**
@@ -40,8 +37,6 @@ import javax.inject.Inject;
 public class UserRestController {
     @Resource
     private CollectService collectService;
-    @Inject
-    private RmqProvdier rmqProvdier;
 
     /**
      * 被动http 用户信息接收，进行操作
@@ -60,12 +55,4 @@ public class UserRestController {
         collectService.insertOrUpdateUser(JSON.fromJSON((String) TrackHelper.getValue("jsonStr"), UserDTO.class));
         return null;
     }
-
-    /**
-     * 测试rocket mq
-     */
-    @GetMapping("/test")
-    public void test() {
-        rmqProvdier.send("hello world");
-    }
 }

common-collect/src/main/java/com/docus/server/collect/user/mq/RmqProvdier.java

@@ -1,24 +0,0 @@
-package com.docus.server.collect.user.mq;
-
-import lombok.extern.slf4j.Slf4j;
-import org.apache.rocketmq.spring.core.RocketMQTemplate;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Component;
-
-@Component
-@Slf4j
-public class RmqProvdier {
-    @Autowired
-    private RocketMQTemplate rocketMQTemplate;
-
-    public boolean send(String message) {
-        try {
-            // 发送消息
-            rocketMQTemplate.convertAndSend("ta-cipher-persist", message);
-        } catch (Exception e) {
-            log.error("send message:{}", message, e);
-            return false;
-        }
-        return true;
-    }
-}

common-collect/src/main/java/com/docus/server/collect/user/mq/impl/UserRabbitMqCollectServiceImpl.java

@@ -0,0 +1,54 @@
+package com.docus.server.collect.user.mq.impl;
+
+import com.docus.core.util.json.JSON;
+import com.docus.log.annotation.TrackGroup;
+import com.docus.log.context.TrackHelper;
+import com.docus.server.collect.user.mq.IUserMqCollectService;
+import com.docus.server.collect.web.processor.VisitorProcessor;
+import com.docus.server.collect.web.service.CollectService;
+import com.docus.server.sys.common.pojo.dto.UserDTO;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+
+@Service("userRabbitMqCollectServiceImpl")
+@Slf4j
+public class UserRabbitMqCollectServiceImpl implements IUserMqCollectService {
+    @Resource
+    private CollectService collectService;
+
+    /**
+     * 订阅队列（注册人员）
+     *
+     * @param message 人员信息
+     */
+    @Override
+    @TrackGroup(
+        group = "RABBIT_MQ_PASSIVE_INSERT_USER",
+        beanNames = {"mqUserConverter", "mqUserResultImpl"},
+        processor = VisitorProcessor.class)
+    public void addUser(String message) {
+        log.info("人员注册 队列接收信息：{}", message);
+        collectService.insertOrUpdateUser(getValue());
+    }
+
+    /**
+     * 订阅队列（变更人员）
+     *
+     * @param message 人员信息
+     */
+    @Override
+    @TrackGroup(
+        group = "RABBIT_MQ_PASSIVE_UPDATE_USER",
+        beanNames = {"mqUserConverter", "mqUserResultImpl"},
+        processor = VisitorProcessor.class)
+    public void updateUser(String message) {
+        log.info("人员变更 队列接收信息：{}", message);
+        collectService.insertOrUpdateUser(getValue());
+    }
+
+    public UserDTO getValue() {
+        return JSON.fromJSON((String) TrackHelper.getValue("jsonStr"), UserDTO.class);
+    }
+}

common-collect/src/main/java/com/docus/server/collect/web/enums/CollectTypeEnum.java

@@ -2,6 +2,9 @@ package com.docus.server.collect.web.enums;
 
 public enum CollectTypeEnum implements IIntegerEnum {
 
+    /**
+     * WS
+     */
     WEBSERVICE_XML_DEPT(0, "WEBSERVICE_XML_DEPT", "新增/修改-科室信息"),
     WEBSERVICE_XML_USER(1, "WEBSERVICE_XML_USER", "新增/修改-用户信息"),
     WEBSERVICE_XML_INSERT_BASIC(2, "WEBSERVICE_XML_BASIC", "新增/修改-基础数据"),
@@ -12,10 +15,16 @@ public enum CollectTypeEnum implements IIntegerEnum {
     WEBSERVICE_XML_INSERT_INSPECTION_REPORT(6, "WEBSERVICE_XML_INSERT_INSPECTION_REPORT", "新增-检查报告的信息"),
     WEBSERVICE_XML_UPDATE_INSPECTION_REPORT(7, "WEBSERVICE_XML_UPDATE_INSPECTION_REPORT", "更新-检查报告的信息"),
 
+    /**
+     * HTTP
+     */
     HTTP_PASSIVE_USER(8, "HTTP_PASSIVE_USER", "新增/修改-用户信息"),
     HTTP_PASSIVE_DEPT(9, "HTTP_PASSIVE_DEPT", "新增/修改-科室信息"),
     HTTP_PASSIVE_BASIC(10, "HTTP_PASSIVE_BASIC", "新增/修改-基础数据"),
 
+    /**
+     * MQ
+     */
     JMS_MQ_PASSIVE_INSERT_USER(11, "JMS_MQ_PASSIVE_INSERT_USER", "新增-用户信息"),
     JMS_MQ_PASSIVE_UPDATE_USER(12, "JMS_MQ_PASSIVE_UPDATE_USER", "修改-用户信息"),
     JMS_MQ_PASSIVE_INSERT_DEPT(13, "JMS_MQ_PASSIVE_INSERT_DEPT", "新增-科室信息"),
@@ -29,6 +38,13 @@ public enum CollectTypeEnum implements IIntegerEnum {
     ROCKET_MQ_PASSIVE_UPDATE_DEPT(20, "ROCKET_MQ_PASSIVE_UPDATE_DEPT", "修改-科室信息"),
     ROCKET_MQ_PASSIVE_INSERT_BASIC(21, "ROCKET_MQ_PASSIVE_INSERT_BASIC", "新增-基础数据"),
     ROCKET_MQ_PASSIVE_UPDATE_BASIC(22, "ROCKET_MQ_PASSIVE_UPDATE_BASIC", "修改-基础数据"),
+
+    RABBIT_MQ_PASSIVE_INSERT_USER(23, "RABBIT_MQ_PASSIVE_INSERT_USER", "新增-用户信息"),
+    RABBIT_MQ_PASSIVE_UPDATE_USER(24, "RABBIT_MQ_PASSIVE_UPDATE_USER", "修改-用户信息"),
+    RABBIT_MQ_PASSIVE_INSERT_DEPT(25, "RABBIT_MQ_PASSIVE_INSERT_DEPT", "新增-科室信息"),
+    RABBIT_MQ_PASSIVE_UPDATE_DEPT(26, "RABBIT_MQ_PASSIVE_UPDATE_DEPT", "修改-科室信息"),
+    RABBIT_MQ_PASSIVE_INSERT_BASIC(27, "RABBIT_MQ_PASSIVE_INSERT_BASIC", "新增-基础数据"),
+    RABBIT_MQ_PASSIVE_UPDATE_BASIC(28, "RABBIT_MQ_PASSIVE_UPDATE_BASIC", "修改-基础数据"),
     ;
 
     private Integer value;

common-collect/src/main/java/com/docus/server/collect/web/processor/VisitorProcessor.java

@@ -41,7 +41,7 @@ public class VisitorProcessor extends AbstractProcessor {
         Map<String, Object> params = context.getParams();
         params.put("taskId", taskId);
         params.put("jsonStr", jsonStr);
-        params.put("methodName", context.getMethodName());
+        params.put("group", context.getGroup());
         return null;
     }
 
@@ -67,6 +67,7 @@ public class VisitorProcessor extends AbstractProcessor {
                 return result.ok(params);
             } else {
                 log.info("=== AOP 异常通知 ===");
+                log.error((String) params.get(context.getExMessageResult()));
                 messageService.updateTaskOriginalMessage(taskId, afterReturnResult, StateEnum.FAIL);
                 return result.fail(params);
             }

pom.xml

@@ -209,12 +209,17 @@
             <artifactId>netty-all</artifactId>
             <version>4.1.87.Final</version>
         </dependency>
-
+        <!--rocketmq-->
         <dependency>
             <groupId>org.apache.rocketmq</groupId>
             <artifactId>rocketmq-spring-boot-starter</artifactId>
             <version>2.2.3</version>
         </dependency>
+        <!--rabbitmq-->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-amqp</artifactId>
+        </dependency>
     </dependencies>
 
     <profiles>