update consumer and producer

2025-11-26 09:44:55 +08:00
parent 40c9c69b3b
commit 11e2ea1181
2 changed files with 152 additions and 0 deletions
@@ -0,0 +1,38 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Platform;
+
+use Exception;
+use Hyperf\Amqp\Annotation\Consumer;
+use Hyperf\Amqp\Message\ConsumerMessage;
+use Hyperf\Amqp\Result;
+use PhpAmqpLib\Message\AMQPMessage;
+
+#[Consumer(exchange: "main.exchange", routingKey: "order.#", queue: "orders.queue", nums: 1, enable: true)]
+class OrderConsumer extends ConsumerMessage
+{
+
+    protected ?array $qos = [
+        // AMQP 默认并没有实现此配置。
+        'prefetch_size' => 0,
+        // 同一个消费者，最高同时可以处理的消息数。
+        'prefetch_count' => 100,
+        // 因为 Hyperf 默认一个 Channel 只消费一个 队列，所以 global 设置为 true/false 效果是一样的。
+        'global' => false,
+    ];
+
+    protected $entityType = 'order';
+
+    public function consumeMessage($data, AMQPMessage $message): Result
+    {
+        dump($data);
+        return Result::NACK;
+    }
+
+    public function isEnable(): bool
+    {
+        return parent::isEnable();
+    }
+}
@@ -0,0 +1,114 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Platform;
+
+use Exception;
+use Hyperf\Amqp\Annotation\Producer;
+use Hyperf\Amqp\Message\ProducerMessage;
+
+// exchange 示例值 'tmall.exchange', routeKey 示例值 'order.tmall'
+#[Producer('', '')]
+class OrderProducer extends ProducerMessage
+{
+    /**
+     * 指定使用的连接池名称
+     */
+    protected string $poolName = 'default_producer';
+
+    /**
+     * VHost
+     */
+    protected string $vhost = 'dataflow';
+
+    /**
+     * 消息持久化
+     */
+    protected array $properties = [
+        'delivery_mode' => 2, // 持久化消息
+    ];
+
+
+    protected string $exchange = '';
+    protected string $entityType = 'order';
+    protected string|array $routingKey = '';
+
+
+    /**
+     * 构造消息
+     *
+     * @param array $data 订单数据
+     * @return string
+     */
+    public function __construct(array $data = [])
+    {
+        if(empty($this->exchange)){
+            throw new Exception('exchange is not set!');
+        }
+
+        if(empty($this->routingKey)){
+            throw new Exception('routingKey is not set!');
+        }
+
+        if(!empty($this->routingKey)){
+            $this->entityType =  \explode( '.', $this->routingKey)[0];
+        }
+
+        if (!empty($data)) {
+            $this->payload = $this->buildMessage($data);
+        }
+    }
+
+    /**
+     * 构建消息格式
+     *
+     * @param array $data 原始订单数据
+     * @return array
+     */
+    protected function buildMessage(array $data): array
+    {
+        // 根据 RabbitMQ.md 中定义的消息格式规范
+        return [
+            'message_id' => $this->generateMessageId($data),
+            'timestamp' => date('c'), // ISO 8601 格式
+            'platform' => 'tmall',
+            'data_type' => 'order',
+            'metadata' => [
+                'platform_id' => $data['platform_id'] ?? null,
+                'company_id' => $data['company_id'] ?? null,
+                'store_id' => $data['store_id'] ?? null,
+                'source_system' => 'tmall-open-api',
+                'retry_count' => 0,
+                'data_version' => $data['data_version'] ?? time(),
+            ],
+            'data' => $data['raw_data'],
+        ];
+    }
+
+    /**
+     * 生成消息ID
+     *
+     * 格式: {prefix}#{app_id}#{company_id}#{platform_id}#{store_id}#{entity_type}#{entity_id}
+     *
+     * @param array $data
+     * @return string
+     */
+    protected function generateMessageId(array $data): string
+    {
+        $company_id = $data['company_id'];
+        $platform_id = $data['platform_id'];
+        $store_id = $data['store_id'];
+       
+        $unique_id = $data['unique_id'];
+
+        return sprintf(
+            '%s#%s#%s#%s#%s',
+            $company_id,
+            $platform_id,
+            $store_id,
+            $this->entityType,
+            $unique_id
+        );
+    }
+}