Yii框架如何使用队列实现RabbitMQ消息驱动

在Yii框架中集成RabbitMQ实现消息驱动，需要完成组件安装、配置定义、生产者编写、消费者编写等多个步骤，整个流程适配Yii的扩展机制，开发和维护成本较低。

环境准备

首先需要确保服务器已经安装RabbitMQ服务，并且PHP环境安装了amqp扩展，可以通过php -m命令查看扩展是否生效。如果未安装，可以根据系统版本执行对应的安装命令，比如Ubuntu系统可以通过apt-get install php-amqp完成安装，安装后重启PHP服务使扩展生效。

接着在Yii项目中通过Composer安装RabbitMQ的PHP客户端库，执行以下命令：

composer require php-amqplib/php-amqplib

配置RabbitMQ组件

在Yii的配置文件（通常是config/web.php或者config/console.php）中定义RabbitMQ的连接组件，方便后续在代码中调用。配置内容如下：

<?php
return [
    // 其他配置项
    'components' => [
        'rabbitmq' => [
            'class' => 'yiibaseComponent',
            'host' => '127.0.0.1',
            'port' => 5672,
            'user' => 'guest',
            'password' => 'guest',
            'vhost' => '/',
        ],
    ],
];

编写生产者代码

生产者负责将消息发送到RabbitMQ的指定队列中，我们可以创建一个控制台命令类来实现生产者逻辑，方便通过命令行触发消息发送。创建commands/RabbitmqController.php文件，内容如下：

<?php
namespace appcommands;

use yiiconsoleController;
use PhpAmqpLibConnectionAMQPStreamConnection;
use PhpAmqpLibMessageAMQPMessage;

class RabbitmqController extends Controller
{
    // 发送消息到队列
    public function actionSend($msg = 'default message')
    {
        $config = Yii::$app->rabbitmq;
        // 创建连接
        $connection = new AMQPStreamConnection(
            $config->host,
            $config->port,
            $config->user,
            $config->password,
            $config->vhost
        );
        $channel = $connection->channel();
        // 声明队列，队列名是test_queue，持久化设置为true
        $channel->queue_declare('test_queue', false, true, false, false);
        // 创建消息，设置投递模式为2表示消息持久化
        $amqpMsg = new AMQPMessage($msg, ['delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT]);
        // 发送消息到队列
        $channel->basic_publish($amqpMsg, '', 'test_queue');
        echo "消息发送成功：{$msg}n";
        // 关闭通道和连接
        $channel->close();
        $connection->close();
    }
}

编写消费者代码

消费者负责监听队列，接收并处理生产者发送的消息，同样可以创建控制台命令来实现。在RabbitmqController.php中添加消费者方法：

<?php
// 省略之前的命名空间和use部分
class RabbitmqController extends Controller
{
    // 之前的actionSend方法

    // 消费队列消息
    public function actionConsume()
    {
        $config = Yii::$app->rabbitmq;
        $connection = new AMQPStreamConnection(
            $config->host,
            $config->port,
            $config->user,
            $config->password,
            $config->vhost
        );
        $channel = $connection->channel();
        $channel->queue_declare('test_queue', false, true, false, false);
        echo "等待接收消息，退出请按Ctrl+Cn";
        // 定义消息回调函数
        $callback = function ($msg) {
            echo "接收到消息：{$msg->body}n";
            // 模拟业务处理，比如发送邮件、生成报表等
            sleep(1);
            // 手动确认消息已处理
            $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
        };
        // 设置公平分发，每次只接收一条未确认的消息
        $channel->basic_qos(null, 1, null);
        // 开始消费，关闭自动确认
        $channel->basic_consume('test_queue', '', false, false, false, false, $callback);
        // 循环监听消息
        while ($channel->is_consuming()) {
            $channel->wait();
        }
        $channel->close();
        $connection->close();
    }
}

测试验证

打开两个终端，第一个终端执行生产者命令发送消息：

php yii rabbitmq/send "这是一条测试消息"

第二个终端执行消费者命令监听队列：

php yii rabbitmq/consume

可以看到消费者终端会输出接收到的消息内容，说明队列消息驱动已经正常工作。

注意事项

• 生产环境不要使用guest默认账号，需要创建专用账号并分配最小权限，提升安全性。
• 队列和消息的持久化配置需要根据业务需求调整，避免消息丢失但也不要过度消耗资源。
• 消费者处理消息时如果出现异常，需要做好重试或者消息重新入队的机制，避免消息丢失。
• 如果队列消息量较大，可以启动多个消费者进程提升处理效率，RabbitMQ会自动分发消息到各个消费者。

YiiRabbitMQqueuemessage_queue修改时间：2026-07-15 20:42:21