fix: wms任务完成取消消费者

nl-framework/nl-spring-boot-starter-execute/src/main/java/cn/code/nl/framework/execute/core/TaskFactory.java

@@ -1,37 +1,31 @@
 package cn.code.nl.framework.execute.core;
 
 import org.springframework.beans.BeansException;
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.config.BeanPostProcessor;
 import org.springframework.stereotype.Component;
 
-import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 /**
- *
  * @Author: liyongde
  * @Date: 2026/7/15 15:32
  */
 @Component
 public class TaskFactory implements BeanPostProcessor {
-    private final Map<String, AbstractTask> taskMap;
 
-    @Autowired
+    private final Map<String, AbstractTask> taskMap = new ConcurrentHashMap<>();
-    public TaskFactory() {
-        taskMap = new HashMap<>();
-    }
 
     @Override
     public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
-        if (bean instanceof AbstractTask) {
+        if (bean instanceof AbstractTask task) {
-            taskMap.put(beanName, (AbstractTask) bean);
+            taskMap.put(beanName, task);
         }
         return bean;
     }
 
     public AbstractTask getTask(String handleCode) {
-        if (handleCode == null) {
+        if (handleCode == null || handleCode.isBlank()) {
             return null;
         }
         return taskMap.get(handleCode);

nl-module-task/nl-module-task-api/src/main/java/cn/code/nl/module/task/message/TaskEventMessage.java

@@ -10,10 +10,15 @@ import java.util.Map;
 @Data
 public class TaskEventMessage {
 
+    /** 任务完成 */
+    public static final String EVENT_TYPE_FINISHED = "TASK_FINISHED";
+    /** 任务取消 */
+    public static final String EVENT_TYPE_CANCELLED = "TASK_CANCELLED";
+
     /** 事件ID */
     private String eventId;
 
-    /** 事件类型：TASK_FINISHED / TASK_CANCELLED */
+    /** 事件类型，取值：{@link #EVENT_TYPE_FINISHED} / {@link #EVENT_TYPE_CANCELLED} */
     private String eventType;
 
     /** 任务ID */

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

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

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

@@ -1,24 +1,54 @@
 package cn.code.nl.module.wms.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.topic.wms-task-status-change}",
+@RocketMQMessageListener(
-        consumerGroup = "${rocketmq.push-consumer.group.wms-task-status-change}",
+        topic = "${rocketmq.consumer.wms-task-operate.topic}",
-        namespace = "${rocketmq.push-consumer.namespace}")
+        consumerGroup = "${rocketmq.consumer.wms-task-operate.group}"
-public class TaskStatusChangeConsumer implements RocketMQListener {
+)
+public class TaskStatusChangeConsumer implements RocketMQListener<TaskEventMessage> {
+
+    @Resource
+    private TaskFactory taskFactory;
 
     @Override
-    public void onMessage(Object o) {
+    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-wms/nl-module-wms-server/src/main/resources/application-dev.yaml

@@ -75,11 +75,6 @@ spring:
 #      password: 123456 # 密码，建议生产环境开启
 
 --- #################### MQ 消息队列相关配置 ####################
-
-# rocketmq 配置项，对应 RocketMQProperties 配置类
-rocketmq:
-  name-server: 127.0.0.1:9876 # RocketMQ Namesrv
-
 spring:
   # RabbitMQ 配置项，对应 RabbitProperties 配置类
   rabbitmq:

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

@@ -0,0 +1,8 @@
+--- #################### MQ 消息队列相关配置 ####################
+# rocketmq 配置项，对应 RocketMQProperties 配置类
+rocketmq:
+  name-server: 127.0.0.1:9876
+  consumer:
+    wms-task-operate:
+      group: wms-task-status-change-dev-group
+      topic: wms-task-status-change-dev-topic

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

@@ -0,0 +1,8 @@
+--- #################### MQ 消息队列相关配置 ####################
+# rocketmq 配置项，对应 RocketMQProperties 配置类
+rocketmq:
+  name-server: 127.0.0.1:9876
+  consumer:
+    wms-task-operate:
+      group: wms-task-status-change-test-group
+      topic: wms-task-status-change-test-topic

nl-module-wms/nl-module-wms-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: