RabbitMQ(二):mandatory标志的作用
本文轉自:http://m.blog.csdn.net/article/details?id=54311277
在生產者通過channel的basicPublish方法發布消息時,通常有幾個參數需要設置,為此我們有必要了解清楚這些參數代表的具體含義及其作用,查看Channel接口,會發現存在3個重載的basicPublish方法
void basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body) throws IOException;void basicPublish(String exchange, String routingKey, boolean mandatory, BasicProperties props, byte[] body)throws IOException;void basicPublish(String exchange, String routingKey, boolean mandatory, boolean immediate, BasicProperties props, byte[] body)throws IOException;他們共有的參數分別是:
??????? exchange:交換機名稱
??????? routingKey:路由鍵
??????? props:消息屬性字段,比如消息頭部信息等等
??????? body:消息主體部分
??????? 除此之外,還有mandatory和immediate這兩個參數,鑒于RabbitMQ3.0不再支持immediate標志,因此我們重點討論mandatory標志
????????mandatory的作用:
??????? 當mandatory標志位設置為true時,如果exchange根據自身類型和消息routingKey無法找到一個合適的queue存儲消息,那么broker會調用basic.return方法將消息返還給生產者;當mandatory設置為false時,出現上述情況broker會直接將消息丟棄;通俗的講,mandatory標志告訴broker代理服務器至少將消息route到一個隊列中,否則就將消息return給發送者;
?????? 下面我們通過幾個實例測試下mandatory標志的作用:
????????測試1:設置mandatory標志,且exchange未綁定隊列
?????? 第45行我們將basicPublish的第三個參數mandatory設置成了true,表示開啟了mandatory標志,但我們沒有為當前exchange綁定任何隊列;
?????? 通過wireshark抓包看到下面輸出:??
?????? 可以看到最后執行了basic.return方法,將發布者發出的消息返還給了發布者,查看協議的Arguments參數部分可以看到,Reply-Text字段值為:NO_ROUTE,表示消息并沒有路由到合適的隊列中;
?????? 那么我們該怎么獲取到沒有被正確路由到合適隊列的消息呢?這時候可以通過為channel信道設置ReturnListener監聽器來實現,具體實現代碼見下:
?
public class ProducerTest {public static void main(String[] args) {String exchangeName = "confirmExchange";String queueName = "confirmQueue";String routingKey = "confirmRoutingKey";String bindingKey = "confirmBindingKey";int count = 3;ConnectionFactory factory = new ConnectionFactory();factory.setHost("172.16.151.74");factory.setUsername("test");factory.setPassword("test");factory.setPort(5672);//創建生產者Sender producer = new Sender(factory, count, exchangeName, routingKey);producer.run();} }class Sender {private ConnectionFactory factory;private int count;private String exchangeName;private String routingKey;public Sender(ConnectionFactory factory,int count,String exchangeName,String routingKey) {this.factory = factory;this.count = count;this.exchangeName = exchangeName;this.routingKey = routingKey;}public void run() {try {Connection connection = factory.newConnection();Channel channel = connection.createChannel();//創建exchangechannel.exchangeDeclare(exchangeName, "direct", true, false, null);//發送持久化消息for(int i = 0;i < count;i++){//第一個參數是exchangeName(默認情況下代理服務器端是存在一個""名字的exchange的,//因此如果不創建exchange的話我們可以直接將該參數設置成"",如果創建了exchange的話//我們需要將該參數設置成創建的exchange的名字),第二個參數是路由鍵channel.basicPublish(exchangeName, routingKey, true, MessageProperties.PERSISTENT_BASIC, ("第"+(i+1)+"條消息").getBytes());}channel.addReturnListener(new ReturnListener() {@Overridepublic void handleReturn(int arg0, String arg1, String arg2, String arg3, BasicProperties arg4, byte[] arg5)throws IOException {//此處便是執行Basic.Return之后回調的地方String message = new String(arg5);System.out.println("Basic.Return返回的結果: "+message);}});} catch (Exception e) {e.printStackTrace();}} }?????? 在設置了ReturnListener監聽器之后,broker(代理服務器)發出basic.return方法之后,就會回調第52行的handleReturn方法,在這個方法里面我們就可以進行消息的重新發布操作啦;
?????? 測試2:設置mandatory標志,且為exchange綁定隊列(路由鍵和綁定鍵一致)
?
public class ProducerTest {public static void main(String[] args) {String exchangeName = "confirmExchange";String queueName = "confirmQueue";String routingKey = "confirmRoutingKey";String bindingKey = "confirmRoutingKey";//String bindingKey = "confirmBindingKey";int count = 3;ConnectionFactory factory = new ConnectionFactory();factory.setHost("172.16.151.74");factory.setUsername("test");factory.setPassword("test");factory.setPort(5672);//創建生產者Sender producer = new Sender(factory, count, exchangeName, queueName,routingKey,bindingKey);producer.run();} }class Sender {private ConnectionFactory factory;private int count;private String exchangeName;private String queueName;private String routingKey;private String bindingKey;public Sender(ConnectionFactory factory,int count,String exchangeName,String queueName,String routingKey,String bindingKey) {this.factory = factory;this.count = count;this.exchangeName = exchangeName;this.queueName = queueName;this.routingKey = routingKey;this.bindingKey = bindingKey;}public void run() {try {Connection connection = factory.newConnection();Channel channel = connection.createChannel();//創建exchangechannel.exchangeDeclare(exchangeName, "direct", true, false, null);//創建隊列channel.queueDeclare(queueName, true, false, false, null);//綁定exchange和queue channel.queueBind(queueName, exchangeName, bindingKey);//發送持久化消息for(int i = 0;i < count;i++){//第一個參數是exchangeName(默認情況下代理服務器端是存在一個""名字的exchange的,//因此如果不創建exchange的話我們可以直接將該參數設置成"",如果創建了exchange的話//我們需要將該參數設置成創建的exchange的名字),第二個參數是路由鍵channel.basicPublish(exchangeName, routingKey, true, MessageProperties.PERSISTENT_BASIC, ("第"+(i+1)+"條消息").getBytes());}channel.addReturnListener(new ReturnListener() {@Overridepublic void handleReturn(int arg0, String arg1, String arg2, String arg3, BasicProperties arg4, byte[] arg5)throws IOException {//此處便是執行Basic.Return之后回調的地方String message = new String(arg5);System.out.println("Basic.Return返回的結果: "+message);}});} catch (Exception e) {e.printStackTrace();}} }??????? 通過抓包發現并不會有basic.return方法被調用,查看RabbitMQ管理界面發現消息已經到達了隊列;
??????? 測試3:設置mandatory標志,且exchange綁定隊列(路由鍵和綁定鍵不一致)
??????? 代碼就是把測試2中第6行注釋,第7行注釋打開,注意到此時的routingKey和bindingKey是不一致的,此時我們運行程序,同時抓包得到下面截圖:
?????? 注意一點,我們發送了三條消息,那么相應的應該執行三次basic.return,其中第一次和第二次basic.return顯示在一行上了,第三次是單獨一行,不要誤認為只執行了兩次,從協議的具體返回內容里我們同樣看到了Reply-Text字段值是NO_ROUTE,這種現象在測試1中已經見過了;
?????? 到此,我們明白了mandatory標志的作用:在消息沒有被路由到合適隊列情況下會將消息返還給消息發布者,同時我們測試了哪些情況下消息不會到達合適的隊列,測試1演示的是創建了exchange但是沒有為他綁定隊列導致的消息未到達合適隊列,測試3演示的是創建了exchange同時創建了queue,但是在將兩者綁定的時候,使用的bindingKey和消息發布者使用的rountingKey不一致導致的消息未到達合適隊列;
總結
以上是生活随笔為你收集整理的RabbitMQ(二):mandatory标志的作用的全部內容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: Java中的数字
- 下一篇: 客户提的功能(需求),不能随口一说就当做