RabbitMQ - 死信队列

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

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

$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();

// 创建一个处理死信的交换机和队列
$dead_exchange = 'dead_direct_exchange';
$dead_queue_name = 'dead_queue';
$dead_routing_key = 'dead';
$channel->exchange_declare($dead_exchange, 'direct', false, false, false);
$channel->queue_declare($dead_queue_name, false, true, false, false);
$channel->queue_bind($dead_queue_name, $dead_exchange, $dead_routing_key);

$ttl_exchange = 'ttl_exchange';
$ttl_routing_key = 'ttl_message';
$channel->exchange_declare($ttl_exchange, 'direct', false, false, false);

// 创建一个 TTL 消息的队列
$arguments = new AMQPTable();
$arguments->set('x-message-ttl', 5000);                             // 设置队列中消息的过期时间  5s
$arguments->set('x-dead-letter-exchange', $dead_exchange);          // 设置处理死信消息的交换机
$arguments->set('x-dead-letter-routing-key', $dead_routing_key);    // 指定一个路由关键字用于死信消息。如果没有设置，则使用消息自己的路由键(fanout 模式不需要)
$ttl_queue_message = 'ttl_queue_message';
$channel->queue_declare($ttl_queue_message, false, true, false, false, false, $arguments);
$channel->queue_bind($ttl_queue_message, $ttl_exchange, $ttl_routing_key);

// 创建一个 TTL 的队列 (过期后不会死信消息！)
$arguments = new AMQPTable();
$arguments->set('x-expires', 10000);                                // 设置队列的过期时间  10s
$arguments->set('x-dead-letter-exchange', $dead_exchange);          // 设置处理死信消息的交换机
$arguments->set('x-dead-letter-routing-key', $dead_routing_key);    // 指定一个路由关键字用于死信消息。如果没有设置，则使用消息自己的路由键(fanout 模式不需要)
$ttl_queue = 'ttl_queue';
$channel->queue_declare($ttl_queue, false, true, false, false, false, $arguments);
$channel->queue_bind($ttl_queue, $ttl_exchange, $ttl_routing_key);

// 往队列中发送 10 条消息
for ($i = 1; $i <= 10; $i++) {
    $msg = new AMQPMessage('AAA test test——' . $i, [
        'expiration' => 10000          // 设置每条消息过期时间。（当同时指定了每队列和每条消息的 TTL 时，将选择两者之间的较低值）
    ]);
    $channel->basic_publish($msg, $ttl_exchange, $ttl_routing_key);
}

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