【RabbitMQ 實戰指南】一 死信隊列

1、死信隊列

DLX，全稱為 Dead-Letter-Exchange，可以稱之為死信交換器。當消息在一個隊列中變成死信（dead message）之后，它能被發送到另一個交換器中，這個交換器就是DLX，綁定DLX的隊列就稱之為死信隊列。

DLX 也是一個正常的交換器，和一般的交換器沒有區別，它能在任何的隊列上被指定，實際上就是設置某個隊列的屬性。當這個隊列中存在死信時，RabbitMQ就會自動地將這個消息重新發布到設置的DLX上去，進而被路由到另一個隊列，即死信隊列。然后可以監聽這個死信隊列中的消息進行相應的處理。

2、消息變成死信的情況

• 消息被拒絕（Basic.Reject/Basic.Nack），並且設置 requeue 參數為false， requeue 表示是否重新入隊 
• 消息過期
• 隊列達到最大長度（聲明隊列的時候設置 x-max-length 參數，表示隊列最大長度）

3、死信隊列設置

可以通過為隊列設置 x-dead-letter-exchange 參數設置 DLX，也可以通過設置 x-dead-letter-routing-key 參數為這個DLX指定路由鍵，如果沒有特殊指定，則使用原隊列的路由鍵。

4、死信隊列測試

4.1 測試過程 

整個過程如下圖：

• 第一步：創建兩個交換器 exchange.normal 和 exchange.dlx, 分別綁定兩個隊列 queue.normal 和 queue.dlx
• 第二步：把 queue.normal 隊列里面的消息配置過期時間，然后通過 x-dead-letter-exchange 指定死信交換器為 exchange.dlx
• 第三步：發送消息到 queue.normal 中，消息過期之后流入 exchange.dlx，然后路由到 queue.dlx 隊列中，然后進行消費

4.2 生產者代碼 

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

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

/**
 * 死信隊列測試
 * 1、創建兩個交換器 exchange.normal 和 exchange.dlx, 分別綁定兩個隊列 queue.normal 和 queue.dlx
 * 2、把 queue.normal 隊列里面的消息配置過期時間，然后通過 x-dead-letter-exchange 指定死信交換器為 exchange.dlx
 * 3、發送消息到 queue.normal 中，消息過期之后流入 exchange.dlx，然后路由到 queue.dlx 隊列中，進行消費
 */

// todo 更改配置
$connection = new AMQPStreamConnection('192.168.33.1', 5672, 'zhangcs', 'zhangcs', '/');

$channel = $connection->channel();

$channel->exchange_declare('exchange.dlx', AMQPExchangeType::DIRECT, false, true);
$channel->exchange_declare('exchange.normal', AMQPExchangeType::FANOUT, false, true);
$args = new AMQPTable();
// 消息過期方式：設置 queue.normal 隊列中的消息10s之后過期
$args->set('x-message-ttl', 10000);
// 設置隊列最大長度方式： x-max-length 
//$args->set('x-max-length', 1);
$args->set('x-dead-letter-exchange', 'exchange.dlx');
$args->set('x-dead-letter-routing-key', 'routingkey');
$channel->queue_declare('queue.normal', false, true, false, false, false, $args);
$channel->queue_declare('queue.dlx', false, true, false, false);

$channel->queue_bind('queue.normal', 'exchange.normal');
$channel->queue_bind('queue.dlx', 'exchange.dlx', 'routingkey');
$message = new AMQPMessage('Hello DLX Message');
$channel->basic_publish($message, 'exchange.normal', 'rk');

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

 

運行生產者代碼之后，queue.normal 隊列會有一條消息，如下圖：

10秒之后，消息會過期，然后被進入 exchange.dlx, 進而路由到 queue.dlx 隊列中：

4.3、消費者代碼

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

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

// todo 更改配置
$connection = new AMQPStreamConnection('192.168.33.1', 5672, 'zhangcs', 'zhangcs', '/');
$channel = $connection->channel();

$channel->exchange_declare('exchange.dlx', AMQPExchangeType::DIRECT, false, true);
$channel->queue_declare('queue.dlx', false, true, false, false);

$channel->queue_bind('queue.dlx', 'exchange.dlx', 'routingkey');

function process_message($message)
{
    echo "\n--------\n";
    echo $message->body;
    echo "\n--------\n";
    $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
}

$channel->basic_consume('queue.dlx', 'consumer_tag', false, false, false, false, 'process_message');

function shutdown($channel, $connection)
{
    $channel->close();
    $connection->close();
}
register_shutdown_function('shutdown', $channel, $connection);

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

運行消費者代碼之后，消費會從 queue.dlx 中消費掉：