feat(TopicTaskMessageSubscriber): update queue parameters to allow 3 message processing and add retry mechanism for lock acquisition

Author: Rock-520

backend/super-magic-module/src/Application/SuperAgent/Event/Subscribe/TopicTaskMessageSubscriber.php

@@ -23,10 +23,10 @@
  * 话题任务消息订阅者.
  */
 #[Consumer(
-    exchange: 'super_magic_topic_task_message', 
-    routingKey: 'super_magic_topic_task_message', 
-    queue: 'super_magic_topic_task_message', 
-    nums: 1
+    exchange: 'super_magic_topic_task_message',
+    routingKey: 'super_magic_topic_task_message',
+    queue: 'super_magic_topic_task_message',
+    nums: 3
 )]
 class TopicTaskMessageSubscriber extends ConsumerMessage
 {
@@ -77,7 +77,6 @@ public function consumeMessage($data, AMQPMessage $message): Result
             $applicationHeaders = $messageProperties['application_headers'] ?? new AMQPTable([]);
             // 直接从原生数据中获取，如果不存在则为 null
             $originalTimestampFromHeader = $applicationHeaders->getNativeData()['x-original-timestamp'] ?? null;
-            
             $currentTimeForLog = time(); // 当前处理时间，主要用于日志和可能的本地逻辑
             $actualOriginalTimestamp = null; // 初始化变量以避免 linter 警告
 
@@ -91,11 +90,11 @@ public function consumeMessage($data, AMQPMessage $message): Result
                     '消息未找到 x-original-timestamp 头部，将使用当前时间作为本次处理的原始时间戳参考: %d (%s). 请确保生产者已设置此头部. Message ID: %s',
                     $actualOriginalTimestamp,
                     date('Y-m-d H:i:s', $actualOriginalTimestamp),
-                    $data['payload']['message_id'] ?? 'N/A' 
+                    $data['payload']['message_id'] ?? 'N/A'
                 ));
                 // 不再尝试修改消息的 application_headers，因为这对于 REQUEUE 后的消息通常无效
             }
-            
+
             // 验证消息格式
             $this->validateMessageFormat($data);
 
@@ -104,24 +103,24 @@ public function consumeMessage($data, AMQPMessage $message): Result
 
             // 创建DTO
             $messageDTO = TopicTaskMessageDTO::fromArray($data);
-            
+
             // 获取sandboxId用于锁定
             $sandboxId = $messageDTO->getMetadata()?->getSandboxId();
             if (empty($sandboxId)) {
                 $this->logger->warning('缺少有效的sandboxId，无法加锁保证消息顺序性，将直接处理消息', [
                     'message_id' => $messageDTO->getPayload()?->getMessageId(),
                     'message' => json_encode($data, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES),
                 ]);
-                return Result::ACK; 
+                return Result::ACK;
             }
 
             // 尝试获取锁
             $lockKey = 'handle_sandbox_message_lock:' . $sandboxId;
             $lockOwner = IdGenerator::getUniqueId32();
-            $lockExpireSeconds = 30; 
-            
+            $lockExpireSeconds = 30;
+
             $lockAcquired = (bool) $this->superAgentAppService->acquireLock($lockKey, $lockOwner, $lockExpireSeconds);
-            
+
             if (!$lockAcquired) {
                 $this->logger->info(sprintf(
                     '无法获取sandbox %s的锁，该sandbox可能有其他消息正在处理中，将消息重新入队等待处理，原始接收秒级时间: %d (%s), message_id: %s',
@@ -132,7 +131,6 @@ public function consumeMessage($data, AMQPMessage $message): Result
                 ));
                 return Result::REQUEUE;
             }
-            
             $this->logger->info(sprintf(
                 '已获取sandbox %s的锁，持有者: %s，开始处理消息，原始接收秒级时间: %d (%s), message_id: %s',
                 $sandboxId,
@@ -141,7 +139,6 @@ public function consumeMessage($data, AMQPMessage $message): Result
                 date('Y-m-d H:i:s', $actualOriginalTimestamp),
                 $messageDTO->getPayload()?->getMessageId()
             ));
-            
             try {
                 $this->superAgentAppService->handleTopicTaskMessage($messageDTO);
                 return Result::ACK;
@@ -322,4 +319,69 @@ private function logMessageDetails(array $data): void
             }
         }
     }
+
+    /**
+     * 带重试机制的获取锁.
+     *
+     * @param string $sandboxId 沙箱ID
+     * @param TopicTaskMessageDTO $messageDTO 消息DTO
+     * @return array 包含锁信息的数组，acquired表示是否成功获取锁，owner表示锁的持有者，key表示锁的键名
+     */
+    private function acquireLockWithRetry(?string $sandboxId, TopicTaskMessageDTO $messageDTO): array
+    {
+        $result = [
+            'acquired' => false,
+            'owner' => '',
+            'key' => '',
+        ];
+
+        if (empty($sandboxId)) {
+            return $result;
+        }
+
+        $lockKey = 'handle_sandbox_message_lock:' . $sandboxId;
+        $lockOwner = IdGenerator::getUniqueId32(); // 使用唯一ID作为锁持有者标识
+        $lockExpireSeconds = 30; // 锁的过期时间（秒），消息处理可能需要更长时间
+
+        $maxRetries = 3;
+        $retryCount = 0;
+        $baseWaitTime = 1; // 基础等待时间（秒）
+
+        while ($retryCount <= $maxRetries) {
+            $lockAcquired = (bool) $this->superAgentAppService->acquireLock($lockKey, $lockOwner, $lockExpireSeconds);
+
+            if ($lockAcquired) {
+                $this->logger->debug(sprintf('已获取sandbox %s的锁，持有者: %s', $sandboxId, $lockOwner));
+                $result['acquired'] = true;
+                $result['owner'] = $lockOwner;
+                $result['key'] = $lockKey;
+                return $result;
+            }
+
+            if ($retryCount === $maxRetries) {
+                $this->logger->error(sprintf(
+                    '在重试%d次后仍无法获取sandbox %s的锁，该sandbox可能有其他消息正在处理中，message_id: %s',
+                    $maxRetries,
+                    $sandboxId,
+                    $messageDTO->getPayload()?->getMessageId()
+                ));
+                // 可以选择将消息重新入队或实现延迟重试策略
+                return $result;
+            }
+
+            $waitTime = $baseWaitTime * pow(2, $retryCount); // 指数退避
+            $this->logger->warning(sprintf(
+                '无法获取sandbox %s的锁，该sandbox可能有其他消息正在处理中，message_id: %s，将在%d秒后进行第%d次重试',
+                $sandboxId,
+                $messageDTO->getPayload()?->getMessageId(),
+                $waitTime,
+                $retryCount + 1
+            ));
+
+            sleep($waitTime);
+            $retryCount++;
+        }
+
+        return $result;
+    }
 }