zhouz/huachuang
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/feature/20260716/wms-module' into feature/20260716/wms-module

Browse Source
This commit is contained in:
zhouz
2026-07-27 14:56:27 +08:00
parent 2269022205 2799b9f54e
commit 73491b412c
67 changed files with 2214 additions and 723 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
nl-module-wms/nl-module-wms-server/src/main/java/cn/code/nl/module/wms/api/WmsTaskExecuteApiImpl.java Normal file
View File

@@ -0,0 +1,21 @@
package cn.code.nl.module.wms.api;
import cn.code.nl.framework.common.enums.RpcConstants;
import cn.code.nl.framework.execute.biz.api.AbstractTaskCommonApiImpl;
import cn.code.nl.framework.execute.biz.api.wms.WmsTaskCommonApi;
import org.springframework.context.annotation.Primary;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
/**
* WMS 任务通用 API 实现
*
* @author liyongde
*/
@RestController
@Validated
@Primary
@RequestMapping(RpcConstants.WMS_PREFIX)
public class WmsTaskExecuteApiImpl extends AbstractTaskCommonApiImpl implements WmsTaskCommonApi {
}

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

@@ -1,54 +0,0 @@
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.consumer.wms-task-operate.topic}",
consumerGroup = "${rocketmq.consumer.wms-task-operate.group}"
)
public class TaskStatusChangeConsumer 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);
}
}
}

118
nl-module-wms/nl-module-wms-server/src/main/java/cn/code/nl/module/wms/mq/consumer/WmsTaskStatusChangeConsumer.java Normal file
View File

@@ -0,0 +1,118 @@
package cn.code.nl.module.wms.mq.consumer;
import cn.code.nl.framework.common.pojo.CommonResult;
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.api.TransportTaskApi;
import cn.code.nl.module.task.dto.TaskInfoDTO;
import cn.code.nl.module.task.enums.TransportTaskStatusEnum;
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.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Component;
import static cn.code.nl.module.task.message.LockKeyConstants.TASK_STATUS_CHANGE_LOCK_KEY;
/**
* 监听任务状态变更，根据 handleCode 定位任务子类并执行完成/取消逻辑
*
* @Author: liyongde
* @Date: 2026/7/20 10:28
*/
@Slf4j
@Component
@RocketMQMessageListener(
topic = "${rocketmq.consumer.wms-task-operate.topic}",
consumerGroup = "${rocketmq.consumer.wms-task-operate.group}"
)
public class WmsTaskStatusChangeConsumer implements RocketMQListener<TaskEventMessage> {
@Resource
private TaskFactory taskFactory;
@Resource
private TransportTaskApi transportTaskApi;
@Resource
private RedissonClient redissonClient;
@Override
public void onMessage(TaskEventMessage message) {
String handleCode = message.getHandleCode();
String eventType = message.getEventType();
log.info("收到任务状态变更消息, handleCode={}, eventType={}, taskId={}", handleCode, eventType, message.getTaskId());
RLock lock = redissonClient.getLock(TASK_STATUS_CHANGE_LOCK_KEY + message.getTaskId());
if (!lock.tryLock()) {
log.warn("任务状态变更消息正在消费中，等待 MQ 重试, taskId={}, eventType={}", message.getTaskId(), eventType);
throw new IllegalStateException("任务状态变更消息正在消费中......");
}
try {
if (!isTaskCallbackPending(message)) {
return;
}
executeTaskHandler(message, handleCode, eventType);
} finally {
if (lock.isHeldByCurrentThread()) {
lock.unlock();
}
}
}
/**
* 判断任务是否处于当前事件对应的待业务处理状态
*/
private boolean isTaskCallbackPending(TaskEventMessage message) {
String eventType = message.getEventType();
String expectedStatus;
if (TaskEventMessage.EVENT_TYPE_FINISHED.equals(eventType)) {
expectedStatus = TransportTaskStatusEnum.FINISHED_CALLBACK_PENDING.getCode();
} else if (TaskEventMessage.EVENT_TYPE_CANCELLED.equals(eventType)) {
expectedStatus = TransportTaskStatusEnum.CANCEL_CALLBACK_PENDING.getCode();
} else {
log.warn("未知任务事件类型, eventType={}, taskId={}", eventType, message.getTaskId());
return false;
}
CommonResult<TaskInfoDTO> result = transportTaskApi.getTaskById(message.getTaskId());
TaskInfoDTO taskInfo = result.getCheckedData();
if (taskInfo == null) {
log.warn("任务不存在，跳过任务状态变更消息, taskId={}, eventType={}", message.getTaskId(), eventType);
return false;
}
if (!expectedStatus.equals(taskInfo.getTaskStatus())) {
log.info("任务状态已处理，跳过重复消息, taskId={}, eventType={}, currentStatus={}, expectedStatus={}",
message.getTaskId(), eventType, taskInfo.getTaskStatus(), expectedStatus);
return false;
}
return true;
}
/**
* 执行任务完成或取消业务处理器
*/
private void executeTaskHandler(TaskEventMessage message, String handleCode, String eventType) {
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);
}
}
}

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

@@ -2,7 +2,10 @@
# 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
topic: wms-task-status-change-dev-topic
group: wms_task_status_change_dev_group
topic: wms_task_status_change_dev_topic