From 1a8474e263ab336681f2fa149f92882eec27e38b Mon Sep 17 00:00:00 2001 From: liyongde <1419499670@qq.com> Date: Mon, 20 Jul 2026 14:33:10 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20wms=E4=BB=BB=E5=8A=A1=E5=AE=8C=E6=88=90?= =?UTF-8?q?=E5=8F=96=E6=B6=88=E6=B6=88=E8=B4=B9=E8=80=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../framework/execute/core/TaskFactory.java | 16 +++---- .../module/task/message/TaskEventMessage.java | 7 +++- nl-module-wms/nl-module-wms-server/pom.xml | 5 +++ .../mq/consumer/TaskStatusChangeConsumer.java | 42 ++++++++++++++++--- .../src/main/resources/application-dev.yaml | 5 --- .../main/resources/application-mq-dev.yaml | 8 ++++ .../main/resources/application-mq-test.yaml | 8 ++++ .../src/main/resources/application.yaml | 1 + 8 files changed, 69 insertions(+), 23 deletions(-) create mode 100644 nl-module-wms/nl-module-wms-server/src/main/resources/application-mq-dev.yaml create mode 100644 nl-module-wms/nl-module-wms-server/src/main/resources/application-mq-test.yaml diff --git a/nl-framework/nl-spring-boot-starter-execute/src/main/java/cn/code/nl/framework/execute/core/TaskFactory.java b/nl-framework/nl-spring-boot-starter-execute/src/main/java/cn/code/nl/framework/execute/core/TaskFactory.java index 4b16f8c9..14d4e455 100644 --- a/nl-framework/nl-spring-boot-starter-execute/src/main/java/cn/code/nl/framework/execute/core/TaskFactory.java +++ b/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 taskMap; - @Autowired - public TaskFactory() { - taskMap = new HashMap<>(); - } + private final Map taskMap = new ConcurrentHashMap<>(); @Override public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { - if (bean instanceof AbstractTask) { - taskMap.put(beanName, (AbstractTask) bean); + if (bean instanceof AbstractTask task) { + 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); diff --git a/nl-module-task/nl-module-task-api/src/main/java/cn/code/nl/module/task/message/TaskEventMessage.java b/nl-module-task/nl-module-task-api/src/main/java/cn/code/nl/module/task/message/TaskEventMessage.java index 7cf0c26b..5d7752bc 100644 --- a/nl-module-task/nl-module-task-api/src/main/java/cn/code/nl/module/task/message/TaskEventMessage.java +++ b/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 */ diff --git a/nl-module-wms/nl-module-wms-server/pom.xml b/nl-module-wms/nl-module-wms-server/pom.xml index 72084eae..ef12b750 100644 --- a/nl-module-wms/nl-module-wms-server/pom.xml +++ b/nl-module-wms/nl-module-wms-server/pom.xml @@ -37,6 +37,11 @@ nl-module-system-api ${revision} + + cn.nl.cloud + nl-module-task-api + ${revision} + diff --git a/nl-module-wms/nl-module-wms-server/src/main/java/cn/code/nl/module/wms/mq/consumer/TaskStatusChangeConsumer.java b/nl-module-wms/nl-module-wms-server/src/main/java/cn/code/nl/module/wms/mq/consumer/TaskStatusChangeConsumer.java index e7634380..57cbb940 100644 --- a/nl-module-wms/nl-module-wms-server/src/main/java/cn/code/nl/module/wms/mq/consumer/TaskStatusChangeConsumer.java +++ b/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}", - consumerGroup = "${rocketmq.push-consumer.group.wms-task-status-change}", - namespace = "${rocketmq.push-consumer.namespace}") -public class TaskStatusChangeConsumer implements RocketMQListener { +@RocketMQMessageListener( + topic = "${rocketmq.consumer.wms-task-operate.topic}", + consumerGroup = "${rocketmq.consumer.wms-task-operate.group}" +) +public class TaskStatusChangeConsumer implements RocketMQListener { + + @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); + } } } diff --git a/nl-module-wms/nl-module-wms-server/src/main/resources/application-dev.yaml b/nl-module-wms/nl-module-wms-server/src/main/resources/application-dev.yaml index 4f34ac9c..884fc0e9 100644 --- a/nl-module-wms/nl-module-wms-server/src/main/resources/application-dev.yaml +++ b/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: diff --git a/nl-module-wms/nl-module-wms-server/src/main/resources/application-mq-dev.yaml b/nl-module-wms/nl-module-wms-server/src/main/resources/application-mq-dev.yaml new file mode 100644 index 00000000..1c2b8847 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/nl-module-wms/nl-module-wms-server/src/main/resources/application-mq-test.yaml b/nl-module-wms/nl-module-wms-server/src/main/resources/application-mq-test.yaml new file mode 100644 index 00000000..91e76da5 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/nl-module-wms/nl-module-wms-server/src/main/resources/application.yaml b/nl-module-wms/nl-module-wms-server/src/main/resources/application.yaml index c5fb3505..a3e0c8e7 100644 --- a/nl-module-wms/nl-module-wms-server/src/main/resources/application.yaml +++ b/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: