fix: 任务并发控制

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

@@ -1,15 +1,23 @@
 package cn.code.nl.module.lms.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 定位任务子类并执行完成/取消逻辑
  *
@@ -27,12 +35,68 @@ public class LmsTaskStatusChangeConsumer implements RocketMQListener<TaskEventMe
     @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);

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

@@ -3,9 +3,9 @@
 rocketmq:
   name-server: 192.168.81.193:9876
   producer:
-    group: lms-producer-dev-group # 事务消息需要配置一样
+    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
+      group: lms_task_status_change_dev_group
+      topic: lms_task_status_change_dev_topic