huahuibao/console/controllers/RabbitmqConsumerController.php

<?php
namespace console\controllers;

use yii\console\Controller;
use yii\console\ExitCode;
use Yii;

/**
 * RabbitMQ 消费者控制器
 * 
 * 用途: 运行 RabbitMQ 队列消费者，持续监听并处理消息
 * 
 * 用法示例:
 *  ./yii rabbitmq-consumer/custom     - 运行客户操作消费者
 *  ./yii rabbitmq-consumer/stock      - 运行库存管理消费者
 *  ./yii rabbitmq-consumer/notify     - 运行消息通知消费者
 *  ./yii rabbitmq-consumer/all        - 运行所有消费者（需要多个终端或后台进程）
 *  ./yii rabbitmq-consumer/status     - 查看消费者状态
 * 
 * 使用 supervisor 管理进程示例:
 * 配置文件位置: /etc/supervisor/conf.d/rabbitmq.conf
 */
class RabbitmqConsumerController extends Controller
{
    /**
     * 超时时间（秒），0 表示不超时
     */
    public $timeout = 0;

    /**
     * 获取命令描述
     */
    public function getHelp()
    {
        return <<<EOT
RabbitMQ 消费者管理命令

使用方法:
  ./yii rabbitmq-consumer/custom     运行客户操作消费者
  ./yii rabbitmq-consumer/stock      运行库存管理消费者
  ./yii rabbitmq-consumer/notify     运行消息通知消费者
  ./yii rabbitmq-consumer/all        运行所有消费者（开发环境使用）
  ./yii rabbitmq-consumer/status     查看消费者状态

选项:
  --timeout=N        设置消费超时时间（秒），默认不超时

示例:
  ./yii rabbitmq-consumer/custom --timeout=3600
  ./yii rabbitmq-consumer/stock
  ./yii rabbitmq-consumer/notify

生产环境部署:
  建议使用 supervisor 或其他进程管理工具管理消费者进程，
  参考 supervisor 配置文件示例。

EOT;
    }

    /**
     * 获取命令选项
     */
    public function options($actionID)
    {
        return ['timeout'];
    }

    /**
     * 运行客户操作消费者
     * 处理: 创建客户、更新客户、删除客户等业务操作
     * 
     * @return int 退出码
     */
    public function actionCustom()
    {
        return $this->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;
        }
    }
}