被动rocket mq 模板实现

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

@@ -0,0 +1,37 @@
+package com.docus.server.collect.basic.mq;
+
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+/**
+ * @author jiashi
+ * 梅州中医医院 MQ 监听
+ */
+@Component
+public class JmsIbmMqBasicConsumer {
+    @Resource(name = "basicJmsMqCollectServiceImpl")
+    private IBasicMqCollectService basicMqCollectService;
+
+    /**
+     * 订阅队列（注册科室）
+     *
+     * @param message 科室信息
+     */
+//    @JmsIbmListener(destination = "TJ_createDepartment")
+    public void addBasic(String message) {
+        basicMqCollectService.addTBasic(message);
+    }
+
+
+    /**
+     * 订阅队列（变更科室）
+     *
+     * @param message 科室信息
+     */
+//    @JmsIbmListener(destination = "TJ_updateDepartment")
+    public void updateBasic(String message) {
+        basicMqCollectService.updateTBasic(message);
+    }
+
+}

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

@@ -0,0 +1,52 @@
+package com.docus.server.collect.basic.mq;
+
+import com.docus.server.collect.IConverter;
+import com.docus.server.sys.common.pojo.dto.DeptDTO;
+import org.springframework.stereotype.Component;
+
+import java.util.UUID;
+
+@Component("mqBasicConverter")
+public class MqBasicConverter implements IConverter {
+
+    @Override
+    public DeptDTO convert(String message, String group) {
+        //TODO： 根据协议进行解析。
+//        XmlUtil xmlParseUtil = XmlUtil.of(message);
+//        Node msgIdNode = xmlParseUtil.getNode("/PRVS_IN000002UV01/id/@extension");
+//        Node receiverNode = xmlParseUtil.getNode("/PRVS_IN000002UV01/receiver/device/id/item/@extension");
+//        Node operateTypeNode = xmlParseUtil.getNode("/PRVS_IN000002UV01/controlActProcess/subject/registrationRequest/subject1/valueSet/valueSetItems/@operateType");
+//        Node deptCodeNode = xmlParseUtil.getNode("/PRVS_IN000002UV01/controlActProcess/subject/registrationRequest/subject1/valueSet/valueSetItems/DEPT_CODE/@value");
+//        Node deptNameNode = xmlParseUtil.getNode("/PRVS_IN000002UV01/controlActProcess/subject/registrationRequest/subject1/valueSet/valueSetItems/DEPT_NAME/@value");
+//        Node authorIdNode = xmlParseUtil.getNode("/PRVS_IN000002UV01/controlActProcess/subject/registrationRequest/author/assignedEntity/id/item/@extension");
+//        Node authorNameNode = xmlParseUtil.getNode("/PRVS_IN000002UV01/controlActProcess/subject/registrationRequest/author/assignedEntity/assignedPerson/name/item/part/@value");
+//
+//        DeptDTO deptDto = new DeptDTO();
+//        deptDto.setMessageId(msgIdNode.getNodeValue());
+//        deptDto.setReceiver(receiverNode.getNodeValue());
+//        deptDto.setOperateType(operateTypeNode.getNodeValue());
+//        deptDto.setDeptCode(deptCodeNode.getNodeValue());
+//        deptDto.setDeptName(deptNameNode.getNodeValue());
+//        deptDto.setAuthorId(authorIdNode.getNodeValue());
+//        deptDto.setAuthorName(authorNameNode.getNodeValue());
+//        deptDto.setSource(message);//原始xml报文
+//
+//        deptDto.setParams(ParamsUtils.addParam("msg", "操作成功！")
+//            .addParam("msgId", deptDto.getMessageId())
+//            .addParam("receiver", deptDto.getReceiver())
+//            .param());
+
+
+        DeptDTO deptDto = new DeptDTO();
+        deptDto.setMessageId(UUID.randomUUID().toString());
+        deptDto.setReceiver("1");
+        deptDto.setOperateType("2");
+        deptDto.setDeptCode("3");
+        deptDto.setDeptName("4");
+        deptDto.setAuthorId("5");
+        deptDto.setAuthorName("6");
+        deptDto.setSource(message);//原始xml报文
+
+        return deptDto;
+    }
+}

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

@@ -0,0 +1,20 @@
+package com.docus.server.collect.basic.mq;
+
+import com.docus.server.collect.IMqResult;
+import org.springframework.stereotype.Component;
+
+import java.util.Map;
+
+@Component("mqBasicResultImpl")
+public class MqBasicResultImpl implements IMqResult {
+
+    @Override
+    public Void ok(Map<String, Object> params) {
+        return null;
+    }
+
+    @Override
+    public Void fail(Map<String, Object> params) {
+        return null;
+    }
+}

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

@@ -0,0 +1,29 @@
+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;
+
+@Component
+@Slf4j
+@RocketMQMessageListener(topic = "ta-cipher-persist", consumerGroup = "ta-cipher-encode")
+public class RocketMqBasicConsumer implements RocketMQListener<String> {
+
+    @Resource(name = "basicRocketMqCollectServiceImpl")
+    private IBasicMqCollectService basicMqCollectService;
+
+    @Override
+    public void onMessage(String message) {
+        try {
+            log.info("RocketMQ message:{}", message);
+            basicMqCollectService.addTBasic(message);
+        } catch (Exception e) {
+            log.error("errorMessage:{}", message);
+            // 抛出异常会重新消费消息
+            throw new RuntimeException("Message processing failed", e);
+        }
+    }
+}

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

@@ -1,6 +1,5 @@
 package com.docus.server.collect.dept.mq;
 
-import com.docus.server.collect.user.mq.IUserMqCollectService;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.Resource;
@@ -10,11 +9,9 @@ import javax.annotation.Resource;
  * 梅州中医医院 MQ 监听
  */
 @Component
-public class JmsIbmListener {
-    @Resource
+public class JmsIbmMqDeptConsumer {
+    @Resource(name = "deptJmsMqCollectServiceImpl")
     private IDeptMqCollectService deptMqCollectService;
-    @Resource
-    private IUserMqCollectService userMqCollectService;
 
     /**
      * 订阅队列（注册科室）
@@ -37,27 +34,4 @@ public class JmsIbmListener {
         deptMqCollectService.updateDept(message);
     }
 
-
-    /**
-     * 订阅队列（注册人员）
-     *
-     * @param message 人员信息
-     */
-//    @JmsIbmListener(destination = "TJ_createPractitioner")
-    public void addUser(String message) {
-        userMqCollectService.addUser(message);
-    }
-
-
-    /**
-     * 订阅队列（变更人员）
-     *
-     * @param message 人员信息
-     */
-//    @JmsIbmListener(destination = "TJ_updatePractitioner")
-    public void updateUser(String message) {
-        userMqCollectService.updateUser(message);
-    }
-
-
 }

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

@@ -0,0 +1,29 @@
+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;
+
+@Component
+@Slf4j
+@RocketMQMessageListener(topic = "ta-cipher-persist", consumerGroup = "ta-cipher-encode")
+public class RocketMqDeptConsumer implements RocketMQListener<String> {
+
+    @Resource(name = "deptJmsMqCollectServiceImpl")
+    private IDeptMqCollectService deptMqCollectService;
+
+    @Override
+    public void onMessage(String message) {
+        try {
+            log.info("RocketMQ message:{}", message);
+            deptMqCollectService.addDept(message);
+        } catch (Exception e) {
+            log.error("errorMessage:{}", message);
+            // 抛出异常会重新消费消息
+            throw new RuntimeException("Message processing failed", e);
+        }
+    }
+}

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

@@ -0,0 +1,38 @@
+package com.docus.server.collect.user.mq;
+
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+/**
+ * @author jiashi
+ * 梅州中医医院 MQ 监听
+ */
+@Component
+public class JmsIbmMqUserConsumer {
+    @Resource(name = "userJmsMqCollectServiceImpl")
+    private IUserMqCollectService userMqCollectService;
+
+    /**
+     * 订阅队列（注册人员）
+     *
+     * @param message 人员信息
+     */
+//    @JmsIbmListener(destination = "TJ_createPractitioner")
+    public void addUser(String message) {
+        userMqCollectService.addUser(message);
+    }
+
+
+    /**
+     * 订阅队列（变更人员）
+     *
+     * @param message 人员信息
+     */
+//    @JmsIbmListener(destination = "TJ_updatePractitioner")
+    public void updateUser(String message) {
+        userMqCollectService.updateUser(message);
+    }
+
+
+}

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

@@ -0,0 +1,29 @@
+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;
+
+@Component
+@Slf4j
+@RocketMQMessageListener(topic = "ta-cipher-persist", consumerGroup = "ta-cipher-encode")
+public class RocketMqUserConsumer implements RocketMQListener<String> {
+
+    @Resource(name = "userRocketMqCollectServiceImpl")
+    private IUserMqCollectService userMqCollectService;
+
+    @Override
+    public void onMessage(String message) {
+        try {
+            log.info("RocketMQ message:{}", message);
+            userMqCollectService.addUser(message);
+        } catch (Exception e) {
+            log.error("errorMessage:{}", message);
+            // 抛出异常会重新消费消息
+            throw new RuntimeException("Message processing failed", e);
+        }
+    }
+}

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

@@ -81,3 +81,8 @@ xxl:
 #      queueManager: MqManager02
 #      channel: chl_server02
 #      connName: 10.222.23.131(1414)
+
+rocketmq:
+  name-server: 127.0.01:9876
+  producer:
+    group: ta-cipher-encode

common-collect/pom.xml

@@ -36,6 +36,5 @@
             <groupId>org.springframework.retry</groupId>
             <artifactId>spring-retry</artifactId>
         </dependency>
-
     </dependencies>
 </project>

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

@@ -0,0 +1,19 @@
+package com.docus.server.collect.basic.mq;
+
+public interface IBasicMqCollectService {
+
+    /**
+     * 订阅队列（基础数据）
+     *
+     * @param message 基础数据
+     */
+    void addTBasic(String message);
+
+    /**
+     * 订阅队列（变更基础数据）
+     *
+     * @param message 基础数据
+     */
+    void updateTBasic(String message);
+
+}

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

@@ -0,0 +1,47 @@
+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.record.pojo.dto.TBasicDTO;
+import com.docus.server.record.service.ITBasicService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+
+/**
+ * 每家mq 都不一样，这部分无法预先写好写入，需要根据每家独立编写。
+ */
+@Slf4j
+@Service("basicJmsMqCollectServiceImpl")
+public class BasicJmsMqCollectServiceImpl implements IBasicMqCollectService {
+    @Resource
+    private ITBasicService tBasicService;
+
+    @Override
+    @TrackGroup(
+        group = "JMS_MQ_PASSIVE_INSERT_BASIC",
+        beanNames = {"mqBasicConverter", "mqBasicResultImpl"},
+        processor = VisitorProcessor.class)
+    public void addTBasic(String message) {
+        log.info("新增基础数据：{}", message);
+        tBasicService.insertTBasic(getValue());
+    }
+
+    @Override
+    @TrackGroup(
+        group = "JMS_MQ_PASSIVE_UPDATE_BASIC",
+        beanNames = {"mqBasicConverter", "mqBasicResultImpl"},
+        processor = VisitorProcessor.class)
+    public void updateTBasic(String message) {
+        log.info("修改基础数据：{}", message);
+        tBasicService.updateTBasic(getValue());
+    }
+
+    public TBasicDTO getValue() {
+        return JSON.fromJSON((String) TrackHelper.getValue("jsonStr"), TBasicDTO.class);
+    }
+}

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

@@ -0,0 +1,54 @@
+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;
+
+@Service("basicRocketMqCollectServiceImpl")
+@Slf4j
+public class BasicRocketMqCollectServiceImpl implements IBasicMqCollectService {
+    @Resource
+    private CollectService collectService;
+
+    /**
+     * 订阅队列（基础数据）
+     *
+     * @param message 基础数据
+     */
+    @Override
+    @TrackGroup(
+        group = "ROCKET_MQ_PASSIVE_INSERT_DEPT",
+        beanNames = {"mqBasicConverter", "mqBasicResultImpl"},
+        processor = VisitorProcessor.class)
+    public void addTBasic(String message) {
+        log.info("注册基础数据 队列接收信息：{}", message);
+        collectService.insertOrUpdateDept(getValue());
+    }
+
+    /**
+     * 订阅队列（变更基础数据）
+     *
+     * @param message 基础数据
+     */
+    @Override
+    @TrackGroup(
+        group = "ROCKET_MQ_PASSIVE_UPDATE_DEPT",
+        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/DeptJmsMqCollectServiceImpl.java

@@ -4,9 +4,9 @@ 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 com.docus.server.collect.web.processor.VisitorProcessor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 
@@ -16,8 +16,8 @@ import javax.annotation.Resource;
  * 每家mq 都不一样，这部分无法预先写好写入，需要根据每家独立编写。
  */
 @Slf4j
-@Service
-public class DeptMqCollectServiceImpl implements IDeptMqCollectService {
+@Service("deptJmsMqCollectServiceImpl")
+public class DeptJmsMqCollectServiceImpl implements IDeptMqCollectService {
     @Resource
     private CollectService collectService;
 
@@ -28,7 +28,7 @@ public class DeptMqCollectServiceImpl implements IDeptMqCollectService {
      */
     @Override
     @TrackGroup(
-        group = "MQ_PASSIVE_INSERT_DEPT",
+        group = "JMS_MQ_PASSIVE_INSERT_DEPT",
         beanNames = {"mqDeptConverter", "mqDeptResultImpl"},
         processor = VisitorProcessor.class)
     public void addDept(String message) {
@@ -44,7 +44,7 @@ public class DeptMqCollectServiceImpl implements IDeptMqCollectService {
      */
     @Override
     @TrackGroup(
-        group = "MQ_PASSIVE_UPDATE_DEPT",
+        group = "JMS_MQ_PASSIVE_UPDATE_DEPT",
         beanNames = {"mqDeptConverter", "mqDeptResultImpl"},
         processor = VisitorProcessor.class)
     public void updateDept(String message) {

common-collect/src/main/java/com/docus/server/collect/dept/mq/impl/DeptRocketMqCollectServiceImpl.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("deptRocketMqCollectServiceImpl")
+@Slf4j
+public class DeptRocketMqCollectServiceImpl implements IDeptMqCollectService {
+    @Resource
+    private CollectService collectService;
+
+    /**
+     * 订阅队列（注册科室）
+     *
+     * @param message 科室信息
+     */
+    @Override
+    @TrackGroup(
+        group = "ROCKET_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 = "ROCKET_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,19 +6,22 @@ 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;
 import com.docus.server.sys.common.pojo.dto.UserDTO;
-import com.docus.server.collect.web.processor.VisitorProcessor;
 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;
 
 
 /**
@@ -37,6 +40,8 @@ import javax.annotation.Resource;
 public class UserRestController {
     @Resource
     private CollectService collectService;
+    @Inject
+    private RmqProvdier rmqProvdier;
 
     /**
      * 被动http 用户信息接收，进行操作
@@ -55,4 +60,12 @@ 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

@@ -0,0 +1,24 @@
+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/UserJmsMqCollectServiceImpl.java

@@ -4,9 +4,9 @@ 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 com.docus.server.collect.web.processor.VisitorProcessor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 
@@ -16,8 +16,8 @@ import javax.annotation.Resource;
  * 每家mq 都不一样，这部分无法预先写好写入，需要根据每家独立编写。
  */
 @Slf4j
-@Service
-public class UserMqCollectServiceImpl implements IUserMqCollectService {
+@Service("userJmsMqCollectServiceImpl")
+public class UserJmsMqCollectServiceImpl implements IUserMqCollectService {
 
     @Resource
     private CollectService collectService;
@@ -29,7 +29,7 @@ public class UserMqCollectServiceImpl implements IUserMqCollectService {
      */
     @Override
     @TrackGroup(
-        group = "MQ_PASSIVE_INSERT_DEPT",
+        group = "JMS_MQ_PASSIVE_INSERT_USER",
         beanNames = {"mqUserConverter", "mqUserResultImpl"},
         processor = VisitorProcessor.class)
     public void addUser(String message) {
@@ -44,7 +44,7 @@ public class UserMqCollectServiceImpl implements IUserMqCollectService {
      */
     @Override
     @TrackGroup(
-        group = "MQ_PASSIVE_UPDATE_USER",
+        group = "JMS_MQ_PASSIVE_UPDATE_USER",
         beanNames = {"mqUserConverter", "mqUserResultImpl"},
         processor = VisitorProcessor.class)
     public void updateUser(String message) {

common-collect/src/main/java/com/docus/server/collect/user/mq/impl/UserRocketMqCollectServiceImpl.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("userRocketMqCollectServiceImpl")
+@Slf4j
+public class UserRocketMqCollectServiceImpl implements IUserMqCollectService {
+    @Resource
+    private CollectService collectService;
+
+    /**
+     * 订阅队列（注册人员）
+     *
+     * @param message 人员信息
+     */
+    @Override
+    @TrackGroup(
+        group = "ROCKET_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 = "ROCKET_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

@@ -16,12 +16,19 @@ public enum CollectTypeEnum implements IIntegerEnum {
     HTTP_PASSIVE_DEPT(9, "HTTP_PASSIVE_DEPT", "新增/修改-科室信息"),
     HTTP_PASSIVE_BASIC(10, "HTTP_PASSIVE_BASIC", "新增/修改-基础数据"),
 
-    MQ_PASSIVE_INSERT_USER(11, "MQ_PASSIVE_USER", "新增-用户信息"),
-    MQ_PASSIVE_UPDATE_USER(12, "MQ_PASSIVE_USER", "修改-用户信息"),
-    MQ_PASSIVE_INSERT_DEPT(13, "MQ_PASSIVE_DEPT", "新增-科室信息"),
-    MQ_PASSIVE_UPDATE_DEPT(14, "MQ_PASSIVE_DEPT", "修改-科室信息"),
-    MQ_PASSIVE_INSERT_BASIC(15, "MQ_PASSIVE_BASIC", "新增-基础数据"),
-    MQ_PASSIVE_UPDATE_BASIC(16, "MQ_PASSIVE_BASIC", "修改-基础数据"),
+    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", "新增-科室信息"),
+    JMS_MQ_PASSIVE_UPDATE_DEPT(14, "JMS_MQ_PASSIVE_UPDATE_DEPT", "修改-科室信息"),
+    JMS_MQ_PASSIVE_INSERT_BASIC(15, "JMS_MQ_PASSIVE_INSERT_BASIC", "新增-基础数据"),
+    JMS_MQ_PASSIVE_UPDATE_BASIC(16, "JMS_MQ_PASSIVE_UPDATE_BASIC", "修改-基础数据"),
+
+    ROCKET_MQ_PASSIVE_INSERT_USER(17, "ROCKET_MQ_PASSIVE_INSERT_USER", "新增-用户信息"),
+    ROCKET_MQ_PASSIVE_UPDATE_USER(18, "ROCKET_MQ_PASSIVE_UPDATE_USER", "修改-用户信息"),
+    ROCKET_MQ_PASSIVE_INSERT_DEPT(19, "ROCKET_MQ_PASSIVE_INSERT_DEPT", "新增-科室信息"),
+    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", "修改-基础数据"),
     ;
 
     private Integer value;

pom.xml

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