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

Merge branch 'refs/heads/feature/20260713/task-module' into feature/20260808/packaging_workshop

Browse Source 
# Conflicts:
#	nl-framework/nl-spring-boot-starter-execute/src/main/java/cn/code/nl/framework/execute/util/http/AcsUtil.java
#	nl-server/src/test/java/cn/code/nl/server/acs/AcsUtilLocalSpringTest.java
This commit is contained in:
liyongde
2026-08-10 14:28:27 +08:00
parent 0bde732829 d9f21147aa
commit df24f4c88b
25 changed files with 834 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
nl-module-task/nl-module-task-api/pom.xml
View File

@@ -49,6 +49,13 @@
<artifactId>nl-spring-boot-starter-execute</artifactId>
</dependency>
<!-- Test 测试相关 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>
</project>

35
nl-module-task/nl-module-task-api/src/main/java/cn/code/nl/module/task/enums/TaskCallbackTypeEnum.java Normal file
View File

@@ -0,0 +1,35 @@
package cn.code.nl.module.task.enums;
import lombok.Getter;
/**
* 任务业务回调方式
*/
@Getter
public enum TaskCallbackTypeEnum {
MQ("mq"),
FEIGN("feign");
private final String code;
TaskCallbackTypeEnum(String code) {
this.code = code;
}
/**
* 根据编码获取任务业务回调方式
*
* @param code 回调方式编码
* @return 任务业务回调方式
*/
public static TaskCallbackTypeEnum getByCode(String code) {
for (TaskCallbackTypeEnum value : values()) {
if (value.getCode().equals(code)) {
return value;
}
}
return null;
}
}

50
nl-module-task/nl-module-task-api/src/main/java/cn/code/nl/module/task/enums/TaskOwnerServiceEnum.java Normal file
View File

@@ -0,0 +1,50 @@
package cn.code.nl.module.task.enums;
import lombok.AllArgsConstructor;
import lombok.Getter;
/**
* 任务业务归属服务枚举
*/
@Getter
@AllArgsConstructor
public enum TaskOwnerServiceEnum {
LMS("lms", "lms-server"),
WMS("wms", "wms-server");
/**
* MQ Topic 前缀
*/
private final String topicPrefix;
/**
* 服务注册名称
*/
private final String serviceName;
/**
* 根据业务归属编码获取枚举
*
* @param code 业务归属编码
* @return 业务归属服务枚举
*/
public static TaskOwnerServiceEnum getByCode(String code) {
for (TaskOwnerServiceEnum value : values()) {
if (value.name().equalsIgnoreCase(code) || value.serviceName.equalsIgnoreCase(code)) {
return value;
}
}
return null;
}
/**
* 构建业务服务消费的 Topic
*
* @param topic 基础 Topic
* @return 完整 Topic
*/
public String buildTopic(String topic) {
return topicPrefix + "_" + topic;
}
}

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

@@ -44,4 +44,15 @@ public class TaskEventMessage {
/** 扩展数据 */
private Map<String, Object> payload;
/**
* 构建任务事件唯一标识
*
* @param taskId 任务标识
* @param eventType 事件类型
* @return 任务事件唯一标识
*/
public static String buildEventId(Long taskId, String eventType) {
return taskId + "_" + eventType;
}
}

56
nl-module-task/nl-module-task-api/src/test/java/cn/code/nl/module/task/enums/TaskOwnerServiceEnumLocalSpringTest.java Normal file
View File

@@ -0,0 +1,56 @@
package cn.code.nl.module.task.enums;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.context.annotation.Configuration;
/**
* 任务业务归属服务本地 Spring 集成测试
*/
@SpringBootTest(classes = TaskOwnerServiceEnumLocalSpringTest.TestConfiguration.class)
class TaskOwnerServiceEnumLocalSpringTest {
/**
* 验证 LMS 业务归属编码生成消费者 Topic
*/
@Test
void buildTopicShouldSupportLmsOwnerService() {
TaskOwnerServiceEnum ownerService = TaskOwnerServiceEnum.getByCode("LMS");
Assertions.assertNotNull(ownerService);
Assertions.assertEquals("lms_task_status_change_dev_topic",
ownerService.buildTopic("task_status_change_dev_topic"));
}
/**
* 验证 WMS 业务归属编码生成消费者 Topic
*/
@Test
void buildTopicShouldSupportWmsOwnerService() {
TaskOwnerServiceEnum ownerService = TaskOwnerServiceEnum.getByCode("WMS");
Assertions.assertNotNull(ownerService);
Assertions.assertEquals("wms_task_status_change_dev_topic",
ownerService.buildTopic("task_status_change_dev_topic"));
}
/**
* 验证兼容服务注册名称并拒绝非法业务归属
*/
@Test
void getByCodeShouldSupportServiceNameAndRejectInvalidCode() {
Assertions.assertEquals(TaskOwnerServiceEnum.LMS,
TaskOwnerServiceEnum.getByCode("lms-server"));
Assertions.assertEquals(TaskOwnerServiceEnum.WMS,
TaskOwnerServiceEnum.getByCode("wms-server"));
Assertions.assertNull(TaskOwnerServiceEnum.getByCode("invalid-server"));
}
/**
* 测试 Spring 配置
*/
@Configuration
static class TestConfiguration {
}
}

152
nl-module-task/nl-module-task-server/src/main/java/cn/code/nl/module/task/manage/TransportTaskOperationManager.java
View File

@@ -1,22 +1,30 @@
package cn.code.nl.module.task.manage;
import cn.code.nl.framework.common.exception.ServiceException;
import cn.code.nl.framework.execute.biz.api.TaskCommonApi;
import cn.code.nl.framework.execute.biz.dto.TaskStatusCallApiReqDTO;
import cn.code.nl.framework.execute.core.TaskCommonApiFactory;
import cn.code.nl.module.task.dal.dataobject.transporttask.TransportTaskDO;
import cn.code.nl.module.task.dal.mysql.transporttask.TransportTaskMapper;
import cn.code.nl.module.task.dto.AcsFeedbackReqDTO;
import cn.code.nl.module.task.dto.TaskCallbackResultReqDTO;
import cn.code.nl.module.task.enums.CallbackStatusEnum;
import cn.code.nl.module.task.enums.FinishedTypeEnum;
import cn.code.nl.module.task.enums.TaskCallbackTypeEnum;
import cn.code.nl.module.task.enums.TaskEventTypeEnum;
import cn.code.nl.module.task.enums.TaskOperationTypeEnum;
import cn.code.nl.module.task.enums.TaskOwnerServiceEnum;
import cn.code.nl.module.task.enums.TransportTaskStatusEnum;
import cn.code.nl.module.task.message.TaskEventMessage;
import cn.code.nl.module.task.mq.producer.TaskEventProducer;
import cn.code.nl.module.task.service.transporttask.TransportTaskCallbackService;
import cn.hutool.core.util.StrUtil;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;
@@ -51,9 +59,21 @@ public class TransportTaskOperationManager {
@Resource
private TransportTaskIssueManager transportTaskIssueManager;
@Resource
private TaskCommonApiFactory taskCommonApiFactory;
@Resource
private TransportTaskCallbackService transportTaskCallbackService;
@Resource
private RedissonClient redissonClient;
@Value("${nl.task.callback-type}")
private String callbackType;
/** 任务业务回调方式 */
private TaskCallbackTypeEnum callbackTypeEnum;
/** 操作类型路由表 */
private final Map<TaskOperationTypeEnum, BiConsumer<TransportTaskDO, AcsFeedbackReqDTO>> operationHandlers =
new EnumMap<>(TaskOperationTypeEnum.class);
@@ -63,6 +83,10 @@ public class TransportTaskOperationManager {
*/
@PostConstruct
public void initOperationHandlers() {
callbackTypeEnum = TaskCallbackTypeEnum.getByCode(callbackType);
if (callbackTypeEnum == null) {
throw new IllegalArgumentException("任务业务回调方式配置错误：" + callbackType);
}
operationHandlers.put(TaskOperationTypeEnum.EXECUTING, this::handleExecuting);
operationHandlers.put(TaskOperationTypeEnum.ISSUE, (task, reqDTO) -> handleIssue(task));
operationHandlers.put(TaskOperationTypeEnum.FINISHED, this::handleFinished);
@@ -91,7 +115,7 @@ public class TransportTaskOperationManager {
throw exception(TRANSPORT_TASK_OPERATION_FAILED, ex.getMessage());
} finally {
if (lock.isHeldByCurrentThread()) {
lock.unlock();
unlockAfterTransaction(lock);
}
}
} else {
@@ -147,19 +171,7 @@ public class TransportTaskOperationManager {
task.setCallbackStatus(CallbackStatusEnum.PENDING.getCode());
transportTaskMapper.updateById(task);
// MQ 在事务提交后发送，避免 Consumer 读到未提交的数据
Map<String, Object> payload = reqDTO.getPayload();
if (TransactionSynchronizationManager.isSynchronizationActive()) {
TransactionSynchronizationManager.registerSynchronization(
new TransactionSynchronization() {
@Override
public void afterCommit() {
publishEvent(task, TaskEventTypeEnum.TASK_FINISHED, payload);
}
});
} else {
publishEvent(task, TaskEventTypeEnum.TASK_FINISHED, payload);
}
executeBusinessCallback(task, TaskEventTypeEnum.TASK_FINISHED, reqDTO.getPayload());
}
/**
@@ -182,19 +194,7 @@ public class TransportTaskOperationManager {
task.setCallbackStatus(CallbackStatusEnum.PENDING.getCode());
transportTaskMapper.updateById(task);
// MQ 在事务提交后发送，避免 Consumer 读到未提交的数据
Map<String, Object> payload = reqDTO.getPayload();
if (TransactionSynchronizationManager.isSynchronizationActive()) {
TransactionSynchronizationManager.registerSynchronization(
new TransactionSynchronization() {
@Override
public void afterCommit() {
publishEvent(task, TaskEventTypeEnum.TASK_CANCELLED, payload);
}
});
} else {
publishEvent(task, TaskEventTypeEnum.TASK_CANCELLED, payload);
}
executeBusinessCallback(task, TaskEventTypeEnum.TASK_CANCELLED, reqDTO.getPayload());
}
/**
@@ -207,6 +207,104 @@ public class TransportTaskOperationManager {
log.info("任务强制完成, taskId={}", task.getTaskId());
}
/**
* 在事务完成后释放任务状态锁
*/
private void unlockAfterTransaction(RLock lock) {
if (!TransactionSynchronizationManager.isSynchronizationActive()) {
lock.unlock();
return;
}
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
@Override
public void afterCompletion(int status) {
if (lock.isHeldByCurrentThread()) {
lock.unlock();
}
}
});
}
/**
* 根据配置执行任务业务回调
*/
private void executeBusinessCallback(TransportTaskDO task, TaskEventTypeEnum eventType,
Map<String, Object> payload) {
TaskOwnerServiceEnum ownerService = TaskOwnerServiceEnum.getByCode(task.getOwnerService());
if (ownerService == null) {
throw new ServiceException(500, "任务业务归属服务配置错误：" + task.getOwnerService());
}
if (callbackTypeEnum == TaskCallbackTypeEnum.MQ) {
publishEventAfterCommit(task, eventType, payload);
return;
}
executeFeignCallback(task, ownerService, eventType, payload);
}
/**
* 在事务提交后发布任务事件
*/
private void publishEventAfterCommit(TransportTaskDO task, TaskEventTypeEnum eventType,
Map<String, Object> payload) {
if (TransactionSynchronizationManager.isSynchronizationActive()) {
TransactionSynchronizationManager.registerSynchronization(
new TransactionSynchronization() {
@Override
public void afterCommit() {
publishEvent(task, eventType, payload);
}
});
return;
}
publishEvent(task, eventType, payload);
}
/**
* 通过 Feign 同步执行任务业务并更新最终状态
*/
private void executeFeignCallback(TransportTaskDO task, TaskOwnerServiceEnum ownerService,
TaskEventTypeEnum eventType, Map<String, Object> payload) {
TaskCommonApi taskCommonApi = taskCommonApiFactory.getByServerName(ownerService.name());
TaskStatusCallApiReqDTO callApiReqDTO = buildCallApiReqDTO(task, eventType, payload);
if (eventType == TaskEventTypeEnum.TASK_FINISHED) {
taskCommonApi.doHandleFinished(callApiReqDTO).getCheckedData();
} else {
taskCommonApi.doHandleCancelled(callApiReqDTO).getCheckedData();
}
transportTaskCallbackService.receiveCallbackResult(buildSuccessCallbackReq(task, eventType));
}
/**
* 构建 Feign 任务业务调用参数
*/
private TaskStatusCallApiReqDTO buildCallApiReqDTO(TransportTaskDO task, TaskEventTypeEnum eventType,
Map<String, Object> payload) {
TaskStatusCallApiReqDTO reqDTO = new TaskStatusCallApiReqDTO();
reqDTO.setTaskId(task.getTaskId());
reqDTO.setEventId(TaskEventMessage.buildEventId(task.getTaskId(), eventType.getCode()));
reqDTO.setTaskCode(task.getTaskCode());
reqDTO.setStatus(eventType.getCode());
reqDTO.setOwnerService(task.getOwnerService());
reqDTO.setBizType(task.getBizType());
reqDTO.setBizId(task.getBizId());
reqDTO.setHandleCode(task.getHandleCode());
reqDTO.setPayload(payload);
return reqDTO;
}
/**
* 构建业务处理成功回调参数
*/
private TaskCallbackResultReqDTO buildSuccessCallbackReq(TransportTaskDO task,
TaskEventTypeEnum eventType) {
TaskCallbackResultReqDTO reqDTO = new TaskCallbackResultReqDTO();
reqDTO.setTaskId(task.getTaskId());
reqDTO.setEventId(TaskEventMessage.buildEventId(task.getTaskId(), eventType.getCode()));
reqDTO.setResult(CallbackStatusEnum.SUCCESS.getCode());
reqDTO.setMessage("任务业务处理成功");
return reqDTO;
}
/**
* 发布 MQ 事件
*/

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

@@ -1,5 +1,6 @@
package cn.code.nl.module.task.mq.producer;
import cn.code.nl.module.task.enums.TaskOwnerServiceEnum;
import cn.code.nl.module.task.message.TaskEventMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
@@ -7,7 +8,6 @@ import org.springframework.stereotype.Component;
import jakarta.annotation.Resource;
import org.springframework.beans.factory.annotation.Value;
import java.util.UUID;
/**
* 任务事件 MQ 生产者
@@ -31,9 +31,12 @@ public class TaskEventProducer {
* @param message 事件消息
*/
public void publishEvent(TaskEventMessage message) {
message.setEventId(message.getTaskId().toString());
String destination = message.getOwnerService() + "_" + topic;
message.setEventId(TaskEventMessage.buildEventId(message.getTaskId(), message.getEventType()));
TaskOwnerServiceEnum ownerService = TaskOwnerServiceEnum.getByCode(message.getOwnerService());
if (ownerService == null) {
throw new IllegalArgumentException("任务业务归属服务配置错误：" + message.getOwnerService());
}
String destination = ownerService.buildTopic(topic);
try {
rocketMQTemplate.syncSend(destination, message);
log.info("MQ 发送成功, destination={}, taskId={}, eventType={}",

61
nl-module-task/nl-module-task-server/src/main/java/cn/code/nl/module/task/service/transporttask/TransportTaskCallbackServiceImpl.java
View File

@@ -1,24 +1,26 @@
package cn.code.nl.module.task.service.transporttask;
import cn.code.nl.framework.common.exception.ServiceException;
import cn.code.nl.module.task.dal.dataobject.transporttask.TransportTaskDO;
import cn.code.nl.module.task.dal.mysql.transporttask.TransportTaskMapper;
import cn.code.nl.module.task.dto.TaskCallbackResultReqDTO;
import cn.code.nl.module.task.enums.CallbackStatusEnum;
import cn.code.nl.module.task.enums.TaskEventTypeEnum;
import cn.code.nl.module.task.enums.TransportTaskStatusEnum;
import cn.code.nl.module.task.message.TaskEventMessage;
import com.mzt.logapi.context.LogRecordContext;
import com.mzt.logapi.starter.annotation.LogRecord;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.validation.annotation.Validated;
import jakarta.annotation.Resource;
import static cn.code.nl.module.system.enums.LogRecordConstants.*;
import static cn.code.nl.module.system.enums.LogRecordConstants.TASK_INFO;
import static cn.code.nl.module.system.enums.LogRecordConstants.TASK_INFO_CALL_BY_RPC;
import static cn.code.nl.module.system.enums.LogRecordConstants.TASK_INFO_OPERATE_TYPE;
/**
* 业务回调结果处理服务实现
*
* @author 诺力管理员
*/
@Slf4j
@Service
@@ -34,35 +36,37 @@ public class TransportTaskCallbackServiceImpl implements TransportTaskCallbackSe
bizNo = "{{#reqDTO.taskId}}",
success = TASK_INFO_CALL_BY_RPC)
public void receiveCallbackResult(TaskCallbackResultReqDTO reqDTO) {
// 1. 查询任务
TransportTaskDO task = transportTaskMapper.selectById(reqDTO.getTaskId());
if (task == null) {
log.warn("回调 taskId 不存在, taskId={}", reqDTO.getTaskId());
log.warn("回调任务不存在，taskId={}", reqDTO.getTaskId());
return;
}
// 2. 幂等：已是终态则直接返回
String currentStatus = task.getTaskStatus();
String finalFinished = statusCode(TransportTaskStatusEnum.FINISHED);
String finalCancelled = statusCode(TransportTaskStatusEnum.CANCELLED);
String finalFinished = TransportTaskStatusEnum.FINISHED.getCode();
String finalCancelled = TransportTaskStatusEnum.CANCELLED.getCode();
if (finalFinished.equals(currentStatus) || finalCancelled.equals(currentStatus)) {
log.info("任务已处终态, taskId={}, status={}, 忽略回调",
log.info("任务已处于终态，忽略重复回调，taskId={}, status={}",
reqDTO.getTaskId(), currentStatus);
return;
}
// 3. 根据结果处理
boolean success = "SUCCESS".equalsIgnoreCase(reqDTO.getResult());
TaskEventTypeEnum eventType = getPendingEventType(currentStatus);
if (eventType == null) {
throw new ServiceException(500, "任务当前状态不允许处理业务回调：" + currentStatus);
}
String expectedEventId = TaskEventMessage.buildEventId(reqDTO.getTaskId(), eventType.getCode());
if (!expectedEventId.equals(reqDTO.getEventId())) {
throw new ServiceException(500, "任务业务回调事件标识不匹配：" + reqDTO.getEventId());
}
boolean success = CallbackStatusEnum.SUCCESS.getCode().equalsIgnoreCase(reqDTO.getResult());
if (success) {
// 根据当前状态决定目标终态
if (statusCode(TransportTaskStatusEnum.FINISHED_CALLBACK_PENDING).equals(currentStatus)) {
task.setTaskStatus(finalFinished);
} else if (statusCode(TransportTaskStatusEnum.CANCEL_CALLBACK_PENDING).equals(currentStatus)) {
task.setTaskStatus(finalCancelled);
}
task.setTaskStatus(eventType == TaskEventTypeEnum.TASK_FINISHED
? finalFinished : finalCancelled);
task.setCallbackStatus(CallbackStatusEnum.SUCCESS.getCode());
task.setCallbackErrorMsg(null);
} else {
// 业务处理失败：记录失败信息，增加重试次数
task.setCallbackStatus(CallbackStatusEnum.FAILED.getCode());
task.setCallbackErrorMsg(reqDTO.getMessage());
task.setCallbackRetryCount(task.getCallbackRetryCount() != null
@@ -70,15 +74,22 @@ public class TransportTaskCallbackServiceImpl implements TransportTaskCallbackSe
}
transportTaskMapper.updateById(task);
LogRecordContext.putVariable("ownerService", task.getOwnerService());
LogRecordContext.putVariable("taskId", task.getTaskId());
log.info("业务回调结果已处理, taskId={}, result={}, newStatus={}",
log.info("业务回调结果已处理，taskId={}, result={}, newStatus={}",
reqDTO.getTaskId(), reqDTO.getResult(), task.getTaskStatus());
}
private String statusCode(TransportTaskStatusEnum statusEnum) {
return String.format("%03d", statusEnum.getCode());
/**
* 根据任务待处理状态获取事件类型
*/
private TaskEventTypeEnum getPendingEventType(String currentStatus) {
if (TransportTaskStatusEnum.FINISHED_CALLBACK_PENDING.getCode().equals(currentStatus)) {
return TaskEventTypeEnum.TASK_FINISHED;
}
if (TransportTaskStatusEnum.CANCEL_CALLBACK_PENDING.getCode().equals(currentStatus)) {
return TaskEventTypeEnum.TASK_CANCELLED;
}
return null;
}
}

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

@@ -59,12 +59,12 @@ spring:
master:
url: jdbc:mysql://127.0.0.1:3306/huachuang_lms_dev?useSSL=false&serverTimezone=Asia/Shanghai&allowPublicKeyRetrieval=true&nullCatalogMeansCurrent=true&rewriteBatchedStatements=true # MySQL Connector/J 8.X 连接的示例
username: root
password: root
password: root123
slave: # 模拟从库，可根据自己需要修改 # 模拟从库，可根据自己需要修改
lazy: true # 开启懒加载，保证启动速度
url: jdbc:mysql://127.0.0.1:3306/huachuang_lms_dev?useSSL=false&serverTimezone=Asia/Shanghai&allowPublicKeyRetrieval=true&nullCatalogMeansCurrent=true&rewriteBatchedStatements=true # MySQL Connector/J 8.X 连接的示例
username: root
password: root
password: root123
# Redis 配置。Redisson 默认的配置足够使用，一般不需要进行调优
data:
@@ -83,7 +83,7 @@ rocketmq:
group: task_producer_dev_group # 生产者分组
consumer:
task:
topic: task-status-change-dev-topic
topic: task_status_change_dev_topic
spring:
# RabbitMQ 配置项，对应 RabbitProperties 配置类
@@ -145,4 +145,4 @@ logging:
# 芋道配置项，设置当前项目所有自定义的配置
nl:
demo: false # 开启演示模式
demo: false # 开启演示模式

2
nl-module-task/nl-module-task-server/src/main/resources/application.yaml
View File

@@ -113,6 +113,8 @@ nl:
info:
version: 1.0.0
base-package: cn.code.nl.module.task
task:
callback-type: mq # 任务完成/取消业务处理方式：mq 异步、feign 同步
web:
admin-ui:
url: http://dashboard.nl.iocoder.cn # Admin 管理后台 UI 的地址