RabbitMQ的基本介紹及與Spring整合

一，場景回顧

​ 最近做電商購物項目，在分布式中搜索服務，商品詳情服務都是獨立的模塊。那么有一個問題就是：

• 商品的原始數據保存在數據庫中，增刪改查都在數據庫中完成。
• 搜索服務數據來源是索引庫，如果數據庫商品發生變化，索引庫數據不能及時更新。
• 商品詳情做了頁面靜態化，靜態頁面數據也不會隨着數據庫商品發生變化。

​ 如果我們在后台修改了商品的價格，搜索頁面和商品詳情頁顯示的依然是舊的價格，這樣顯然不對。該如何解決？

​ 通常有兩種解決方案：

• 方案1：每當后台對商品做增刪改操作，同時要修改索引庫數據及靜態頁面
• 方案2：搜索服務和商品頁面服務對外提供操作接口，后台在商品增刪改后，調用接口

​ 以上兩種方式都有同一個嚴重問題：就是代碼耦合，后台服務中需要嵌入搜索和商品頁面服務，違背了微服務的獨立原則。

​ 所以，我們會通過另外一種方式來解決這個問題：消息隊列

二，消息隊列

2.1，什么是消息隊列

​ 首先我們應該知道隊列（Queue）是一種數據結構，一種特殊的線性表。消息隊列（Message Queue）是應用程序之間相互通信的方法，我們可以將消息隊列比作為一個容器。提供消息的一方將消息存放到消息隊列中，而要接收消息的一方同樣去消息隊列中去獲取，因而消息隊列是分布式系統中一個重要的中間組件。使用消息隊列異步處理系統的性能，降低系統的耦合性，實現高可用。

​ 消息隊列其實是一種典型的生產者，消費者模型。生產者不斷向消息隊列中生產消息，消費者不斷從消息隊列中獲取消息，兩者之間都是通過異步進行的。因此只關心消息的生產和消費，兩者之間沒有業務代碼的侵入，實現了生產者和消費者之間的解耦。

結合場景回顧所說的問題：

• 商品服務對商品增刪改以后，無需去操作索引庫或靜態頁面，只是發送一條消息，也不關心消息被誰接收。
• 搜索服務和靜態頁面服務接收消息，分別去處理索引庫和靜態頁面。

如果以后有其它系統也依賴商品服務的數據，同樣監聽消息即可，商品服務無需任何代碼修改。

2.2，AMQP和JMS

​ MQ是消息通信的模型，並不是具體實現。現在實現MQ的有兩種主流方式：AMQP、JMS。

2.2.1，AMQP

​ AMQP（advanced message queuing protocol）在2003年時被提出，最早用於解決金融領不同平台之間的消息傳遞交互問題。顧名思義，AMQP是一種協議，更准確的說是一種binary wire-level protocol（鏈接協議）。這是其和JMS的本質差別，AMQP不從API層進行限定，而是直接定義網絡交換的數據格式。這使得實現了AMQP的provider天然性就是跨平台的。意味着我們可以使用Java的AMQP provider，同時使用一個python的producer加一個rubby的consumer。從這一點看，AQMP可以用http來進行類比，不關心實現的語言，只要大家都按照相應的數據格式去發送報文請求，不同語言的client均可以和不同語言的server鏈接。

2.2.2， JMS

​ 通常而言提到JMS（Java Message Service）實際上是指JMS API。JMS是由Sun公司早期提出的消息標准，旨在為java應用提供統一的消息操作，包括create，send、receive等。JMS已經成為Java Enterprise Edition的一部分。從使用角度看，JMS和JDBC擔任差不多的角色，用戶都是根據相應的接口可以和實現了JMS的服務進行通信，進行相關的操作。

​ 兩者之間的區別：

	AMQP	JMS
平台	多技術平台	針對JAVA
消息模型	direct，topic，fanout，headers	點對點，發布訂閱模型
消息數據類型	二進制消息	StreamMessage，MapMessage，TextMessage，ObjectMessage和BytesMessage
消息路由	routing key	使用客戶端的選擇過濾器實現路由

2.3，常見的MQ產品

• ActiveMQ：基於JMS
• RabbitMQ：基於AMQP協議，erlang語言開發，穩定性好
• RocketMQ：基於JMS，阿里巴巴產品，目前交由Apache基金會
• Kafka：分布式消息系統，高吞吐量

三，RabbitMQ

​ RabbitMQ是一個開源的AMQP實現的消息管理系統，服務器端用Erlang語言編寫，支持多種客戶端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用於在分布式系統中存儲轉發消息，在易用性、擴展性、高可用性等方面非常優秀。

​ 官網地址：https://www.rabbitmq.com/

​ 在使用RabbitMQ之前應將環境准備好，該文章中不講述安裝步驟，網上有很多優秀的博客，大家可以去參考以下。

​ RabbitMQ提供6種消息模型，但是最后一種為RPC遠程調用，所以只有５種。

​ 使用之前導入RabbitMQ依賴，后面會介紹與spring boot的整合。

<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

​ 自定義一個連接工具類，方便多程序使用。

public class Rec {
    private static final String QUEUE_NAME = "simple_queue";
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1，獲取連接
        Connection connection = ConnectUtil.getConnection();
        // 2，創建通道
        Channel channel = connection.createChannel();
        // 3，聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 4，聲明消費者
        DefaultConsumer consumer = new DefaultConsumer(channel) {// 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                       byte[] body) {
                // body 即消息體
                String msg = new String(body);
                System.out.println(" [x] received : " + msg + "!");
            }
        };

        // 5，監聽隊列
        channel.basicConsume(QUEUE_NAME,true,consumer);
    }
}

3.1，基本消息模型（simple）

​ 基本模型介紹：

​ 文中指出，我們可以將RabbitMQ比作一個郵局，發件人將信封放到郵箱，此時郵差先生拿到你的郵件根據上面地址可以郵遞到指定的收件人種。在這個過程中RabbitMQ充當郵局，郵差，郵箱三種角色。

​ 基本模型如圖：

​ P（producer/ publisher）：生產者，一個發送消息的用戶應用程序。

​ C（consumer）：消費者，用於接收並處理邏輯的應用程序。

​ 隊列（紅色區域）：rabbitmq內部類似於郵箱的一個概念。雖然消息流經rabbitmq和你的應用程序，但是它們只能存儲在隊列中。隊列只受主機的內存和磁盤限制，實質上是一個大的消息緩沖區。許多生產者可以發送消息到一個隊列，許多消費者可以嘗試從一個隊列接收數據。

​ 接下來通過代碼來實現。

​ 發送者：

public class send {

    private static final String QUEUE_NAME = "simple_queue";

    public static void main(String[] args) throws IOException, TimeoutException {

        // 1，獲取連接
        Connection connection = ConnectionUtil.getConnection();
        // 2，創建通道
        Channel channel = connection.createChannel();
        // 3，聲明隊列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        // 4，定義消息
        String msg = "Hello world";
        // 5，發送消息
        channel.basicPublish(null,QUEUE_NAME,null,msg.getBytes());
        System.out.println(" [x] Sent '" + msg + "'");

        //關閉通道和連接
        channel.close();
        connection.close();
    }
}

​ 結果如下：

消費者1：

public class Rec {

    private static final String QUEUE_NAME = "simple_queue";

    public static void main(String[] args) throws IOException, TimeoutException {

        // 1，獲取連接
        Connection connection = ConnectionUtil.getConnection();
        // 2，創建通道
        Channel channel = connection.createChannel();
        // 3，聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 4，聲明消費者
        DefaultConsumer consumer = new DefaultConsumer(channel) {// 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                       byte[] body) {
                // body 即消息體
                String msg = new String(body);
                System.out.println(" [x] received : " + msg + "!");
            }
        };
        // 5，監聽隊列
        channel.basicConsume(QUEUE_NAME,true,consumer);
    }
}

​ 接收者結果如圖：

3.1.1，消息確認機制（ACK）--經典面試問題

通過剛才的案例可以看出，消息一旦被消費者接收，隊列中的消息就會被刪除。

那么問題來了：RabbitMQ怎么知道消息被接收了呢？

如果消費者領取消息后，還沒執行操作就掛掉了呢？或者拋出了異常？消息消費失敗，但是RabbitMQ無從得知，這樣消息就丟失了！

因此，RabbitMQ有一個ACK機制。當消費者獲取消息后，會向RabbitMQ發送回執ACK，告知消息已經被接收。不過這種回執ACK分兩種情況：

• 自動ACK：消息一旦被接收，消費者自動發送ACK
• 手動ACK：消息接收后，不會發送ACK，需要手動調用

那么該如何選擇呢？？

這需要看消息的重要性：

• 如果消息不太重要，丟失也沒有影響，那么自動ACK會比較方便
• 如果消息非常重要，不容丟失。那么最好在消費完成后手動ACK，否則接收消息后就自動ACK，RabbitMQ就會把消息從隊列中刪除。如果此時消費者宕機，那么消息就丟失了。

上圖中有一行代碼為：

其中第二個參數就是是否使用ACK機制，true代表使用，false代表不使用。

源碼如圖：

​ 但是需要注意的是，手動ACK存在一定的問題。如果消息接收方已經接收但是在業務處理過程中出現異常，那么該消息並未處理完成。但此時消息隊列中消息已經被自動刪除，因而就會造成消息丟失，這是個非常嚴重的問題。

​ 解決方法：將autoAck改為false，並且添加如下代碼

 // 手動進行ACK
channel.basicAck(envelope.getDeliveryTag(), false);

​ 當ACK機制改為手動時，只有當消息真正消費結束時，通知消息隊列並刪除消息。

3.2 ，Work消息模型

​

​ 介紹：

​ 工作隊列，又稱任務隊列。主要思想就是避免執行資源密集型任務時，必須等待它執行完成。相反我們稍后完成任務，我們將任務封裝為消息並將其發送到隊列。 在后台運行的工作進程將獲取任務並最終執行作業。當你運行許多工人時，任務將在他們之間共享，但是一個消息只能被一個消費者獲取。

​ 這個概念在Web應用程序中特別有用，因為在短的HTTP請求窗口中無法處理復雜的任務。

​ 接下來我們來模擬這個流程：

​ P：生產者：任務的發布者

​ C1：消費者，領取任務並且完成任務，假設完成速度較快

C2：消費者2：領取任務並完成任務，假設完成速度慢

​ 接下來生產者發送50條數據：

public class Send {
    private final static String QUEUE_NAME = "Work_queue";

    public static void main(String[] argv) throws Exception {
        // 獲取到連接
        Connection connection = ConnectionUtil.getConnection();
        // 獲取通道
        Channel channel = connection.createChannel();
        // 聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 循環發布任務
        for (int i = 0; i < 50; i++) {
            // 消息內容
            String message = "task .. " + i;
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");

            Thread.sleep(i * 2);
        }
        // 關閉通道和連接
        channel.close();
        connection.close();
    }
}

​ 消費者1：該消費者模擬性能較差的情況下

public class Recv {
    private final static String QUEUE_NAME = "Work_queue";

    public static void main(String[] argv) throws Exception {
        // 獲取到連接
        Connection connection = ConnectionUtil.getConnection();
        // 獲取通道
        final Channel channel = connection.createChannel();
        // 聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 設置每個消費者同時只能處理一條消息
        channel.basicQos(1);
        // 定義隊列的消費者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息體
                String msg = new String(body);
                System.out.println(" [消費者1] received : " + msg + "!");
                try {
                    // 模擬完成任務的耗時：1000ms
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                }
                // 手動ACK
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        // 監聽隊列。
        channel.basicConsume(QUEUE_NAME, false, consumer);
    }
}

​ 消費者2：

public class Recv2 {
    private final static String QUEUE_NAME = "Work_queue";

    public static void main(String[] argv) throws Exception {
        // 獲取到連接
        Connection connection = ConnectionUtil.getConnection();
        // 獲取通道
        final Channel channel = connection.createChannel();
        // 聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 設置每個消費者同時只能處理一條消息
        channel.basicQos(1);
        // 定義隊列的消費者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息體
                String msg = new String(body);
                System.out.println(" [消費者2] received : " + msg + "!");
                // 手動ACK
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        // 監聽隊列。
        channel.basicConsume(QUEUE_NAME, false, consumer);
    }
}

​ 結果如下圖：

​ 從圖中看出發送者50條消息被兩個消費者各消費25條，以此可以提高效率，實現任務分配。

​ 經典面試題：如何避免消息堆積？

​ 1） 采用workqueue，多個消費者監聽同一隊列。

​ 2）接收到消息以后，通過線程池，異步消費。

3.2.1，能者多勞

但是剛才的實現有一個問題：

• 消費者1比消費者2的效率要低，一次任務的耗時較長
• 然而兩人最終消費的消息數量是一樣的
• 消費者2大量時間處於空閑狀態，消費者1一直忙碌

現在的狀態屬於是把任務平均分配，正確的做法應該是消費越快的人，消費的越多。

​ 我們可以使用basicQos方法和prefetchCount = 1設置。 這告訴RabbitMQ一次不要向消費者發送多於一條消息。 或者換句話說，不要向消費者發送新消息，直到它處理並確認了前一個消息。 相反，它會將其分派給不是仍然忙碌的下一個工消費者。

// 設置每個消費者同時只能處理一條消息
channel.basicQos(1);

再次運行結果：

3.3，訂閱者消息模型（Publish/Subscribe）

詳解：

1、1個生產者，多個消費者

2、每一個消費者都有自己的一個隊列

3、生產者沒有將消息直接發送到隊列，而是發送到了交換機

4、每個隊列都要綁定到交換機

5、生產者發送的消息，經過交換機到達隊列，實現一個消息被多個消費者獲取的目的

X（Exchanges）：交換機一方面：接收生產者發送的消息。另一方面：知道如何處理消息，例如遞交給某個特別隊列、遞交給所有隊列、或是將消息丟棄。到底如何操作，取決於Exchange的類型。

Exchange類型有以下幾種：

​ Fanout：廣播，將消息交給所有綁定到交換機的隊列

​ Direct：定向，把消息交給符合指定routing key 的隊列

​ Topic：通配符，把消息交給符合routing pattern（路由模式） 的隊列

​ 注意：Exchange（交換機）只負責轉發消息，不具備存儲消息的能力，因此如果沒有任何隊列與Exchange綁定，或者沒有符合路由規則的隊列，那么消息會丟失！

3.3.1，Fanout：廣播模式

​ 在廣播模式下，消息發送流程是這樣的：

• 1） 可以有多個消費者

• 2） 每個消費者有自己的queue（隊列）

• 3） 每個隊列都要綁定到Exchange（交換機）

• 4） 生產者發送的消息，只能發送到交換機，交換機來決定要發給哪個隊列，生產者無法決定。

• 5） 交換機把消息發送給綁定過的所有隊列

• 6） 隊列的消費者都能拿到消息。實現一條消息被多個消費者消費

  創建生產者：

  1） 聲明Exchange，不再聲明Queue

  2） 發送消息到Exchange，不再發送到Queue

  public class Send {
  
      private final static String EXCHANGE_NAME = "fanout_exchange";
  
      public static void main(String[] argv) throws Exception {
          // 獲取到連接
          Connection connection = ConnectionUtil.getConnection();
          // 獲取通道
          Channel channel = connection.createChannel();
          
          // 聲明exchange，指定類型為fanout
          channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
          
          // 消息內容
          String message = "Hello everyone";
          // 發布消息到Exchange
          channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
          System.out.println(" [生產者] Sent '" + message + "'");
  
          channel.close();
          connection.close();
      }
  }
  

  ​ 消費者1：

  public class Recv {
      private final static String QUEUE_NAME = "fanout_exchange_queue_1";
  
      private final static String EXCHANGE_NAME = "fanout_exchange";
  
      public static void main(String[] argv) throws Exception {
          // 獲取到連接
          Connection connection = ConnectionUtil.getConnection();
          // 獲取通道
          Channel channel = connection.createChannel();
          // 聲明隊列
          channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  
          // 綁定隊列到交換機
          channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
  
          // 定義隊列的消費者
          DefaultConsumer consumer = new DefaultConsumer(channel) {
              // 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
              @Override
              public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                      byte[] body) throws IOException {
                  // body 即消息體
                  String msg = new String(body);
                  System.out.println(" [消費者1] received : " + msg + "!");
              }
          };
          // 監聽隊列，自動返回完成
          channel.basicConsume(QUEUE_NAME, true, consumer);
      }
  }
  

  要注意代碼中：隊列需要和交換機綁定

  消費者2：

  public class Recv2 {
      private final static String QUEUE_NAME = "fanout_exchange_queue_2";
  
      private final static String EXCHANGE_NAME = "fanout_exchange";
  
      public static void main(String[] argv) throws Exception {
          // 獲取到連接
          Connection connection = ConnectionUtil.getConnection();
          // 獲取通道
          Channel channel = connection.createChannel();
          // 聲明隊列
          channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  
          // 綁定隊列到交換機
          channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
          
          // 定義隊列的消費者
          DefaultConsumer consumer = new DefaultConsumer(channel) {
              // 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
              @Override
              public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                      byte[] body) throws IOException {
                  // body 即消息體
                  String msg = new String(body);
                  System.out.println(" [消費者2] received : " + msg + "!");
              }
          };
          // 監聽隊列，手動返回完成
          channel.basicConsume(QUEUE_NAME, true, consumer);
      }
  }
  

  ​ 運行結果:

  ​

    ![](https://img2018.cnblogs.com/blog/1655301/201908/1655301-20190808204357990-1531661533.png)
  

  3.3.2， Direct：定向模式

  有選擇性的接收消息

  在訂閱模式中，生產者發布消息，所有消費者都可以獲取所有消息。

  在路由模式中，我們將添加一個功能 - 我們將只能訂閱一部分消息。 例如，我們只能將重要的錯誤消息引導到日志文件（以節省磁盤空間），同時仍然能夠在控制台上打印所有日志消息。

  但是，在某些場景下，我們希望不同的消息被不同的隊列消費。這時就要用到Direct類型的Exchange。

  在Direct模型下，隊列與交換機的綁定，不能是任意綁定了，而是要指定一個RoutingKey（路由key）

  消息的發送方在向Exchange發送消息時，也必須指定消息的routing key。

P：生產者，向Exchange發送消息，發送消息時，會指定一個routing key。

X：Exchange（交換機），接收生產者的消息，然后把消息遞交給 與routing key完全匹配的隊列

C1：消費者，其所在隊列指定了需要routing key 為 error 的消息

C2：消費者，其所在隊列指定了需要routing key 為 info、error、warning 的消息

生產者：

此處我們模擬商品的增刪改，發送消息的RoutingKey分別是：insert、update、delete

public class Send {
    private final static String EXCHANGE_NAME = "direct_exchange";

    public static void main(String[] argv) throws Exception {
        // 獲取到連接
        Connection connection = ConnectionUtil.getConnection();
        // 獲取通道
        Channel channel = connection.createChannel();
        // 聲明exchange，指定類型為direct
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        // 消息內容
        String message = "商品新增了， id = 1001";
        // 發送消息，並且指定routing key 為：insert ,代表新增商品
        channel.basicPublish(EXCHANGE_NAME, "insert", null, message.getBytes());
        System.out.println(" [商品服務：] Sent '" + message + "'");

        channel.close();
        connection.close();
    }
}

​ 消費者：

​ 我們此處假設消費者1只接收兩種類型的消息：更新商品和刪除商品。

public class Recv {
    private final static String QUEUE_NAME = "direct_exchange_queue_1";
    private final static String EXCHANGE_NAME = "direct_exchange";

    public static void main(String[] argv) throws Exception {
        // 獲取到連接
        Connection connection = ConnectionUtil.getConnection();
        // 獲取通道
        Channel channel = connection.createChannel();
        // 聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        // 綁定隊列到交換機，同時指定需要訂閱的routing key。假設此處需要update和delete消息
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");

        // 定義隊列的消費者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息體
                String msg = new String(body);
                System.out.println(" [消費者1] received : " + msg + "!");
            }
        };
        // 監聽隊列，自動ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

​ 消費者2：

​ 我們此處假設消費者2接收所有類型的消息：新增商品，更新商品和刪除商品。

public class Recv2 {
    private final static String QUEUE_NAME = "direct_exchange_queue_2";
    private final static String EXCHANGE_NAME = "direct_exchange";

    public static void main(String[] argv) throws Exception {
        // 獲取到連接
        Connection connection = ConnectionUtil.getConnection();
        // 獲取通道
        Channel channel = connection.createChannel();
        // 聲明隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        // 綁定隊列到交換機，同時指定需要訂閱的routing key。訂閱 insert、update、delete
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "insert");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");

        // 定義隊列的消費者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                    byte[] body) throws IOException {
                // body 即消息體
                String msg = new String(body);
                System.out.println(" [消費者2] received : " + msg + "!");
            }
        };
        // 監聽隊列，自動ACK
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

3.3.3，Topic：通配符

Topic類型的Exchange與Direct相比，都是可以根據RoutingKey把消息路由到不同的隊列。只不過Topic類型Exchange可以讓隊列在綁定Routing key 的時候使用通配符！

Routingkey 一般都是有一個或多個單詞組成，多個單詞之間以”.”分割，例如： item.insert

通配符規則：

​ #：匹配一個或多個詞

​ *：匹配不多不少恰好1個詞

生產者在路由中指定insert操作：

 // 發送消息，並且指定routing key 為：insert ,代表新增商品
channel.basicPublish(EXCHANGE_NAME, "yx.insert", null, message.getBytes());

消費者1的隊列接收更新，刪除操作：

 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.update");
  channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.delete");

消費者2

 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.insert");

3.4，持久化

如何避免消息丟失？

1） 消費者的ACK機制。可以防止消費者丟失消息。

2） 但是，如果在消費者消費之前，RabbitMQ就宕機了，消息就沒了。

如何解決？

應該將隊列、Exchange，消息都進行持久化。

隊列持久化：

交換機持久化：

消息持久化：

四，MQ與Spring Boot整合

注意： Spring-amqp是對AMQP協議的抽象實現，而spring-rabbit 是對協議的具體實現，也是目前的唯一實現。底層使用的就是RabbitMQ。

4.1，配置和依賴

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

application.yml中配置信息：

spring:
  rabbitmq:
    host: 192.168.129.130
    username: yx
    password: yx
    virtual-host: /yx

4.2，編寫監聽者

​ 在SpringAmqp中，對消息的消費者進行了封裝和抽象，一個普通的JavaBean中的普通方法，只要通過簡單的注解，就可以成為一個消費者。

@Component
public class Listener {
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "spring.queue", durable = "true"),
            exchange = @Exchange(
                    value = "spring.exchange",
                    ignoreDeclarationExceptions = "true",
                    type = ExchangeTypes.TOPIC
            ),
            key = {"#.#"}))
    public void listen(String msg){
        System.out.println("接收到消息：" + msg);
    }
}

五，總結

​ 以上就是RabbitMQ所提供的消息模型，總體來說大致為三種，一種基本模型，一種work模型，而剩下的訂閱模型都大同小異。並沒有涉及到一些難點，只是有些需要注意的細節問題。文章中如有不恰當之處，歡迎大牛評論指教。