add tmall test command

2025-11-26 09:45:30 +08:00
parent 11e2ea1181
commit a1ed0f2ecc
6 changed files with 202 additions and 133 deletions
@@ -0,0 +1,92 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Command;
+
+use Hyperf\Command\Command as HyperfCommand;
+use Hyperf\Command\Annotation\Command;
+use Psr\Container\ContainerInterface;
+use App\Platform\OrderConsumer;
+use Hyperf\Amqp\ConnectionFactory;
+use Hyperf\Amqp\Result;
+use PhpAmqpLib\Message\AMQPMessage;
+use Hyperf\Di\Annotation\Inject;
+
+#[Command]
+class AppMessageQueuePullTmall extends HyperfCommand
+{
+
+    #[Inject]
+    private OrderConsumer $orderConsumer;
+
+    public function __construct(protected ContainerInterface $container)
+    {
+        parent::__construct('app:mq-pull:tmall');
+    }
+
+    public function configure()
+    {
+        parent::configure();
+        $this->setDescription('Manually pull and consume one message from orders.queue');
+    }
+
+    public function handle()
+    {
+        $connectionFactory = $this->container->get(ConnectionFactory::class);
+        $connection = $connectionFactory->getConnection('default_consumer');
+        $channel = $connection->getChannel();
+
+        $queueName = 'orders.queue';
+
+        $this->info("Pulling one message from queue: {$queueName}");
+
+        // 使用 basic_get 手动拉取一条消息（非阻塞）
+        $message = $channel->basic_get($queueName, false);
+
+        if ($message === null) {
+            $this->warn('No messages available in queue');
+            return 0;
+        }
+
+        try {
+            // 解析消息体
+            $data = json_decode($message->body, true);
+
+            $this->info('Received message:');
+            $this->line(json_encode($data, JSON_PRETTY_PRINT | JSON_UNESCAPED_UNICODE));
+
+            // 创建消费者实例并处理消息
+            $result = $this->orderConsumer->consumeMessage($data, $message);
+
+            // 根据消费结果进行 ACK/NACK
+            switch ($result) {
+                case Result::ACK:
+                    $channel->basic_ack($message->getDeliveryTag());
+                    $this->info('Message consumed successfully (ACK)');
+                    break;
+                case Result::NACK:
+                    $channel->basic_nack($message->getDeliveryTag(), false, true);
+                    $this->warn('Message consumption failed (NACK - requeue)');
+                    break;
+                case Result::DROP:
+                    $channel->basic_nack($message->getDeliveryTag(), false, false);
+                    $this->warn('Message dropped (NACK - no requeue)');
+                    break;
+                case Result::REQUEUE:
+                    $channel->basic_nack($message->getDeliveryTag(), false, true);
+                    $this->warn('Message requeued');
+                    break;
+            }
+
+            return 0;
+        } catch (\Throwable $e) {
+            $this->error('Error consuming message: ' . $e->getMessage());
+            $this->error($e->getTraceAsString());
+
+            // 发生异常时，NACK 并重新入队
+            $channel->basic_nack($message->getDeliveryTag(), false, true);
+            return 1;
+        }
+    }
+}