datahub/backend/app/Command/AppMqClear.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 PhpAmqpLib\Connection\AMQPStreamConnection;
use Hyperf\Contract\ConfigInterface;
use Symfony\Component\Console\Input\InputArgument;

#[Command]
class AppMqClear extends HyperfCommand
{
    public function __construct(protected ContainerInterface $container)
    {
        parent::__construct('app:mq:clear');
    }

    public function configure()
    {
        parent::configure();
        $this->setDescription('Clear all messages from a specified queue (development mode only)');
        $this->addArgument('queue', InputArgument::REQUIRED, 'The queue name to clear');
        $this->addOption('force', 'f', null, 'Force clear without confirmation');
    }

    public function handle()
    {
        $queueName = $this->input->getArgument('queue');
        $force = $this->input->getOption('force');

        $this->warn("You are about to clear all messages from queue: {$queueName}");

        // 如果不是强制模式，需要确认
        if (!$force) {
            $confirm = $this->confirm('Are you sure you want to continue?', false);
            if (!$confirm) {
                $this->info('Operation cancelled.');
                return 0;
            }
        }

        try {
            $config = $this->container->get(ConfigInterface::class);
            $consumerConfig = $config->get('amqp.default_consumer');

            // 创建连接
            $connection = new AMQPStreamConnection(
                $consumerConfig['host'],
                $consumerConfig['port'],
                $consumerConfig['user'],
                $consumerConfig['password'],
                $consumerConfig['vhost'],
                false,
                'AMQPLAIN',
                null,
                'en_US',
                $consumerConfig['params']['connection_timeout'] ?? 3.0,
                $consumerConfig['params']['read_write_timeout'] ?? 3.0,
                null,
                $consumerConfig['params']['keepalive'] ?? false,
                $consumerConfig['params']['heartbeat'] ?? 0
            );

            $channel = $connection->channel();

            // 先检查队列是否存在并获取当前消息数
            try {
                [$queue, $messageCount, $consumerCount] = $channel->queue_declare(
                    $queueName,
                    true,  // passive - 只检查，不创建
                    true,  // durable
                    false, // exclusive
                    false  // auto_delete
                );

                $this->line("Queue '{$queueName}' has {$messageCount} messages before clearing.", 'info');

                if ($messageCount === 0) {
                    $this->info('Queue is already empty. Nothing to clear.');
                    $channel->close();
                    $connection->close();
                    return 0;
                }

                // 清除队列中的所有消息
                $channel->queue_purge($queueName);

                // 再次检查确认已清除
                [$queue, $remainingCount, $consumerCount] = $channel->queue_declare(
                    $queueName,
                    true,  // passive
                    true,  // durable
                    false, // exclusive
                    false  // auto_delete
                );

                $this->line('');
                $this->info("Successfully cleared {$messageCount} messages from queue '{$queueName}'.");
                $this->line("Remaining messages: {$remainingCount}", 'comment');

            } catch (\PhpAmqpLib\Exception\AMQPProtocolChannelException $e) {
                $this->error("Queue '{$queueName}' does not exist.");
                $this->line('Available queues can be found using: php bin/hyperf.php app:mq:status', 'comment');
                $channel->close();
                $connection->close();
                return 1;
            }

            // 关闭连接
            $channel->close();
            $connection->close();

            return 0;

        } catch (\Exception $e) {
            $this->error('Failed to clear queue: ' . $e->getMessage());
            $this->line('Trace: ' . $e->getTraceAsString(), 'comment');
            return 1;
        }
    }
}