ThinkPHP 5.1.39 LTS 引入 RabbitMQ/PHP-amqplib 教程-ThinkPHP-PHP中文网

直接用php-amqplib可行但不推荐裸写，因thinkphp5.1.39缺乏重试、监控、优雅重启等能力；think-queue虽支持rabbitmq驱动，但不支持topic/fanout交换机、ttl、dlx、手动ack等高级特性。

thinkphp 5.1.39 lts 引入 rabbitmq/php-amqplib 教程

直接用 php-amqplib 是可行的，但不推荐在 ThinkPHP 5.1.39 LTS 中“裸写” RabbitMQ 逻辑 —— 它缺乏任务重试、失败记录、队列监控、优雅重启等关键能力，容易在生产环境丢消息或卡死。

为什么不用 think-queue + RabbitMQ 驱动

ThinkPHP 官方 think-queue 扩展从 v2.0.4 起支持 RabbitMQ 驱动（基于 php-amqplib 封装），但仅限于「简单 direct exchange + 默认 queue 绑定」场景。它不支持：

topic 或 fanout 交换机类型
自定义 routing_key 动态路由
消息 TTL、死信队列（DLX）、优先级队列等高级特性
消费者端手动 basic_ack / basic_nack 控制粒度

如果你的业务需要这些，就得绕过 think-queue，直接用 php-amqplib，但必须自己补全可靠性机制。

AMQPStreamConnection 初始化必须带超时和重连逻辑

默认 AMQPStreamConnection 在网络抖动或 RabbitMQ 重启时会直接抛出 PhpAmqpLib\Exception\AMQPRuntimeException 并中断进程，而 ThinkPHP 的命令行模式（如 php think queue:work）不会自动恢复连接。

实操建议：

把连接封装进 try/catch，并在失败后 sleep(3) 后重试，最多 5 次
显式设置 $connection = new AMQPStreamConnection(..., ['timeout' => 3, 'keepalive' => true])
不要复用全局 $connection 实例；每次消费/发布前新建连接，用完立即 close()
避免在 __destruct 或 register_shutdown_function 里关连接 —— CLI 进程可能已提前终止，导致 channel->close() 失败并静默丢消息

basic_publish 必须设 delivery_mode=2 且 exchange 声明 durable=true

否则消息写入内存队列，RabbitMQ 重启后全部丢失。常见错误是只设了 queue_declare(..., true, ...)（队列持久化），却忘了 exchange_declare(..., ..., true, ...)（交换机持久化）和 AMQPMessage 的 delivery_mode。

btpanel phpsite 宝塔面板PHP网站

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

下载

正确写法示例：

$message = new AMQPMessage(
    json_encode($data),
    [
        'content_type' => 'application/json',
        'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT // ← 必须是 2
    ]
);
$channel->basic_publish($message, self::exchange, $routingKey);

同时确保声明时三者都开启持久化：

$channel->exchange_declare(self::exchange, 'direct', false, true, false)（第 4 参数 true）
$channel->queue_declare(self::queue, false, true, false, false)（第 3 参数 true）
$channel->queue_bind(self::queue, self::exchange, $routingKey)

消费者必须手动 basic_ack，且不能在异常后跳过

ThinkPHP 默认的命令行生命周期短，一旦 PHP 进程 crash 或被 kill，未 ack 的消息会重回队列 —— 看似安全，但若消费者逻辑里有数据库写入、文件操作等副作用，重复消费就会出问题。

关键点：

务必在业务逻辑执行成功后调用 $message->ack()，而不是靠 auto-ack
捕获所有异常（包括 Fatal error），并在 finally 块中判断是否已 ack；没 ack 就 $message->nack(requeue: true) 或 requeue: false（进死信）
避免在 process_message 回调里做耗时操作（如 curl、大文件读写），应拆成子进程或投递到二级队列
用 rabbitmqctl list_queues name messages_unacknowledged 定期检查堆积量，防止 channel 被占满阻塞新消息

最易被忽略的是：ThinkPHP 的日志写入（Log::write()）本身可能触发 autoload 或 DB 连接，在消费者进程里若未预加载类或关闭事务，会导致 ack 前就崩溃 —— 此时消息既没处理完，也没 nack，陷入“灰色状态”。