126 lines
4.3 KiB
PHP
126 lines
4.3 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 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;
|
|
}
|
|
}
|
|
}
|