# ?RabbitMQ入門教程 For Java【4】 -?[Routing](http://www.rabbitmq.com/tutorials/tutorial-four-python.html) ### **我的開發環境：** 操作系統：**Windows7 64bit** 開發環境：**JDK 1.7 -?1.7.0_55** 開發工具：**Eclipse Kepler SR2** RabbitMQ版本：?**3.6.0** Elang版本：**erl7.2.1** 關于Windows7下安裝RabbitMQ的教程請先在網上找一下，有空我再補安裝教程。 ### 源碼地址 https://github.com/chwshuang/rabbitmq.git ### 消息路由 ? ? ? ? 上一章教程中我們建立了一個簡單的日志記錄系統，能夠將消息廣播到多個消費者。本章，我們將添加一個新功能，類似訂閱消息的子集。例如：我們只接收日志文件中ERROR類型的日志。 ### 綁定關系 ? ? ? ??在之前的例子中也使用了類似的方式： ~~~ channel.queueBind(queueName, EXCHANGE_NAME, ""); ~~~ ? ? ? ??綁定是交換器和隊列之間的一種關系，用戶微博，微信的例子可以簡單的理解為關注，就是隊列（某屌絲）對交換器（女神）非常感興趣，關注了她，以后女神發的每條微博，屌絲都能看到。 ? ? ? ??綁定可以使用routingkey這個參數，是為了避免所有的消息都使用同一個路由線索帶來的麻煩。為了區分路由規則，我們創建創建一個唯一的路由線索。 ~~~ channel.queueBind(queueName, EXCHANGE_NAME, "black"); ~~~ ? ? ? ??綁定關系中使用的路由關鍵字【routingkey】是否有效取決于交換器的類型。如果交換器是分發【fanout】類型，就會忽略路由關鍵字【routingkey】的作用。 ### 直連類型交換器 ? ? ? ??上一章的例子是通過分發【fanout】類型的交換器【logs】廣播日志信息，現在我們將日志分debug、info、warn、error這幾種基本的級別，實際在生產環境中，避免磁盤空間浪費，應用只會將error級別的日志打印出來。而分發【fanout】類型的交換器會將所有基本的日志都發送出來，如果我們想只接收某一級別的日志信息，就需要使用直連【direct】類型的交換器了, 下面的圖中，隊列1通過ERROR這個routingkey綁定到E交換器，隊列2通過WARN和INFO綁定到E交換器，E交換器的類型是直連【direct】的，如果生產者【P】發出ERROR的日志，只會有隊列1會收到，如果生產者【P】發出INFO和WARN的日志，只有隊列2會收到，如果生產者【P】發出DEBUG級別的日志，隊列1和隊列2都會忽略它。  ### 多重綁定 ? ? ? ? 我們允許多個隊列以相同的路由關鍵字綁定到同一個交換器中，可以看到，交換器雖然是直連類型，但是綁定后的效果卻跟分發類型的交換器類似，相同的是隊列1和隊列2都會收到同一條來自交換器的消息。 ? ? ? ? 他們的區別：分發模式下，隊列1、隊列2會收到所有級別（除ERROR級別以外）的消息，而直連模式下，他們僅僅只會收到ERROR關鍵字類型的消息。  ### 發送日志消息 ? ? ? ? 我們還是用日志系統進行講解，現在我們用日志的級別來作為路由關鍵字【routingkey】，這樣，消費者端就可以按照他關心的日志級別進行接收，我們先看看如何發送日志： ? ? ? ??先聲明交換器 ~~~ <span style="font-size: 18px;">? ? ? ??</span>channel.exchangeDeclare(EXCHANGE_NAME, "direct"); ~~~ ? ? ? ??然后發送消息到交換器 ~~~ for (String severity : routingKeys) { channel.queueBind(queueName, EXCHANGE_NAME, severity); System.out.println("ReceiveLogsDirect1 exchange:"+EXCHANGE_NAME+", queue:"+queueName+", BindRoutingKey:" + severity); } ~~~ ### 訂閱消息 ? ? ? ??我們先獲取一個隨機的隊列名稱，然后根據多個路由關鍵字【routingkey】將隊列和交換器綁定起來： ~~~ String queueName = channel.queueDeclare().getQueue(); for(String severity : argv){ channel.queueBind(queueName, EXCHANGE_NAME, severity); } ~~~ ### 項目說明 ### 流程圖  ### 包圖  ### 代碼 RoutingSendDirect.java ~~~ import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.Connection; import com.rabbitmq.client.Channel; /** * @author hushuang * */ public class RoutingSendDirect { private static final String EXCHANGE_NAME = "direct_logs"; // 路由關鍵字 private static final String[] routingKeys = new String[]{"info" ,"warning", "error"}; public static void main(String[] argv) throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); // 聲明交換器 channel.exchangeDeclare(EXCHANGE_NAME, "direct"); // 發送消息 for(String severity :routingKeys){ String message = "Send the message level:" + severity; channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes()); System.out.println(" [x] Sent '" + severity + "':'" + message + "'"); } channel.close(); connection.close(); } } ~~~ ReceiveLogsDirect1.java和ReceiveLogsDirect2.java ~~~ import com.rabbitmq.client.*; import java.io.IOException; public class ReceiveLogsDirect1 { // 交換器名稱 private static final String EXCHANGE_NAME = "direct_logs"; // 路由關鍵字 private static final String[] routingKeys = new String[]{"info" ,"warning", "error"}; public static void main(String[] argv) throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); // 聲明交換器 channel.exchangeDeclare(EXCHANGE_NAME, "direct"); // 獲取匿名隊列名稱 String queueName = channel.queueDeclare().getQueue(); // 根據路由關鍵字進行多重綁定 for (String severity : routingKeys) { channel.queueBind(queueName, EXCHANGE_NAME, severity); System.out.println("ReceiveLogsDirect1 exchange:"+EXCHANGE_NAME+", queue:"+queueName+", BindRoutingKey:" + severity); } System.out.println("ReceiveLogsDirect1 [*] Waiting for messages. To exit press CTRL+C"); Consumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String message = new String(body, "UTF-8"); System.out.println(" [x] Received '" + envelope.getRoutingKey() + "':'" + message + "'"); } }; channel.basicConsume(queueName, true, consumer); } } ~~~ ~~~ import com.rabbitmq.client.*; import java.io.IOException; public class ReceiveLogsDirect2 { // 交換器名稱 private static final String EXCHANGE_NAME = "direct_logs"; // 路由關鍵字 private static final String[] routingKeys = new String[]{"error"}; public static void main(String[] argv) throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); // 聲明交換器 channel.exchangeDeclare(EXCHANGE_NAME, "direct"); // 獲取匿名隊列名稱 String queueName = channel.queueDeclare().getQueue(); // 根據路由關鍵字進行多重綁定 for (String severity : routingKeys) { channel.queueBind(queueName, EXCHANGE_NAME, severity); System.out.println("ReceiveLogsDirect1 exchange:"+EXCHANGE_NAME+", queue:"+queueName+", BindRoutingKey:" + severity); } System.out.println("ReceiveLogsDirect1 [*] Waiting for messages. To exit press CTRL+C"); Consumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String message = new String(body, "UTF-8"); System.out.println(" [x] Received '" + envelope.getRoutingKey() + "':'" + message + "'"); } }; channel.basicConsume(queueName, true, consumer); } } ~~~ 先運行ReceiveLogsDirect1.java和ReceiveLogsDirect2.java 查看日志，我們發現RabbitMQ中已經創建了direct_logs的交換器，以及amq.gen-dVUpkqxmladY3Jg1upDsDQ 和amq.gen-skrmBAlYKSDzELKtVg_zFw這兩個臨時隊列， ~~~ ReceiveLogsDirect1 exchange:direct_logs, queue:amq.gen-skrmBAlYKSDzELKtVg_zFw, BindRoutingKey:info ReceiveLogsDirect1 exchange:direct_logs, queue:amq.gen-skrmBAlYKSDzELKtVg_zFw, BindRoutingKey:warning ReceiveLogsDirect1 exchange:direct_logs, queue:amq.gen-skrmBAlYKSDzELKtVg_zFw, BindRoutingKey:error ReceiveLogsDirect1 [*] Waiting for messages. To exit press CTRL+C ~~~ ~~~ ReceiveLogsDirect2 exchange:direct_logs, queue:amq.gen-dVUpkqxmladY3Jg1upDsDQ, BindRoutingKey:error ReceiveLogsDirect2 [*] Waiting for messages. To exit press CTRL+C ~~~ 運行RoutingSendDirect.java發送消息： ### 運行結果 查看日志： RoutingSendDirect.java ~~~ [x] Sent 'info':'Send the message level:info' [x] Sent 'warning':'Send the message level:warning' [x] Sent 'error':'Send the message level:error' ~~~ ReceiveLogsDirect1.java ~~~ [x] Received 'info':'Send the message level:info' [x] Received 'warning':'Send the message level:warning' [x] Received 'error':'Send the message level:error' ~~~ ReceiveLogsDirect2.java ~~~ [x] Received 'error':'Send the message level:error' ~~~ 我們看到，隊列1收到了所有的消息，隊列2只收到了error級別的消息。這與我們的預期一樣。 下一階段我們可以進入第五章-主題的學習了。 本教程所有文章： [RabbitMQ入門教程 For Java【1】 - Hello World](http://blog.csdn.net/chwshuang/article/details/50521708)? -?你好世界！? [RabbitMQ入門教程 For Java【2】 - Work Queues](http://blog.csdn.net/chwshuang/article/details/50506284)??- 工作隊列 [RabbitMQ入門教程 For Java【3】 - Publish/Subscribe](http://blog.csdn.net/chwshuang/article/details/50512057)?- 發布/訂閱 [RabbitMQ入門教程 For Java【4】 - Routing](http://blog.csdn.net/chwshuang/article/details/50505060)?- ?消息路由 [RabbitMQ入門教程 For Java【5】 - Topic](http://blog.csdn.net/chwshuang/article/details/50516904)? - ?模糊匹配 [RabbitMQ入門教程 For Java【6】 - Remote procedure call (RPC)](http://blog.csdn.net/chwshuang/article/details/50518570)?- 遠程調用