如何在CodeIgniter框架中集成RabbitMQ实现异步任务处理

在CodeIgniter框架开发中，同步执行耗时任务会导致接口响应缓慢，影响用户体验，引入RabbitMQ消息队列实现异步任务处理可以有效解决这个问题。通过生产者将任务投递到队列，消费者异步处理任务，主流程无需等待任务完成即可返回结果。

环境准备

在开始集成之前，需要准备好对应的运行环境：

• 已部署完成的CodeIgniter 3.x或4.x框架项目
• 安装并启动RabbitMQ服务，默认端口为5672
• PHP环境安装php-amqplib扩展，用于PHP与RabbitMQ通信

安装php-amqplib扩展

如果使用Composer管理依赖，直接在项目根目录执行以下命令安装扩展：

composer require php-amqplib/php-amqplib

安装完成后，Composer会自动将扩展加载到项目中，无需额外配置自动加载。

CodeIgniter配置RabbitMQ

在CodeIgniter中创建RabbitMQ配置文件，方便统一管理连接参数。以CodeIgniter 4为例，在app/Config目录下创建RabbitMQ.php文件：

<?php
namespace Config;

use CodeIgniterConfigBaseConfig;

class RabbitMQ extends BaseConfig
{
    // RabbitMQ服务地址
    public $host = '127.0.0.1';
    // 端口
    public $port = 5672;
    // 用户名
    public $user = 'guest';
    // 密码
    public $password = 'guest';
    // 虚拟机，默认/
    public $vhost = '/';
    // 队列名称
    public $queueName = 'ci_async_queue';
}

编写生产者代码

生产者负责将需要异步处理的任务投递到RabbitMQ队列中，我们可以创建一个公共的服务类来封装生产者逻辑。在app/Services目录下创建RabbitMQProducer.php：

<?php
namespace AppServices;

use PhpAmqpLibConnectionAMQPStreamConnection;
use PhpAmqpLibMessageAMQPMessage;
use ConfigRabbitMQ;

class RabbitMQProducer
{
    private $connection;
    private $channel;
    private $config;

    public function __construct()
    {
        $this->config = new RabbitMQ();
        // 创建RabbitMQ连接
        $this->connection = new AMQPStreamConnection(
            $this->config->host,
            $this->config->port,
            $this->config->user,
            $this->config->password,
            $this->config->vhost
        );
        // 创建通道
        $this->channel = $this->connection->channel();
        // 声明队列，确保队列存在
        $this->channel->queue_declare(
            $this->config->queueName,
            false,
            true,  // 队列持久化
            false,
            false
        );
    }

    /**
     * 投递任务到队列
     * @param array $taskData 任务数据
     * @return bool
     */
    public function pushTask(array $taskData): bool
    {
        $msg = new AMQPMessage(
            json_encode($taskData),
            ['delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT]  // 消息持久化
        );
        $this->channel->basic_publish($msg, '', $this->config->queueName);
        return true;
    }

    public function __destruct()
    {
        // 关闭通道和连接
        if ($this->channel) {
            $this->channel->close();
        }
        if ($this->connection) {
            $this->connection->close();
        }
    }
}

编写消费者代码

消费者需要持续监听队列，获取任务并执行对应的处理逻辑。可以创建一个命令行脚本作为消费者，在服务器后台运行。在CodeIgniter 4中，创建命令类app/Commands/QueueConsumer.php：

<?php
namespace AppCommands;

use CodeIgniterCLIBaseCommand;
use CodeIgniterCLICLI;
use PhpAmqpLibConnectionAMQPStreamConnection;
use ConfigRabbitMQ;

class QueueConsumer extends BaseCommand
{
    protected $group = 'Queue';
    protected $name = 'queue:consume';
    protected $description = '消费RabbitMQ队列中的异步任务';

    public function run(array $params)
    {
        $config = new RabbitMQ();
        $connection = new AMQPStreamConnection(
            $config->host,
            $config->port,
            $config->user,
            $config->password,
            $config->vhost
        );
        $channel = $connection->channel();
        $channel->queue_declare($config->queueName, false, true, false, false);

        CLI::write('队列消费者启动，等待任务投递...', 'green');

        $callback = function ($msg) {
            CLI::write('收到新任务: ' . $msg->body, 'yellow');
            $taskData = json_decode($msg->body, true);
            // 根据任务类型执行对应处理逻辑
            $this->handleTask($taskData);
            // 确认消息已消费
            $msg->ack();
        };

        $channel->basic_qos(null, 1, null);  // 公平分发，每次只处理一个任务
        $channel->basic_consume($config->queueName, '', false, false, false, false, $callback);

        // 持续监听队列
        while ($channel->is_consuming()) {
            $channel->wait();
        }

        $channel->close();
        $connection->close();
    }

    /**
     * 处理具体任务逻辑
     * @param array $taskData
     */
    private function handleTask(array $taskData)
    {
        $taskType = $taskData['type'] ?? '';
        switch ($taskType) {
            case 'send_email':
                // 执行发送邮件逻辑
                CLI::write('执行发送邮件任务，收件人: ' . ($taskData['email'] ?? ''), 'cyan');
                // 模拟耗时操作
                sleep(2);
                break;
            case 'generate_report':
                // 执行生成报表逻辑
                CLI::write('执行生成报表任务，报表ID: ' . ($taskData['report_id'] ?? ''), 'cyan');
                sleep(3);
                break;
            default:
                CLI::write('未知任务类型: ' . $taskType, 'red');
        }
        CLI::write('任务处理完成', 'green');
    }
}

使用示例

在控制器中调用生产者投递异步任务，比如用户注册后发送欢迎邮件的场景：

<?php
namespace AppControllers;

use AppServicesRabbitMQProducer;
use CodeIgniterController;

class User extends Controller
{
    public function register()
    {
        // 模拟用户注册逻辑
        $userId = 1001;
        $userEmail = 'test@ipipp.com';

        // 投递发送邮件的异步任务
        $producer = new RabbitMQProducer();
        $producer->pushTask([
            'type' => 'send_email',
            'email' => $userEmail,
            'subject' => '欢迎注册',
            'content' => '亲爱的用户，欢迎注册我们的平台',
            'user_id' => $userId
        ]);

        // 主流程直接返回注册成功结果，无需等待邮件发送完成
        return $this->response->setJSON(['code' => 0, 'msg' => '注册成功']);
    }
}

消费者启动与守护

在服务器上启动消费者命令，使用nohup让进程在后台持续运行：

nohup php spark queue:consume > /tmp/queue_consume.log 2>&1 &

如果需要保证消费者进程稳定运行，可以使用supervisor等进程管理工具监控消费者进程，进程意外退出时自动重启。

注意事项

• 队列和消息都建议开启持久化，避免RabbitMQ重启后任务丢失
• 消费者处理任务时要做好异常捕获，避免单个任务失败导致整个消费者进程退出
• 根据任务耗时合理设置消费者的并发数量，避免服务器资源占用过高
• 生产环境不要使用guest默认用户，需要创建专属的RabbitMQ用户并分配最小权限

CodeIgniterRabbitMQ异步任务处理消息队列修改时间：2026-07-15 13:54:50