RabbitMQ中的路由模型（direct）

路由模型

RabbitMQ 提供了五種不同的通信模型，上一篇文章中，簡單的介紹了一下RabbitMQ的發(fā)布訂閱模型模型。這篇文章來學習一下RabbitMQ中的路由模型（direct）。

路由模型（direct）：路由模式相當于是分布訂閱模式的升級版，多了一個 路由key來約束隊列與交換機的綁定。

在路由模型中，生產(chǎn)者將消息發(fā)送到交換機，交換機根據(jù)消息的路由鍵將消息轉發(fā)到對應的隊列中。每個隊列可以綁定多個路由鍵，每個路由鍵可以綁定到多個隊列中。消費者從隊列中接收消息并處理。當一個路由鍵被多個隊列綁定時，交換機會將消息發(fā)送到所有綁定的隊列中。當一個隊列綁定多個路由鍵時，該隊列將能夠接收到所有路由鍵對應的消息。

適用場景

路由模型適用于需要點對點通信的場景，例如：

1. 系統(tǒng)監(jiān)控告警通知；
2. 任務分發(fā)；
3. 用戶私信系統(tǒng)；
4. 訂單確認通知等。

演示

1. 生產(chǎn)者

   // 生產(chǎn)者
   public class Producer {
       private static final String EXCHANGE_NAME = "exchange_direct_1";
       // 定義路由的key，key值是可以隨意定義的
       private static final String EXCHANGE_ROUTING_KEY1 = "direct_km1";
       private static final String EXCHANGE_ROUTING_KEY2 = "direct_km2";
   
       public static void main(String[] args) throws IOException, TimeoutException {
           Connection connection = ConnectionUtils.getConnection();
           Channel channel = connection.createChannel();
           channel.exchangeDeclare(EXCHANGE_NAME, "direct");
           for (int i = 0; i < 100; i++) {
               if (i % 2 == 0) {
                   channel.basicPublish(EXCHANGE_NAME, EXCHANGE_ROUTING_KEY1, null, ("路由模型發(fā)送的第 " + i + " 條信息").getBytes());
               } else {
                   channel.basicPublish(EXCHANGE_NAME, EXCHANGE_ROUTING_KEY2, null, ("路由模型發(fā)送的第 " + i + " 條信息").getBytes());
               }
           }
           channel.close();
           connection.close();
       }
   }
   

2. 消費者

   // 消費者1
   public class Consumer {
       private static final String QUEUE_NAME = "queue_direct_1";
       private static final String EXCHANGE_NAME = "exchange_direct_1";
       private static final String EXCHANGE_ROUTING_KEY1 = "direct_km1";
   
       public static void main(String[] args) throws IOException, TimeoutException {
           Connection connection = ConnectionUtils.getConnection();
           Channel channel = connection.createChannel();
           channel.queueDeclare(QUEUE_NAME, false, false, false, null);
           channel.exchangeDeclare(EXCHANGE_NAME, "direct");
           channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, EXCHANGE_ROUTING_KEY1);
           DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
               @Override
               public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                   System.out.println("消費者1接收到的消息是：" + new String(body));
               }
           };
           channel.basicConsume(QUEUE_NAME, true, defaultConsumer);
       }
   }
   

   // 消費者2
   public class Consumer2 {
       private static final String QUEUE_NAME = "queue_direct_2";
       private static final String EXCHANGE_NAME = "exchange_direct_1";
       private static final String EXCHANGE_ROUTING_KEY2 = "direct_km2";
   
       public static void main(String[] args) throws IOException, TimeoutException {
           Connection connection = ConnectionUtils.getConnection();
           Channel channel = connection.createChannel();
           channel.queueDeclare(QUEUE_NAME, false, false, false, null);
           channel.exchangeDeclare(EXCHANGE_NAME, "direct");
           channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, EXCHANGE_ROUTING_KEY2);
           DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
               @Override
               public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                   System.out.println("消費者2接收到的消息是：" + new String(body));
               }
           };
           channel.basicConsume(QUEUE_NAME, true, defaultConsumer);
       }
   }
   

3. 測試
   先啟動2個消費者，再啟動生產(chǎn)者
   可以得到結果是消費者1得到了序號是偶數(shù)的消息
   消費者2得到了序號是奇數(shù)的消息

小結

本文介紹了 RabbitMQ 通信模型中的路由模型的使用，通過交換機和路由鍵實現(xiàn)點對點通信，適合于需要點對點通信的場景。在實際使用過程中，需要注意以下幾點：

1. 路由鍵必須要與消費者綁定隊列時的路由鍵相同，否則無法接收到消息；
2. 可以通過多個交換機和路由鍵來實現(xiàn)更靈活的消息路由。