ThinkPHP6怎么集成RabbitMQ实现异步消费？实战代码分享-php教程-PHP中文网

thinkphp6 不内置 rabbitmq 支持，需用 php-amqplib 手动集成；消费者必须为常驻 cli 进程，配置持久化、manual ack、basic_qos，否则消息丢失、重复或连接爆炸。

thinkphp6怎么集成rabbitmq实现异步消费？实战代码分享

直接说结论：ThinkPHP6 本身不内置 RabbitMQ 支持，必须用 php-amqplib 手动集成；消费者不能写在 Web 请求里，必须是独立 CLI 进程 + 手动确认 + 持久化配置，否则消息会丢、重复消费、连接爆炸。

为什么不能把消费者逻辑塞进控制器或命令行命令里？

很多人试过在 app/command/Consume.php 里写个 handle() 方法，然后执行 php think consume —— 表面能跑，实际埋雷：

每次执行都新建一个 TCP 连接，AMQPStreamConnection 实例无法复用，RabbitMQ 很快报 too many connections
没做 basic_qos 限流，消费者一启动就拉走全量积压消息，OOM 或 DB 被打爆
没调 $channel->basic_ack($delivery_info['delivery_tag'])，进程退出后消息自动重回队列，反复重发
没捕获异常并拒绝（basic_nack），失败消息卡死，后续消息被阻塞

消费者必须是常驻守护进程，怎么写？

用 ThinkPHP6 的 think-queue 插件不是最优解——它对 RabbitMQ 封装太薄，不控制 delivery_mode、不暴露 basic_qos、不支持死信绑定。推荐直接用原生 php-amqplib 写一个独立脚本：

新建 app/queue/Consumer.php：

<?php require_once __DIR__ . '/../../vendor/autoload.php';

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

$connection = new AMQPStreamConnection(
    '127.0.0.1', 
    5672, 
    'admin', 
    '123456', 
    '/'
);
$channel = $connection->channel();

// 声明持久化队列（第三个参数 true！）
$channel->queue_declare('order_notify', false, true, false, false);

// 限制消费者一次最多处理 1 条，避免堆积
$channel->basic_qos(null, 1, null);

$callback = function ($msg) use ($channel) {
    try {
        $data = json_decode($msg->body, true);
        // ✅ 真实业务逻辑：发短信、写日志、更新状态...
        file_put_contents('/tmp/consume.log', date('Y-m-d H:i:s') . " processed: {$data['order_id']}\n", FILE_APPEND);

        // ✅ 必须手动确认
        $channel->basic_ack($msg->getDeliveryTag());
    } catch (\Exception $e) {
        // ❌ 不要吞异常！拒绝并重新入队（可加 retry_count 头防无限循环）
        $channel->basic_nack($msg->getDeliveryTag(), false, true);
    }
};

// 开启监听，no_ack=false（禁用自动确认）
$channel->basic_consume('order_notify', '', false, false, false, false, $callback);

while ($channel->is_consuming()) {
    $channel->wait();
}

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

然后用 supervisor 守护它：

btpanel phpsite 宝塔面板PHP网站

宝塔面板 PHP 网站管理：站点创建、删除、启停、PHP 版本切换、域名管理、SSL证书管理、伪静态管理、数据库管理

下载

[program:rabbitmq-consumer]
command=php /var/www/tp6/app/queue/Consumer.php
autostart=true
autorestart=true
user=www-data
redirect_stderr=true
stdout_logfile=/var/log/rabbitmq-consumer.log

生产者发消息时最容易漏的三个参数

只调 basic_publish 不够，关键配置全在 AMQPMessage 构造函数里：

delivery_mode => 2：不加这个，RabbitMQ 重启后消息全丢，等于白搭
content_type => 'application/json'：显式声明类型，避免消费者收到二进制乱码
priority => 10（可选）：配合队列声明时的 x-max-priority，实现高优订单优先处理

正确示例：

$message = new AMQPMessage(
    json_encode(['order_id' => 'ORD123456', 'mobile' => '13800138000']), 
    [
        'delivery_mode' => 2,
        'content_type'  => 'application/json',
        'priority'      => 10
    ]
);
$channel->basic_publish($message, '', 'order_notify');

测试连通性时先绕过代码，用管理界面验证

很多“连接失败”根本不是 PHP 问题，而是服务层没配好：

访问 http://localhost:15672，用 admin/123456 登录，看 Queues 标签页能否列出 order_notify
点进队列 → Publish message → 手动发一条 {"order_id":"test"}，看是否成功入队
再点 Get Messages，勾选 Acknowledge mode: Acknowledge all messages in the queue，看能否正常取出并确认