RabbitMQ PHP 客户端操作

php-amqplib 是 PHP 中最常用的 RabbitMQ 客户端库，纯 PHP 实现，无需额外扩展，支持 AMQP 0-9-1 协议。本节将深入讲解如何在 PHP 中连接 RabbitMQ、生产消息、消费消息、ACK/NACK 确认机制、死信队列等核心操作，并提供生产级封装示例。

前置知识

阅读本节前，建议先了解：RabbitMQ 基础 和 Composer 包管理

安装与配置

# 安装 php-amqplib（支持 PHP 8.0+）
composer require php-amqplib/php-amqplib

# 如果需要异步消费，推荐使用 Swoole 或 ReactPHP 配合
# composer require swoole/ide-helper

基础连接封装

<?php

declare(strict_types=1);

namespace App\MessageQueue;

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Connection\AbstractConnection;

class RabbitMQConnectionFactory
{
    private static ?AbstractConnection $connection = null;

    /**
     * 创建连接（单例，复用 TCP 连接）
     */
    public static function getConnection(): AbstractConnection
    {
        if (self::$connection === null || !self::$connection->isConnected()) {
            self::$connection = new AMQPStreamConnection(
                host: getenv('RABBITMQ_HOST') ?: '127.0.0.1',
                port: (int) (getenv('RABBITMQ_PORT') ?: 5672),
                user: getenv('RABBITMQ_USER') ?: 'guest',
                password: getenv('RABBITMQ_PASSWORD') ?: 'guest',
                vhost: getenv('RABBITMQ_VHOST') ?: '/',
                insist: false,
                login_method: 'AMQPLAIN',
                login_response: null,
                locale: 'en_US',
                connection_timeout: 3.0,
                read_write_timeout: 10.0,
                context: null,
                keepalive: false,
                heartbeat: 60,
                channel_rpc_timeout: 5.0,
                ssl_protocol: null,
            );
        }

        return self::$connection;
    }

    /**
     * 创建 Channel（每次操作建议使用新的 Channel）
     */
    public static function createChannel(): AMQPChannel
    {
        return self::getConnection()->channel();
    }

    /**
     * 关闭连接
     */
    public static function close(): void
    {
        if (self::$connection !== null) {
            self::$connection->close();
            self::$connection = null;
        }
    }
}

生产者（Producer）

基础消息生产

<?php

declare(strict_types=1);

namespace App\MessageQueue;

use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Exchange\AMQPExchangeType;

class RabbitMQProducer
{
    public function __construct(
        private readonly AMQPChannel $channel
    ) {}

    /**
     * 声明 Exchange
     */
    public function declareExchange(
        string $exchangeName,
        string $type = AMQPExchangeType::DIRECT,
        bool $durable = true,
        bool $autoDelete = false
    ): void {
        $this->channel->exchange_declare(
            exchange: $exchangeName,
            type: $type,
            passive: false,
            durable: $durable,
            auto_delete: $autoDelete,
            internal: false,
            nowait: false,
            arguments: [],
        );
    }

    /**
     * 发送消息
     */
    public function publish(
        string $exchangeName,
        string $routingKey,
        array|string $message,
        array $properties = []
    ): void {
        $body = is_array($message) ? json_encode($message) : $message;

        $msg = new AMQPMessage(
            body: $body,
            properties: array_merge([
                'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT, // 持久化消息
                'content_type'  => 'application/json',
                'timestamp'     => time(),
                'app_id'        => 'php-producer',
            ], $properties)
        );

        $this->channel->basic_publish(
            msg: $msg,
            exchange: $exchangeName,
            routing_key: $routingKey,
            mandatory: false,
            immediate: false,
            ticket: null,
        );
    }

    /**
     * 发送延迟消息（使用死信交换机模拟）
     * RabbitMQ 需要安装 delayed_message_exchange 插件
     */
    public function publishDelayed(
        string $exchangeName,
        string $routingKey,
        array|string $message,
        int $delayMs,
        array $properties = []
    ): void {
        $body = is_array($message) ? json_encode($message) : $message;

        $msg = new AMQPMessage(
            body: $body,
            properties: array_merge([
                'delivery_mode'  => AMQPMessage::DELIVERY_MODE_PERSISTENT,
                'content_type'   => 'application/json',
                'expiration'     => (string) $delayMs,
                'timestamp'      => time(),
            ], $properties)
        );

        // 使用延迟交换机（需要插件支持）
        // 插件安装：rabbitmq-plugins enable rabbitmq_delayed_message_exchange
        $this->channel->basic_publish(
            msg: $msg,
            exchange: $exchangeName . '.delayed',
            routing_key: $routingKey,
        );
    }
}

确认发布（Publisher Confirm）

<?php

declare(strict_types=1);

namespace App\MessageQueue;

use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Message\AMQPMessage;

class ConfirmPublisher
{
    public function __construct(
        private readonly AMQPChannel $channel
    ) {}

    /**
     * 开启 Publisher Confirm 模式
     */
    public function enableConfirm(int $timeout = 5): void
    {
        $this->channel->confirm_select();

        // 注册 ACK 回调
        $this->channel->set_ack_handler(
            function (int $deliveryTag, bool $multiple): void {
                // 消息被 Broker 确认接收
                // log: "Message {$deliveryTag} confirmed"
            }
        );

        // 注册 NACK 回调
        $this->channel->set_nack_handler(
            function (int $deliveryTag, bool $multiple, bool $requeue): void {
                // 消息被 Broker 拒绝
                // log: "Message {$deliveryTag} nacked, requeue: " . ($requeue ? 'yes' : 'no')
            }
        );

        // 注册超时回调
        $this->channel->set_return_listener(
            function (int $replyCode, string $replyText, string $exchange, string $routingKey): void {
                // 消息无法路由
                // log: "Message returned: {$replyText}, exchange={$exchange}, routing_key={$routingKey}"
            }
        );
    }

    /**
     * 等待所有消息确认
     */
    public function waitForConfirms(float $timeout = 5.0): void
    {
        $this->channel->wait_for_pending_acks($timeout);
    }
}

消费者（Consumer）

基础消息消费

<?php

declare(strict_types=1);

namespace App\MessageQueue;

use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Exception\AMQPRuntimeException;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Exchange\AMQPExchangeType;

class RabbitMQConsumer
{
    public function __construct(
        private readonly AMQPChannel $channel
    ) {}

    /**
     * 声明队列并绑定到 Exchange
     */
    public function declareAndBind(
        string $queueName,
        string $exchangeName,
        string $routingKey = '',
        bool $durable = true,
        array $arguments = []
    ): void {
        $this->channel->queue_declare(
            queue: $queueName,
            passive: false,
            durable: $durable,
            exclusive: false,
            auto_delete: false,
            nowait: false,
            arguments: $arguments,
        );

        $this->channel->queue_bind(
            queue: $queueName,
            exchange: $exchangeName,
            routing_key: $routingKey,
            nowait: false,
            arguments: [],
        );
    }

    /**
     * 消费消息（自动 ACK）
     */
    public function consumeAutoAck(string $queueName, callable $callback): void
    {
        $this->channel->basic_consume(
            queue: $queueName,
            consumer_tag: '',
            no_local: false,
            no_ack: true,           // 自动确认
            exclusive: false,
            nowait: false,
            callback: function (AMQPMessage $message) use ($callback): void {
                $data = json_decode($message->body, true) ?? $message->body;
                $callback($data, $message->getDeliveryTag());
            },
            ticket: null,
            arguments: [],
        );

        // 等待消息
        while ($this->channel->is_consuming()) {
            try {
                $this->channel->wait();
            } catch (AMQPRuntimeException $e) {
                // 连接断开等异常处理
                break;
            }
        }
    }

    /**
     * 消费消息（手动 ACK）
     */
    public function consumeManualAck(string $queueName, callable $callback): void
    {
        $this->channel->basic_qos(
            prefetch_size: 0,
            prefetch_count: 10,   // 每次最多获取 10 条未确认消息
            global: false,
        );

        $this->channel->basic_consume(
            queue: $queueName,
            consumer_tag: '',
            no_local: false,
            no_ack: false,          // 手动确认
            exclusive: false,
            nowait: false,
            callback: function (AMQPMessage $message) use ($callback): void {
                $data = json_decode($message->body, true) ?? $message->body;
                $deliveryTag = $message->getDeliveryTag();

                try {
                    $result = $callback($data, $deliveryTag);

                    if ($result === true) {
                        // 处理成功，ACK
                        $message->ack();
                    } elseif ($result === false) {
                        // 处理失败，NACK 并重新入队
                        $message->nack(requeue: true);
                    } else {
                        // 需要延迟重试，NACK 不重新入队
                        $message->nack(requeue: false);
                    }
                } catch (\Throwable $e) {
                    // 异常处理：NACK 不重新入队（避免无限重试）
                    $message->nack(requeue: false);
                    // 可记录到死信队列或错误日志
                }
            },
            ticket: null,
            arguments: [],
        );

        while ($this->channel->is_consuming()) {
            try {
                $this->channel->wait(timeout: 30);
            } catch (AMQPRuntimeException $e) {
                break;
            }
        }
    }

    /**
     * 优雅关闭消费者
     */
    public function gracefulShutdown(string $consumerTag = ''): void
    {
        if ($consumerTag) {
            $this->channel->basic_cancel($consumerTag, false, false);
        }
        $this->channel->close();
    }
}

死信队列（DLX）

死信队列原理

消息成为死信（Dead Letter）的条件：

1. 消息被消费者 NACK 且 requeue: false
2. 消息在队列中的 TTL 过期
3. 队列达到最大长度，新消息被丢弃

死信消息会被路由到配置的死信 Exchange（DLX），再由 DLX 路由到死信队列。

正常流程:
  Producer → Exchange → Queue → Consumer (ACK) → 完成

死信流程:
  Producer → Exchange → Queue → Consumer (NACK, requeue=false)
                         → 消息过期 (TTL)
                         → 队列满
                              ↓
                      Dead Letter Exchange → Dead Letter Queue → DLX Consumer

PHP 配置死信队列

<?php

declare(strict_types=1);

namespace App\MessageQueue;

use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Exchange\AMQPExchangeType;
use PhpAmqpLib\Message\AMQPMessage;

class DeadLetterQueueSetup
{
    public function __construct(
        private readonly AMQPChannel $channel
    ) {}

    /**
     * 配置死信队列
     * 
     * 1. 声明死信 Exchange（DLX）
     * 2. 声明死信 Queue（DLQ）
     * 3. 绑定 DLX 到 DLQ
     * 4. 在主队列上配置 x-dead-letter-exchange
     */
    public function setup(
        string $mainExchangeName,
        string $mainQueueName,
        string $mainRoutingKey,
        string $dlxExchangeName,
        string $dlxQueueName,
        string $dlxRoutingKey = ''
    ): void {
        // 1. 声明死信 Exchange
        $this->channel->exchange_declare(
            exchange: $dlxExchangeName,
            type: AMQPExchangeType::DIRECT,
            durable: true,
        );

        // 2. 声明死信 Queue
        $this->channel->queue_declare(
            queue: $dlxQueueName,
            durable: true,
            arguments: [],
        );

        // 3. 绑定 DLX 到 DLQ
        $this->channel->queue_bind(
            queue: $dlxQueueName,
            exchange: $dlxExchangeName,
            routing_key: $dlxRoutingKey,
        );

        // 4. 声明主队列（配置 DLX 参数）
        $this->channel->queue_declare(
            queue: $mainQueueName,
            durable: true,
            arguments: [
                'x-dead-letter-exchange'    => $dlxExchangeName,
                'x-dead-letter-routing-key' => $dlxRoutingKey,
                'x-message-ttl'              => 86400000,   // 消息最长保存 24 小时
                'x-max-length'               => 1000000,   // 队列最大 100 万条消息
            ],
        );

        // 5. 声明主 Exchange
        $this->channel->exchange_declare(
            exchange: $mainExchangeName,
            type: AMQPExchangeType::DIRECT,
            durable: true,
        );

        // 6. 绑定主 Exchange 到主 Queue
        $this->channel->queue_bind(
            queue: $mainQueueName,
            exchange: $mainExchangeName,
            routing_key: $mainRoutingKey,
        );
    }

    /**
     * 消费死信队列（重试或记录）
     */
    public function consumeDeadLetters(string $dlxQueueName, callable $handler): void
    {
        $this->channel->basic_consume(
            queue: $dlxQueueName,
            no_ack: false,
            callback: function (AMQPMessage $message) use ($handler): void {
                $data = json_decode($message->body, true) ?? $message->body;

                try {
                    $result = $handler($data, $message->getDeliveryTag());

                    if ($result === true) {
                        // 处理成功（如重试成功），ACK
                        $message->ack();
                    } else {
                        // 重试失败，记录到告警系统
                        // AlertService::send("Dead letter message failed: {$message->body}");
                        $message->ack(); // 最终 ACK，避免无限死信循环
                    }
                } catch (\Throwable $e) {
                    $message->ack(); // 记录后 ACK
                }
            },
        );

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

实战示例：订单消息队列

<?php

declare(strict_types=1);

namespace App\MessageQueue;

use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Message\AMQPMessage;

/**
 * 订单消息生产者
 */
class OrderProducer
{
    private const EXCHANGE = 'order.exchange';
    private const QUEUE_CREATED    = 'order.created.queue';
    private const QUEUE_CANCELLED = 'order.cancelled.queue';
    private const QUEUE_PAID      = 'order.paid.queue';

    public function __construct(
        private readonly AMQPChannel $channel
    ) {
        $this->setupTopology();
    }

    private function setupTopology(): void
    {
        // 声明 Exchange
        $this->channel->exchange_declare(self::EXCHANGE, 'direct', durable: true);

        // 声明队列
        $this->channel->queue_declare(self::QUEUE_CREATED, durable: true);
        $this->channel->queue_declare(self::QUEUE_CANCELLED, durable: true);
        $this->channel->queue_declare(self::QUEUE_PAID, durable: true);

        // 绑定路由
        $this->channel->queue_bind(self::QUEUE_CREATED, self::EXCHANGE, 'order.created');
        $this->channel->queue_bind(self::QUEUE_CANCELLED, self::EXCHANGE, 'order.cancelled');
        $this->channel->queue_bind(self::QUEUE_PAID, self::EXCHANGE, 'order.paid');
    }

    public function publishOrderCreated(array $orderData): void
    {
        $this->publish('order.created', $orderData);
    }

    public function publishOrderCancelled(array $orderData): void
    {
        $this->publish('order.cancelled', $orderData);
    }

    private function publish(string $routingKey, array $data): void
    {
        $msg = new AMQPMessage(
            body: json_encode($data),
            properties: [
                'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
                'content_type'  => 'application/json',
                'message_id'    => uniqid('order_', true),
                'timestamp'     => time(),
            ]
        );

        $this->channel->basic_publish($msg, self::EXCHANGE, $routingKey);
    }
}

注意事项

1. 连接管理：使用单例 Connection，每次操作创建新 Channel
2. 消息持久化：delivery_mode: 2 确保消息不因 RabbitMQ 重启丢失
3. 手动 ACK：生产环境禁用 auto_ack，手动确认消息处理完成
4. prefetch_count：根据消费者处理能力设置，避免消息堆积
5. 错误处理：捕获连接异常，实现重连机制

最佳实践

1. Exchange/Queue 预声明：应用启动时统一声明，避免运行时创建
2. 消息幂等：消费者实现幂等处理，避免重复消费
3. 消息格式：统一使用 JSON，在消息头中标注消息类型和版本
4. 日志追踪：使用 message_id 便于追踪消息流转
5. 监控：监控队列长度、消费速率、未确认消息数

下一节

继续学习：RabbitMQ 最佳实践

参考链接

• php-amqplib GitHub
• RabbitMQ Publisher Confirms
• RabbitMQ Dead Letter Exchange
• RabbitMQ Consumer Prefetch