runConsumer('customConsumer', 'customQueue', 'custom'); } /** * 运行库存管理消费者 * 处理: 库存增加、库存扣减、库存调整等业务操作 * * @return int 退出码 */ public function actionStock() { return $this->runConsumer('stockConsumer', 'stockQueue', 'stock'); } /** * 运行消息通知消费者 * 处理: 新订单、新客户、充值、销账、配送状态等通知 * * @return int 退出码 */ public function actionNotify() { return $this->runConsumer('notifyConsumer', 'notifyQueue', 'notify'); } /** * 运行所有消费者(开发环境或测试环境使用) * * 注意: 生产环境建议为每个消费者启动独立进程 * * @return int 退出码 */ public function actionAll() { $this->stdout("启动所有 RabbitMQ 消费者\n"); $this->stdout("提示: 建议使用 supervisor 管理多个消费者进程\n\n"); $consumers = [ ['name' => 'customConsumer', 'queue' => 'customQueue', 'type' => 'custom'], ['name' => 'stockConsumer', 'queue' => 'stockQueue', 'type' => 'stock'], ['name' => 'notifyConsumer', 'queue' => 'notifyQueue', 'type' => 'notify'], ]; foreach ($consumers as $consumer) { $this->stdout("启动消费者: {$consumer['name']}\n"); $this->runConsumer($consumer['name'], $consumer['queue'], $consumer['type']); } return ExitCode::OK; } /** * 查看消费者状态 * * @return int 退出码 */ public function actionStatus() { $this->stdout("RabbitMQ 消费者状态\n"); $this->stdout(str_repeat('=', 50) . "\n"); try { // 获取 RabbitMQ 配置 $rabbitmqConfig = require(Yii::getAlias('@common/config/rabbitMQ.php')); // 显示消费者信息 if (isset($rabbitmqConfig['consumers'])) { $this->stdout("\n已配置的消费者:\n"); foreach ($rabbitmqConfig['consumers'] as $index => $consumer) { $name = $consumer['name'] ?? 'unknown'; $callbacks = $consumer['callbacks'] ?? []; $queueNames = array_keys($callbacks); $this->stdout(" [{$index}] 消费者名称: {$name}\n"); foreach ($queueNames as $queueName) { $this->stdout(" └─ 队列: {$queueName}\n"); } } } // 显示队列信息 if (isset($rabbitmqConfig['queues'])) { $this->stdout("\n已配置的队列:\n"); foreach ($rabbitmqConfig['queues'] as $index => $queue) { $name = $queue['name'] ?? 'unknown'; $durable = $queue['durable'] ?? false ? '✓' : '✗'; $this->stdout(" [{$index}] 队列名称: {$name} (持久化: {$durable})\n"); } } $this->stdout("\n" . str_repeat('=', 50) . "\n"); $this->stdout("提示: 使用 'rabbitmqctl list_consumers' 查看 RabbitMQ 中的实时消费者\n"); return ExitCode::OK; } catch (\Exception $e) { $this->stderr("错误: " . $e->getMessage() . "\n"); return ExitCode::UNSPECIFIED_ERROR; } } /** * 运行消费者的核心方法 * * @param string $consumerName 消费者名称 * @param string $queueName 队列名称 * @param string $type 消费者类型(用于日志区分) * @return int 退出码 */ private function runConsumer($consumerName, $queueName, $type) { try { $this->stdout("启动 {$type} 消费者: {$consumerName}\n"); $this->stdout("监听队列: {$queueName}\n"); $this->stdout("按 Ctrl+C 停止消费者\n\n"); // 获取消费者实例 $consumer = Yii::$app->rabbitmq->getConsumer($consumerName); if (!$consumer) { $this->stderr("错误: 无法获取消费者 '{$consumerName}'\n"); return ExitCode::CONFIG; } // 运行消费者(阻塞式) Yii::info("Consumer started: {$consumerName} on queue: {$queueName}", "rabbitmq.{$type}"); $consumer->consume(); return ExitCode::OK; } catch (\Exception $e) { $this->stderr("消费者异常: " . $e->getMessage() . "\n"); Yii::error("Consumer exception: " . $e->getMessage(), "rabbitmq.{$type}"); return ExitCode::UNSPECIFIED_ERROR; } } }