RabbitMQ 中消費者ACK與重回隊列機制

1. 前言

Hello，大家好。本小節(jié)會為大家介紹 RabbitMQ 中的消費者 ACK 與消息的重回隊列機制。消費者 ACK 與我們上節(jié)介紹的消息確認機制類似，都是針對消息而言的，而消息的重回隊列機制確是針對特殊的消息類型而來。

消費者 ACK 與消息重回隊列機制和上節(jié)中介紹的消息確認機制與消息返回機制意義相同，都屬于 RabbitMQ 自帶的補償機制，只不過他們是針對于不同的消息來說的，下面就讓我們來看一下究竟什么是 RabbitMQ 中的消費者 ACK 與重回隊列機制。

本節(jié)主要內容：

• 什么是消費者 ACK；

• 消息重回隊列機制概述；

2. 什么是消費者 ACK

基礎概念：

消費者 ACK ，是描述消息與消費者之間的一種確認關系，其主要內容就是用來監(jiān)聽，消息是否已經被消費者成功消費了。

我們都知道，當消息成功被發(fā)送到 RabbitMQ Server 中，并且經交換機和頻道，被路由到了相應的消息隊里之后，這些消息就需要等待合適的消費者來接收這些消息，并最終將這些消息進行消費。消費者 ACK 就是在這個過程中間充當了一種監(jiān)聽器的作用，主要就是用來監(jiān)聽這些消息被消費者進行消費的結果，并將消息消費的結果返回給消費者。

從上述基礎概念中，可以很清晰的得出消費者 ACK 的概念圖，如下圖所示：

根據上圖，消息在被成功發(fā)送到 RabbitMQ Server 中之后，消費端從相應的消息隊列中獲取到了消息，并在消息被消費之后，給消費端返回了 ACK 信號，這個 ACK 信號的返回一共分為兩種結果。

第一種 ACK 信號返回結果就是消息已經被成功消費了，這個時候返回給消費端的是 ack 信號，即消息消費成功的確認信號；另一種 ACK 信號返回結果是消息沒有被消費成功，這個時候返回給消費端的是 nack 信號，即消息沒有被消費者消費成功。

對于這兩種返回信號來說，第一種 ACK 返回信息是 RabbitMQ Server 中的消息一旦被成功消費后就會返回給消費端，這個過程是自動的；而另一種 ACK 信號，由于消息在沒有被成功消息后，RabbitMQ 會有自帶的解決措施，所以此種 ACK 信號需要我們手動來選擇，到底是使用 RabbitMQ 自帶的解決措施，還是使用消費者 ACK 機制。

而無論 RabbitMQ Server 返回給了消費端哪種信號，都需要我們手動來獲取消息隊列中的消息，并且手動對這些消息進行消費。

我們在了解了消費者 ACK 機制的基礎概念和作用之后，我們還需要了解在 RabbitMQ 中，如何通過代碼來實現(xiàn) RabbitMQ 的消費者 ACK 機制。

代碼實現(xiàn)：

實現(xiàn)消費者 ACK 機制，只需要在消費端進行配置即可，代碼如下：

ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("xx");
connectionFactory.setPort("5672");
connectionFactory.setVirtualHost("/");
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChanel();
DefaultConsumer defaultConsumer = new DefaultCnsumer(channel) {
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        channel.basicAck(envelope.getDeliveryTag(), false);
    }
};

代碼解釋：

第 1-5 行，我們使用 ConnectionFactory 創(chuàng)建了一個客戶端連接 RabbitMQ Server 的連接。

第 6 行，我們使用建立好的連接，來創(chuàng)建了一個頻道 channel 。

第 7-10 行，我們使用手動創(chuàng)建 DefaultConsumer 的方式來手動從消息隊列中獲取消息并進行消費，其中，此種手動創(chuàng)建 Consumer 的方式需要通過匿名內部類的方式來實現(xiàn)，在類的內部，還需要重寫 handleDelivery 方法，該方法的第一個參數，是消費者的標簽，第二個參數是當前消費者所處的環(huán)境信息，第三個參數和第四個參數分別為消息的 properties 參數和消息體。

第 12 行，我們使用了 channel 的 basicAck 方法，來對消息進行消費 ACK ，即一旦消息被成功消費，就會通過 basicAck 方法將第一種 ACK 信號返回給消費端。

Tips: 1.handleDeliery 方法是進行消息有沒有被消費的方法，我們只有重寫了該方法，才能手動監(jiān)聽到消息有沒有被消費；
2. channel 的 basicAck 方法的第一個參數就是當前消費者的標簽，第二個參數表示是否啟用 RabbitMQ 的消費者自動 ACK ，如果要想手動對消費端進行 ACK ，那么這個屬性一定不要開啟，即將該屬性的值設置為 false ，關閉消費端的 autoAck 才行。

3. 消息重回隊列機制概述

基礎概念：

消息重回隊列機制，是描述那些沒有被成功消費的消息與消費端之間的一種保障策略， 其主要目的就是為了存儲那些沒有被消費者成功消費掉的消息，即在 RabbitMQ 諸多的消息隊列中，專門用來存儲那些沒有被消費者成功消費掉的消息的隊列，這種隊列就被稱為重回隊列。

那么為什么被稱為重回隊列呢？因為，在 RabbitMQ 中，如果一條消息沒有被消費者成功消費掉，那么這條消息除了會被放到重回隊列中之外，RabbitMQ Server 還會從該重回隊列中獲取到這些消息，并且將這些沒有被消費成功的消息重新發(fā)送到 RabbitMQ Server 中，再次由消費者進行消費，直到這些消息成功地被消費者消費。

從上述消息重回隊列的基礎概念可以得出，消息重回隊列的原理圖，如下圖所示：

我們只看上圖的消費端部分，當消息隊列中的消息沒有被消費者成功消費后，首先 RabbitMQ Server 會返回給消費端一個 nack 的信號；其次，該條沒有被消費的消息會被放入重回隊列中；最后，RabbitMQ Server 會從這個重回隊列中獲取該條消息，并重新發(fā)送到 RabbitMQ Server 中等待消費。

上述過程就是我們的重回隊列機制處理未被消費的消息的全過程，同學們需要對此有所了解才行。

接下來我們來看一下，消息重回隊列如何來進行代碼配置。

代碼實現(xiàn)：

實現(xiàn)消息重回隊列機制，也是只需要在消費端進行配置即可，代碼如下：

// 省略客戶端連接 RabbitMQ Server 的過程
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChanel();
DefaultConsumer defaultConsumer = new DefaultCnsumer(channel) {
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        if (properties.getHeaders().get("type") != "utf-8") {
            channel.basicNack(envelope.getDeliveryTag(), false, true)
        }
    }
};

代碼解釋：

第 1-2 行，我們創(chuàng)建了客戶端連接 RabbitMQ Server 的連接，并且創(chuàng)建了一個 channel 。

第 8-11 行，我們在 handleDelivery 方法的內部，添加了一個判斷條件，當滿足該條件時，意味著該條消息沒有被成功消費，返回給消費端 nack 信號，這個過程是通過 channel 的 basicNack 方法實現(xiàn)。

Tips: 1. 在 channel 的 basicNack 方法中，方法的第二個參數表示是否批量確認，這里設置為了 false ，表示不進行批量確認，第三個參數表示是否啟用消息重回隊列，這里設置為了 true ，表示啟用重回隊列，默認情況下，該屬性是 false ，即不啟用重回隊列；
2. 在實際工作中，我們一般也會關閉消息重回隊列，使用 RabbitMQ 自帶的措施和打印日志的方式來進行補償，RabbitMQ 默認會將那些沒有被成功消費的消息刪除，不會放入重回隊列中。

4. 小結

本小節(jié)為同學們介紹了 RabbitMQ 中的消費者 ACK ，以及消息重回隊里機制。從消費者 ACK 與消息重回隊列機制的基礎概念開始，到不同機制的代碼實現(xiàn)結束，詳細介紹了什么是消費者 ACK、什么是重回隊列，以及消息重回隊列機制，且通過不同機制的代碼實現(xiàn)，分別闡述了如何通過代碼來對兩種機制進行配置。希望同學們可以完全理解兩種機制的基礎概念和實現(xiàn)方式。