ThinkPHP6 不内置 RabbitMQ 支持，需用 php-amqplib 手动集成；消费者必须为常驻 CLI 进程，配置持久化、manual ack、basic_qos，否则消息丢失、重复或连接爆炸。
直接说结论：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免费学习笔记（深入）”；
<?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 守护它：
[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，看能否正常取出并确认
这三步通了，再跑 PHP 脚本；不通就别碰代码，先查 rabbitmqctl list_users、firewall-cmd --list-ports、systemctl status rabbitmq-server。
最常被忽略的是 vhost 权限和远程访问限制：默认 guest 用户只允许 localhost 连接，生产环境必须用 rabbitmqctl add_user 创建专用账号，并用 set_permissions 绑定 vhost。
