*/ private array $queues; public function __construct() { $this->config = config('queue.connections.rabbitmq'); $this->queues = config('mint.rabbitmq.queues', []); $this->connect(); } private function connect(): void { $this->connection = new AMQPStreamConnection( $this->config['host'], $this->config['port'], $this->config['user'], $this->config['password'], $this->config['virtual_host'] ); $this->channel = $this->connection->channel(); // QoS：每次只投递一条未确认消息 $this->channel->basic_qos(null, 1, null); } public function isConnected(): bool { return $this->connection->isConnected(); } public function getChannel(): AMQPChannel { return $this->channel; } /** * 声明配置中的全部队列。 * * @return list 队列名 */ public function createQueue(): array { foreach (array_keys($this->queues) as $name) { $this->setupQueue($name); } return array_keys($this->queues); } /** * 声明一个持久化队列（不存在则创建）。 */ public function setupQueue(string $queueName): void { $queueConfig = $this->queues[$queueName] ?? []; $args = []; if (! empty($queueConfig['max_length'])) { $args['x-max-length'] = (int) $queueConfig['max_length']; } $this->channel->queue_declare( $queueName, false, // passive true, // durable false, // exclusive false, // auto_delete false, // nowait new AMQPTable($args) ); } public function publishMessage(string $queueName, array $data): string { try { $this->setupQueue($queueName); $msgId = (string) Str::uuid(); $message = new AMQPMessage( json_encode($data, JSON_UNESCAPED_UNICODE), [ 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT, 'timestamp' => time(), 'message_id' => $msgId, 'content_type' => 'application/json; charset=utf-8', ] ); $this->channel->basic_publish($message, '', $queueName); Log::info("Message published to queue: {$queueName} msg id={$msgId}"); return $msgId; } catch (\Exception $e) { Log::error("Failed to publish message to queue: {$queueName}", [ 'error' => $e->getMessage(), ]); throw $e; } } /** * 阻塞消费队列。 * * 每条消息处理结束（成功、抛异常、JSON 解析失败）都 ack，不重试。 * 处理完 $loopLimit 条或检测到 .stop 文件后退出；$loopLimit 为 0 表示不限。 * * @param callable(mixed $data, AMQPMessage $message): void $callback * @param int|null $loopLimit 默认取 mint.mq.consumer_loop_limit */ public function consume(string $queueName, callable $callback, ?int $loopLimit = null): void { $this->setupQueue($queueName); $loopLimit ??= (int) config('mint.mq.consumer_loop_limit', 0); $processed = 0; $consumerCallback = function (AMQPMessage $msg) use ($callback, $queueName, $loopLimit, &$processed): void { $messageId = $msg->has('message_id') ? $msg->get('message_id') : null; $context = ['queue' => $queueName, 'message_id' => $messageId]; try { $data = json_decode($msg->getBody(), true, flags: JSON_THROW_ON_ERROR); Log::info('mq message start', $context); $callback($data, $msg); Log::info('mq message done', $context); } catch (\Throwable $e) { Log::error('mq message failed', $context + [ 'error' => $e->getMessage(), 'exception' => $e::class, ]); } finally { $msg->ack(); } $processed++; if (($loopLimit > 0 && $processed >= $loopLimit) || Tools::isStop()) { Log::info('mq consumer exit', $context + ['processed' => $processed]); $msg->getChannel()->basic_cancel($msg->getConsumerTag()); } }; $this->channel->basic_qos(null, 1, null); $this->channel->basic_consume( $queueName, '', // consumer_tag false, // no_local false, // no_ack：手动 ack false, // exclusive false, // nowait $consumerCallback ); Log::info("mq consumer start queue={$queueName}", ['loop_limit' => $loopLimit]); // 阻塞等待，直到回调里 basic_cancel $this->channel->consume(); Log::info("mq consumer stopped queue={$queueName}", ['processed' => $processed]); } public function close(): void { $this->channel->close(); $this->connection->close(); } }