opt: 任务发送消费者topic定义

nl-module-lms/nl-module-lms-server/pom.xml

@@ -32,7 +32,11 @@
             <artifactId>nl-module-lms-api</artifactId>
             <version>${revision}</version>
         </dependency>
-
+        <dependency>
+            <groupId>cn.nl.cloud</groupId>
+            <artifactId>nl-module-task-api</artifactId>
+            <version>${revision}</version>
+        </dependency>
         <!-- 业务组件 -->
         <dependency>
             <groupId>cn.nl.cloud</groupId>

nl-module-lms/nl-module-lms-server/src/main/java/cn/code/nl/module/lms/demo/DemoApi.java

@@ -1,80 +0,0 @@
-//package cn.code.nl.module.lms.demo;
-//
-//import cn.code.nl.framework.common.pojo.CommonResult;
-//import cn.code.nl.framework.execute.biz.api.lms.LmsTaskCommonApi;
-//import cn.code.nl.framework.execute.biz.dto.TaskStatusCallApiReqDTO;
-//import cn.code.nl.framework.execute.biz.vo.AcsApplyActionRespVO;
-//import com.alibaba.fastjson.JSON;
-//import org.springframework.validation.annotation.Validated;
-//import org.springframework.web.bind.annotation.RestController;
-//
-//import static cn.code.nl.framework.common.pojo.CommonResult.success;
-//
-///**
-// *
-// * @Author: liyongde
-// * @Date: 2026/7/15 15:25
-// */
-//@RestController // 提供 RESTful API 接口，给 Feign 调用
-//@Validated
-//public class DemoApi implements LmsTaskCommonApi {
-//    @Override
-//    public CommonResult<AcsApplyActionRespVO> doHandlePicked(TaskStatusCallApiReqDTO taskStatusCallApiReqDTO) {
-//        // 构建外层VO
-//        AcsApplyActionRespVO respVO = new AcsApplyActionRespVO();
-//        respVO.setTaskId(10001L);
-//        respVO.setTaskCode("ACS20260716001");
-//        // 内层data赋值 "success"
-//        return success(respVO);
-//    }
-//
-//    @Override
-//    public CommonResult<AcsApplyActionRespVO> doHandleApplyAgain(TaskStatusCallApiReqDTO taskStatusCallApiReqDTO) {
-//        DemoRespVO demoRespVO = new DemoRespVO();
-//        demoRespVO.setTargetPoint("A_10001");
-//
-//        // 构建外层VO
-//        AcsApplyActionRespVO respVO = new AcsApplyActionRespVO();
-//        respVO.setTaskId(10001L);
-//        respVO.setTaskCode("ACS20260716001");
-//        // 内层data赋值 "success"
-//        respVO.setData(JSON.parseObject(JSON.toJSONString(demoRespVO)));
-//        return success(respVO);
-//    }
-//
-//    @Override
-//    public CommonResult<AcsApplyActionRespVO> doHandleRequestRelease(TaskStatusCallApiReqDTO taskStatusCallApiReqDTO) {
-//        // 构建外层VO
-//        AcsApplyActionRespVO respVO = new AcsApplyActionRespVO();
-//        respVO.setTaskId(10001L);
-//        respVO.setTaskCode("ACS20260716001");
-//        return success(respVO);
-//    }
-//
-//    @Override
-//    public CommonResult<AcsApplyActionRespVO> doHandleRequestPick(TaskStatusCallApiReqDTO taskStatusCallApiReqDTO) {
-//        // 构建外层VO
-//        AcsApplyActionRespVO respVO = new AcsApplyActionRespVO();
-//        respVO.setTaskId(10001L);
-//        respVO.setTaskCode("ACS20260716001");
-//        return success(respVO);
-//    }
-//
-//    @Override
-//    public CommonResult<AcsApplyActionRespVO> doHandleRequestLeave(TaskStatusCallApiReqDTO taskStatusCallApiReqDTO) {
-//        // 构建外层VO
-//        AcsApplyActionRespVO respVO = new AcsApplyActionRespVO();
-//        respVO.setTaskId(10001L);
-//        respVO.setTaskCode("ACS20260716001");
-//        return success(respVO);
-//    }
-//
-//    @Override
-//    public CommonResult<AcsApplyActionRespVO> doHandleRequestEnter(TaskStatusCallApiReqDTO taskStatusCallApiReqDTO) {
-//        // 构建外层VO
-//        AcsApplyActionRespVO respVO = new AcsApplyActionRespVO();
-//        respVO.setTaskId(10001L);
-//        respVO.setTaskCode("ACS20260716001");
-//        return success(respVO);
-//    }
-//}

nl-module-lms/nl-module-lms-server/src/main/java/cn/code/nl/module/lms/mq/consumer/LmsTaskStatusChangeConsumer.java

@@ -0,0 +1,54 @@
+package cn.code.nl.module.lms.mq.consumer;
+
+import cn.code.nl.framework.execute.core.AbstractTask;
+import cn.code.nl.framework.execute.core.TaskFactory;
+import cn.code.nl.framework.execute.core.dto.TaskExecuteDTO;
+import cn.code.nl.module.task.message.TaskEventMessage;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.stereotype.Component;
+
+/**
+ * 监听任务状态变更，根据 handleCode 定位任务子类并执行完成/取消逻辑
+ *
+ * @Author: liyongde
+ * @Date: 2026/7/20 10:28
+ */
+@Slf4j
+@Component
+@RocketMQMessageListener(
+        topic = "${rocketmq.consumer.lms-task-operate.topic}",
+        consumerGroup = "${rocketmq.consumer.lms-task-operate.group}"
+)
+public class LmsTaskStatusChangeConsumer implements RocketMQListener<TaskEventMessage> {
+
+    @Resource
+    private TaskFactory taskFactory;
+
+    @Override
+    public void onMessage(TaskEventMessage message) {
+        String handleCode = message.getHandleCode();
+        String eventType = message.getEventType();
+        log.info("收到任务状态变更消息, handleCode={}, eventType={}, taskId={}", handleCode, eventType, message.getTaskId());
+
+        AbstractTask task = taskFactory.getTask(handleCode);
+        if (task == null) {
+            log.warn("未找到对应任务处理器, handleCode={}", handleCode);
+            return;
+        }
+
+        TaskExecuteDTO dto = new TaskExecuteDTO();
+        dto.setTaskId(message.getTaskId());
+        dto.setPayload(message.getPayload());
+
+        if (TaskEventMessage.EVENT_TYPE_FINISHED.equals(eventType)) {
+            task.doHandleFinish(dto);
+        } else if (TaskEventMessage.EVENT_TYPE_CANCELLED.equals(eventType)) {
+            task.doHandleCancel(dto);
+        } else {
+            log.warn("未知事件类型, eventType={}, handleCode={}", eventType, handleCode);
+        }
+    }
+}

nl-module-lms/nl-module-lms-server/src/main/java/cn/code/nl/module/lms/mq/package-info.java

@@ -0,0 +1,6 @@
+/**
+ *
+ * @Author: liyongde
+ * @Date: 2026/7/23 16:21
+ */
+package cn.code.nl.module.lms.mq;

nl-module-lms/nl-module-lms-server/src/main/resources/application-mq-dev.yaml

@@ -0,0 +1,11 @@
+--- #################### MQ 消息队列相关配置 ####################
+# rocketmq 配置项，对应 RocketMQProperties 配置类
+rocketmq:
+  name-server: 192.168.81.193:9876
+  producer:
+    group: lms-producer-dev-group # 事务消息需要配置一样
+    send-message-timeout: 3000
+  consumer:
+    lms-task-operate:
+      group: lms-task-status-change-dev-group
+      topic: lms-task-status-change-dev-topic

nl-module-lms/nl-module-lms-server/src/main/resources/application.yaml

@@ -13,6 +13,7 @@ spring:
     import:
       - optional:classpath:application-${spring.profiles.active}.yaml # 加载【本地】配置
       - optional:nacos:${spring.application.name}-${spring.profiles.active}.yaml # 加载【Nacos】的配置
+      - optional:classpath:application-mq-${spring.profiles.active}.yaml # 加载 MQ 配置
 
   # Servlet 配置
   servlet:

nl-module-task/nl-module-task-server/src/main/java/cn/code/nl/module/task/mq/producer/TaskEventProducer.java

@@ -6,19 +6,21 @@ import org.apache.rocketmq.spring.core.RocketMQTemplate;
 import org.springframework.stereotype.Component;
 
 import jakarta.annotation.Resource;
+import org.springframework.beans.factory.annotation.Value;
 import java.util.UUID;
 
 /**
  * 任务事件 MQ 生产者
  * <p>
- * Topic: TASK_EVENT_TOPIC
+ * Topic 从 YAML 配置 rocketmq.consumer.task.topic 读取
  * Tag: ownerService (LMS / WMS)
  */
 @Slf4j
 @Component
 public class TaskEventProducer {
 
-    private static final String TOPIC = "TASK_EVENT_TOPIC";
+    @Value("${rocketmq.consumer.task.topic}")
+    private String topic;
 
     @Resource
     private RocketMQTemplate rocketMQTemplate;
@@ -34,7 +36,7 @@ public class TaskEventProducer {
             message.setEventId(UUID.randomUUID().toString());
         }
 
-        String destination = TOPIC + "_" + message.getOwnerService();
+        String destination = message.getOwnerService() + "_" + topic;
         try {
             rocketMQTemplate.syncSend(destination, message);
             log.info("MQ 发送成功, destination={}, taskId={}, eventType={}",

nl-module-task/nl-module-task-server/src/main/resources/application-dev.yaml

@@ -78,9 +78,12 @@ spring:
 
 # rocketmq 配置项，对应 RocketMQProperties 配置类
 rocketmq:
-  name-server: 127.0.0.1:9876 # RocketMQ Namesrv
+  name-server: 192.168.81.193:9876 # RocketMQ Namesrv
   producer:
-    group: ${spring.application.name}_TASK_DEV_PRODUCER # 生产者分组
+    group: task-producer-dev-group  # 生产者分组
+  consumer:
+    task:
+      topic: task-status-change-dev-topic
 
 spring:
   # RabbitMQ 配置项，对应 RabbitProperties 配置类

nl-module-wms/nl-module-wms-server/src/main/java/cn/code/nl/module/wms/mq/consumer/WmsTaskStatusChangeConsumer.java

@@ -22,7 +22,7 @@ import org.springframework.stereotype.Component;
         topic = "${rocketmq.consumer.wms-task-operate.topic}",
         consumerGroup = "${rocketmq.consumer.wms-task-operate.group}"
 )
-public class TaskStatusChangeConsumer implements RocketMQListener<TaskEventMessage> {
+public class WmsTaskStatusChangeConsumer implements RocketMQListener<TaskEventMessage> {
 
     @Resource
     private TaskFactory taskFactory;

nl-module-wms/nl-module-wms-server/src/main/resources/application-mq-dev.yaml

@@ -2,6 +2,9 @@
 # rocketmq 配置项，对应 RocketMQProperties 配置类
 rocketmq:
   name-server: 192.168.81.193:9876
+  producer:
+    group: wms-producer-dev-group # 事务消息需要配置一样
+    send-message-timeout: 3000
   consumer:
     wms-task-operate:
       group: wms-task-status-change-dev-group

nl-server/src/main/resources/application-mq-dev.yaml

@@ -2,6 +2,9 @@
 # rocketmq 配置项，对应 RocketMQProperties 配置类
 rocketmq:
   name-server: 192.168.81.193:9876
+  producer:
+    group: nl-producer-dev-group
+    send-message-timeout: 3000
   consumer:
     wms-task-operate:
       group: wms-task-status-change-dev-group