【RabbitMQ 實戰指南】一 死信隊列
- 2019 年 10 月 25 日
- 筆記
1、死信隊列
DLX，全稱為 Dead-Letter-Exchange，可以稱之為死信交換器。當消息在一個隊列中變成死信（dead message）之後，它能被發送到另一個交換器中，這個交換器就是DLX，綁定DLX的隊列就稱之為死信隊列。
DLX 也是一個正常的交換器，和一般的交換器沒有區別，它能在任何的隊列上被指定，實際上就是設置某個隊列的屬性。當這個隊列中存在死信時，RabbitMQ就會自動地將這個消息重新發佈到設置的DLX上去，進而被路由到另一個隊列，即死信隊列。然後可以監聽這個死信隊列中的消息進行相應的處理。
2、消息變成死信的情況
- 消息被拒絕
- 消息過期
- 隊列達到最大長度
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 PhpAmqpLibWireAMQPTable; use PhpAmqpLibMessageAMQPMessage; use PhpAmqpLibExchangeAMQPExchangeType; use PhpAmqpLibConnectionAMQPStreamConnection; /** * 死信隊列測試 * 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); // 設置 queue.normal 隊列中的消息10s之後過期 $args = new AMQPTable(); $args->set('x-message-ttl', 10000); $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 PhpAmqpLibMessageAMQPMessage; use PhpAmqpLibExchangeAMQPExchangeType; use PhpAmqpLibConnectionAMQPStreamConnection; // 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 中消費掉：
