93 lines
2.9 KiB
PHP
93 lines
2.9 KiB
PHP
<?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;
|
|
}
|
|
}
|
|
}
|