【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 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); // 設定 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 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 中消費掉:
&nb