implement FailedMessageTrait

2026-03-17 10:36:38 +08:00
parent 20fe8ccf5e
commit fcd376cab4
4 changed files with 114 additions and 295 deletions
@@ -5,18 +5,15 @@ declare(strict_types=1);
 namespace App\Platform;

 use App\Entity\Parse\EntityParseFactory;
-use App\Model\FailedMessage;
+use App\Platform\Traits\FailedMessageTrait;
 use App\Utils\Log;
 use Hyperf\Amqp\Annotation\Consumer;
 use Hyperf\Amqp\Builder\QueueBuilder;
 use Hyperf\Amqp\Message\ConsumerMessage;
 use Hyperf\Amqp\Result;
-use Hyperf\Amqp\Producer;
 use PhpAmqpLib\Message\AMQPMessage;
 use Hyperf\DbConnection\Db;
 use App\Model\OrderItem;
-use Exception;
-use Hyperf\Context\ApplicationContext;
 use Throwable;

 use function Hyperf\Support\env;
@@ -24,6 +21,7 @@ use function Hyperf\Support\env;
 #[Consumer(exchange: "main.exchange", routingKey: "order.#", queue: "orders.queue", pool: "default_consumer", nums: 1, enable: true)]
 class OrderConsumer extends ConsumerMessage
 {
+    use FailedMessageTrait;

    protected ?array $qos = [
        // AMQP 默认并没有实现此配置。
@@ -208,83 +206,6 @@ class OrderConsumer extends ConsumerMessage
        return parent::isEnable();
    }

-    /**
-     * 获取消息的重试次数
-     * 通过检查 x-death header 中的 count 字段
-     *
-     * @param AMQPMessage $message
-     * @return int
-     */
-    protected function getRetryCount(AMQPMessage $message): int
-    {
-        // 检查是否存在 application_headers 属性
-        // 首次处理的消息不会有这个属性，必须先用 has() 检查
-        if (!$message->has('application_headers')) {
-            dump(">>> No application_headers, first time processing");
-            return 0;
-        }
-
-        $headers = $message->get('application_headers');
-        if (!$headers) {
-            dump(">>> application_headers exists but is empty");
-            return 0;
-        }
-
-        $headerData = $headers->getNativeData();
-        $xDeath = $headerData['x-death'] ?? [];
-
-        dump(">>> x-death header data:");
-        dump($xDeath);
-
-        if (empty($xDeath)) {
-            dump(">>> x-death is empty");
-            return 0; // 首次失败
-        }
-
-        // x-death 是一个数组，第一个元素包含 count 字段
-        // count 表示消息从该队列死信的次数
-        $count = $xDeath[0]['count'] ?? 0;
-        dump(">>> Extracted count from x-death: {$count}");
-        return $count;
-    }
-
-    /**
-     * 发送消息到错误队列
-     * 当重试次数超过上限时调用
-     *
-     * @param AMQPMessage $message 原始消息
-     * @param Throwable $error 错误信息
-     * @return void
-     */
-    protected function sendToErrorQueue(AMQPMessage $message, Throwable $error): void
-    {
-        try {
-            $retry_count = $this->getRetryCount($message);
-
-            // 使用 ErrorProducer 发送到错误队列
-            $producer = ApplicationContext::getContainer()->get(Producer::class);
-            $error_producer = new ErrorProducer($message, $error, $retry_count);
-            $producer->produce($error_producer);
-
-            // 同步写入 failed_messages 表
-            $this->persistFailedMessage($error_producer->payload);
-
-            // 记录日志
-            Log::get()->warning('Message sent to error queue after exceeding retry limit', [
-                'error_id' => $error_producer->payload['error_id'] ?? 'unknown',
-                'retry_count' => $retry_count,
-                'error_message' => $error->getMessage(),
-            ]);
-        } catch (Throwable $e) {
-            // 发送到错误队列失败，记录日志
-            dump($e->getMessage());
-            Log::get()->error('Failed to send message to error queue', [
-                'error' => $e->getMessage(),
-                'original_error' => $error->getMessage(),
-            ]);
-        }
-    }
-
    /**
     * 处理订单子项的批量同步（优化版本）
     *
@@ -458,33 +379,4 @@ class OrderConsumer extends ConsumerMessage
        }
    }

-    /**
-     * 持久化失败消息到数据库
-     */
-    protected function persistFailedMessage(array $payload): void
-    {
-        try {
-            FailedMessage::query()->create([
-                'error_id' => $payload['error_id'],
-                'data_type' => $payload['metadata']['data_type'] ?? 'order',
-                'platform' => $payload['metadata']['platform'] ?? null,
-                'platform_id' => $payload['metadata']['platform_id'] ?? null,
-                'company_id' => $payload['metadata']['company_id'] ?? null,
-                'store_id' => $payload['metadata']['store_id'] ?? null,
-                'error_type' => $payload['error']['type'] ?? 'Unknown',
-                'error_message' => $payload['error']['message'] ?? '',
-                'error_code' => $payload['error']['code'] ?? 0,
-                'error_trace' => $payload['error']['trace'] ?? '',
-                'original_message' => $payload['original_message'] ?? [],
-                'retry_count' => $payload['metadata']['retry_count'] ?? 0,
-                'message_id' => $payload['metadata']['message_id'] ?? null,
-                'failed_at' => $payload['metadata']['failed_at'] ?? date('c'),
-            ]);
-        } catch (Throwable $e) {
-            Log::get()->error('Failed to persist failed message to database', [
-                'error' => $e->getMessage(),
-                'error_id' => $payload['error_id'] ?? 'unknown',
-            ]);
-        }
-    }
 }
@@ -5,16 +5,14 @@ declare(strict_types=1);
 namespace App\Platform;

 use App\Entity\Parse\EntityParseFactory;
-use App\Model\FailedMessage;
+use App\Platform\Traits\FailedMessageTrait;
 use App\Utils\Log;
 use Hyperf\Amqp\Annotation\Consumer;
 use Hyperf\Amqp\Builder\QueueBuilder;
 use Hyperf\Amqp\Message\ConsumerMessage;
 use Hyperf\Amqp\Result;
-use Hyperf\Amqp\Producer;
 use PhpAmqpLib\Message\AMQPMessage;
 use Hyperf\DbConnection\Db;
-use Hyperf\Context\ApplicationContext;
 use Throwable;

 use function Hyperf\Support\env;
@@ -22,6 +20,8 @@ use function Hyperf\Support\env;
 #[Consumer(exchange: "main.exchange", routingKey: "product.#", queue: "products.queue", pool: "default_consumer", nums: 1, enable: true)]
 class ProductConsumer extends ConsumerMessage
 {
+    use FailedMessageTrait;
+
    protected ?array $qos = [
        'prefetch_size' => 0,
        'prefetch_count' => 1,
@@ -149,93 +149,4 @@ class ProductConsumer extends ConsumerMessage
        return parent::isEnable();
    }

-    /**
-     * 获取消息的重试次数
-     *
-     * @param AMQPMessage $message
-     * @return int
-     */
-    protected function getRetryCount(AMQPMessage $message): int
-    {
-        if (!$message->has('application_headers')) {
-            return 0;
-        }
-
-        $headers = $message->get('application_headers');
-        if (!$headers) {
-            return 0;
-        }
-
-        $headerData = $headers->getNativeData();
-        $xDeath = $headerData['x-death'] ?? [];
-
-        if (empty($xDeath)) {
-            return 0;
-        }
-
-        return $xDeath[0]['count'] ?? 0;
-    }
-
-    /**
-     * 发送消息到错误队列
-     *
-     * @param AMQPMessage $message 原始消息
-     * @param Throwable $error 错误信息
-     * @return void
-     */
-    protected function sendToErrorQueue(AMQPMessage $message, Throwable $error): void
-    {
-        try {
-            $retry_count = $this->getRetryCount($message);
-
-            $producer = ApplicationContext::getContainer()->get(Producer::class);
-            $error_producer = new ErrorProducer($message, $error, $retry_count);
-            $producer->produce($error_producer);
-
-            // 同步写入 failed_messages 表
-            $this->persistFailedMessage($error_producer->payload);
-
-            Log::get()->warning('Product message sent to error queue after exceeding retry limit', [
-                'error_id' => $error_producer->payload['error_id'] ?? 'unknown',
-                'retry_count' => $retry_count,
-                'error_message' => $error->getMessage(),
-            ]);
-        } catch (Throwable $e) {
-            dump($e->getMessage());
-            Log::get()->error('Failed to send product message to error queue', [
-                'error' => $e->getMessage(),
-                'original_error' => $error->getMessage(),
-            ]);
-        }
-    }
-
-    /**
-     * 持久化失败消息到数据库
-     */
-    protected function persistFailedMessage(array $payload): void
-    {
-        try {
-            FailedMessage::query()->create([
-                'error_id' => $payload['error_id'],
-                'data_type' => $payload['metadata']['data_type'] ?? 'product',
-                'platform' => $payload['metadata']['platform'] ?? null,
-                'platform_id' => $payload['metadata']['platform_id'] ?? null,
-                'company_id' => $payload['metadata']['company_id'] ?? null,
-                'store_id' => $payload['metadata']['store_id'] ?? null,
-                'error_type' => $payload['error']['type'] ?? 'Unknown',
-                'error_message' => $payload['error']['message'] ?? '',
-                'error_code' => $payload['error']['code'] ?? 0,
-                'error_trace' => $payload['error']['trace'] ?? '',
-                'original_message' => $payload['original_message'] ?? [],
-                'retry_count' => $payload['metadata']['retry_count'] ?? 0,
-                'message_id' => $payload['metadata']['message_id'] ?? null,
-                'failed_at' => $payload['metadata']['failed_at'] ?? date('c'),
-            ]);
-        } catch (Throwable $e) {
-            Log::get()->error('Failed to persist failed message to database', [
-                'error' => $e->getMessage(),
-                'error_id' => $payload['error_id'] ?? 'unknown',
-            ]);
-        }
-    }
 }
@@ -5,17 +5,15 @@ declare(strict_types=1);
 namespace App\Platform;

 use App\Entity\Parse\EntityParseFactory;
-use App\Model\FailedMessage;
 use App\Model\RefundItem;
+use App\Platform\Traits\FailedMessageTrait;
 use App\Utils\Log;
 use Hyperf\Amqp\Annotation\Consumer;
 use Hyperf\Amqp\Builder\QueueBuilder;
 use Hyperf\Amqp\Message\ConsumerMessage;
 use Hyperf\Amqp\Result;
-use Hyperf\Amqp\Producer;
 use PhpAmqpLib\Message\AMQPMessage;
 use Hyperf\DbConnection\Db;
-use Hyperf\Context\ApplicationContext;
 use Throwable;

 use function Hyperf\Support\env;
@@ -23,6 +21,7 @@ use function Hyperf\Support\env;
 #[Consumer(exchange: "main.exchange", routingKey: "refund.#", queue: "refunds.queue", pool: "default_consumer", nums: 1, enable: true)]
 class RefundConsumer extends ConsumerMessage
 {
+    use FailedMessageTrait;

    protected ?array $qos = [
        'prefetch_size' => 0,
@@ -191,96 +190,6 @@ class RefundConsumer extends ConsumerMessage
        return parent::isEnable();
    }

-    /**
-     * 获取消息的重试次数
-     *
-     * @param AMQPMessage $message
-     * @return int
-     */
-    protected function getRetryCount(AMQPMessage $message): int
-    {
-        if (!$message->has('application_headers')) {
-            return 0;
-        }
-
-        $headers = $message->get('application_headers');
-        if (!$headers) {
-            return 0;
-        }
-
-        $header_data = $headers->getNativeData();
-        $x_death = $header_data['x-death'] ?? [];
-
-        if (empty($x_death)) {
-            return 0;
-        }
-
-        return $x_death[0]['count'] ?? 0;
-    }
-
-    /**
-     * 发送消息到错误队列
-     *
-     * @param AMQPMessage $message 原始消息
-     * @param Throwable $error 错误信息
-     * @return void
-     */
-    protected function sendToErrorQueue(AMQPMessage $message, Throwable $error): void
-    {
-        try {
-            $retry_count = $this->getRetryCount($message);
-
-            $producer = ApplicationContext::getContainer()->get(Producer::class);
-            $error_producer = new ErrorProducer($message, $error, $retry_count);
-            $producer->produce($error_producer);
-
-            // 同步写入 failed_messages 表
-            $this->persistFailedMessage($error_producer->payload);
-
-            Log::get()->warning('Refund message sent to error queue after exceeding retry limit', [
-                'error_id' => $error_producer->payload['error_id'] ?? 'unknown',
-                'retry_count' => $retry_count,
-                'error_message' => $error->getMessage(),
-            ]);
-        } catch (Throwable $e) {
-            dump($e->getMessage());
-            Log::get()->error('Failed to send refund message to error queue', [
-                'error' => $e->getMessage(),
-                'original_error' => $error->getMessage(),
-            ]);
-        }
-    }
-
-    /**
-     * 持久化失败消息到数据库
-     */
-    protected function persistFailedMessage(array $payload): void
-    {
-        try {
-            FailedMessage::query()->create([
-                'error_id' => $payload['error_id'],
-                'data_type' => $payload['metadata']['data_type'] ?? 'refund',
-                'platform' => $payload['metadata']['platform'] ?? null,
-                'platform_id' => $payload['metadata']['platform_id'] ?? null,
-                'company_id' => $payload['metadata']['company_id'] ?? null,
-                'store_id' => $payload['metadata']['store_id'] ?? null,
-                'error_type' => $payload['error']['type'] ?? 'Unknown',
-                'error_message' => $payload['error']['message'] ?? '',
-                'error_code' => $payload['error']['code'] ?? 0,
-                'error_trace' => $payload['error']['trace'] ?? '',
-                'original_message' => $payload['original_message'] ?? [],
-                'retry_count' => $payload['metadata']['retry_count'] ?? 0,
-                'message_id' => $payload['metadata']['message_id'] ?? null,
-                'failed_at' => $payload['metadata']['failed_at'] ?? date('c'),
-            ]);
-        } catch (Throwable $e) {
-            Log::get()->error('Failed to persist failed message to database', [
-                'error' => $e->getMessage(),
-                'error_id' => $payload['error_id'] ?? 'unknown',
-            ]);
-        }
-    }
-
    /**
     * 处理退款子项的批量同步
     *
@@ -0,0 +1,107 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Platform\Traits;
+
+use App\Model\FailedMessage;
+use App\Platform\ErrorProducer;
+use App\Utils\Log;
+use Hyperf\Amqp\Producer;
+use Hyperf\Context\ApplicationContext;
+use PhpAmqpLib\Message\AMQPMessage;
+use Throwable;
+
+/**
+ * 失败消息处理 Trait
+ *
+ * 提供 Consumer 共用的重试计数、错误队列发送、失败消息持久化能力
+ */
+trait FailedMessageTrait
+{
+    /**
+     * 获取消息的重试次数
+     */
+    protected function getRetryCount(AMQPMessage $message): int
+    {
+        if (!$message->has('application_headers')) {
+            return 0;
+        }
+
+        $headers = $message->get('application_headers');
+        if (!$headers) {
+            return 0;
+        }
+
+        $header_data = $headers->getNativeData();
+        $x_death = $header_data['x-death'] ?? [];
+
+        if (empty($x_death)) {
+            return 0;
+        }
+
+        return $x_death[0]['count'] ?? 0;
+    }
+
+    /**
+     * 发送消息到错误队列
+     * 当重试次数超过上限时调用
+     *
+     * @param AMQPMessage $message 原始消息
+     * @param Throwable $error 错误信息
+     */
+    protected function sendToErrorQueue(AMQPMessage $message, Throwable $error): void
+    {
+        try {
+            $retry_count = $this->getRetryCount($message);
+
+            $producer = ApplicationContext::getContainer()->get(Producer::class);
+            $error_producer = new ErrorProducer($message, $error, $retry_count);
+            $producer->produce($error_producer);
+
+            // 同步写入 failed_messages 表
+            $this->persistFailedMessage($error_producer->payload);
+
+            Log::get()->warning('Message sent to error queue after exceeding retry limit', [
+                'error_id' => $error_producer->payload['error_id'] ?? 'unknown',
+                'retry_count' => $retry_count,
+                'error_message' => $error->getMessage(),
+            ]);
+        } catch (Throwable $e) {
+            Log::get()->error('Failed to send message to error queue', [
+                'error' => $e->getMessage(),
+                'original_error' => $error->getMessage(),
+            ]);
+        }
+    }
+
+    /**
+     * 持久化失败消息到数据库
+     */
+    protected function persistFailedMessage(array $payload): void
+    {
+        try {
+            FailedMessage::query()->create([
+                'error_id' => $payload['error_id'],
+                'data_type' => $payload['metadata']['data_type'] ?? 'unknown',
+                'platform' => $payload['metadata']['platform'] ?? null,
+                'platform_id' => $payload['metadata']['platform_id'] ?? null,
+                'company_id' => $payload['metadata']['company_id'] ?? null,
+                'store_id' => $payload['metadata']['store_id'] ?? null,
+                'error_type' => $payload['error']['type'] ?? 'Unknown',
+                'error_message' => $payload['error']['message'] ?? '',
+                'error_code' => $payload['error']['code'] ?? 0,
+                'error_trace' => $payload['error']['trace'] ?? '',
+                'original_message' => $payload['original_message'] ?? [],
+                'retry_count' => $payload['metadata']['retry_count'] ?? 0,
+                'message_id' => $payload['metadata']['message_id'] ?? null,
+                'failed_at' => $payload['metadata']['failed_at'] ?? date('c'),
+            ]);
+        } catch (Throwable $e) {
+            Log::get()->error('Failed to persist failed message to database', [
+                'error' => $e->getMessage(),
+                'error_id' => $payload['error_id'] ?? 'unknown',
+            ]);
+        }
+    }
+}