1. 本篇概要

其實，還有1種場景需要考慮：當消費者接收到消息後，還沒處理完業務邏輯，消費者掛掉了，那消息也算丟失了？，比如用戶下單，訂單中心發送了1個消息到RabbitMQ里的隊列，積分中心收到這個消息，準備給這個下單的用戶增加20積分，但積分還沒增加成功呢，積分中心自己掛掉了，導致數據出現問題。

那麼如何解決這種問題呢？

為了保證消息被消費者成功的消費，RabbitMQ提供了消息確認機制(message acknowledgement)，本文主要講解RabbitMQ中，如何使用消息確認機制來保證消息被消費者成功的消費，避免因為消費者突然宕機而引起的消息丟失。

2. 開啟顯式Ack模式

我們開啟一個消費者的代碼是這樣的：

// 創建隊列消費者
com.rabbitmq.client.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("Received Message " + message + "");
}
};
channel.basicConsume(QUEUE_NAME, true, consumer);

這裡的重點是channel.basicConsume(QUEUE_NAME, true, consumer);方法的第2個參數，讓我們先看下basicConsume()的源碼：

public String basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException {
return this.basicConsume(queue, autoAck, "", callback);
}

這裡的autoAck參數指的是是否自動確認，如果設置為ture，RabbitMQ會自動把發送出去的消息置為確認，然後從內存（或者磁碟）中刪除，而不管消費者接收到消息是否處理成功；如果設置為false，RabbitMQ會等待消費者顯式的回復確認信號後才會從內存（或者磁碟）中刪除。

建議將autoAck設置為false，這樣消費者就有足夠的時間處理消息，不用擔心處理消息過程中消費者宕機造成消息丟失。

此時，隊列里的消息就分成了2個部分：

  1. 等待投遞給消費者的消息(下圖中的Ready部分)
  2. 已經投遞給消費者，但是還沒有收到消費者確認信號的消息(下圖中的Unacked部分)

如果RabbitMQ一直沒有收到消費者的確認信號，並且消費此消息的消費者已經斷開連接，則RabbitMQ會安排該消息重新進入隊列，等待投遞給下一個消費者，當然也有可能還是原來的那個消費者。

RabbitMQ不會為未確認的消息設置過期時間，它判斷此消息是否需要重新投遞給消費者的唯一依據是消費該消息的消費者連接是否已經斷開，這麼設計的原因是RabbitMQ允許消費者消費一條消息的時間可以很久很久。

為了便於理解，我們舉個具體的例子，生產者的話的我們延用上文中的DurableProducer：

package com.zwwhnly.springbootaction.rabbitmq.durable;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class DurableProducer {
private final static String EXCHANGE_NAME = "durable-exchange";
private final static String QUEUE_NAME = "durable-queue";

public static void main(String[] args) throws IOException, TimeoutException {
// 創建連接
ConnectionFactory factory = new ConnectionFactory();
// 設置 RabbitMQ 的主機名
factory.setHost("localhost");
// 創建一個連接
Connection connection = factory.newConnection();
// 創建一個通道
Channel channel = connection.createChannel();
// 創建一個Exchange
channel.exchangeDeclare(EXCHANGE_NAME, "direct", true);
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");

// 發送消息
String message = "durable exchange test";
AMQP.BasicProperties props = new AMQP.BasicProperties().builder().deliveryMode(2).build();
channel.basicPublish(EXCHANGE_NAME, "", props, message.getBytes());

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

然後新建一個消費者AckConsumer類：

package com.zwwhnly.springbootaction.rabbitmq.ack;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class AckConsumer {
private final static String QUEUE_NAME = "durable-queue";

public static void main(String[] args) throws IOException, TimeoutException {
// 創建連接
ConnectionFactory factory = new ConnectionFactory();
// 設置 RabbitMQ 的主機名
factory.setHost("localhost");
// 創建一個連接
Connection connection = factory.newConnection();
// 創建一個通道
Channel channel = connection.createChannel();
// 創建隊列消費者
com.rabbitmq.client.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");
int result = 1 / 0;
System.out.println("Received Message " + message + "");
}
};
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}

我們先將autoAck參數設置為ture，即自動確認，並在消費消息時故意寫個異常，然後先運行生產者客戶端將消息寫入隊列中，然後運行消費者客戶端，發現消息未消費成功但是卻消失了：

然後我們將autoAck設置為false：

channel.basicConsume(QUEUE_NAME, false, consumer);

再次運行生產者客戶端將消息寫入隊列中，然後運行消費者客戶端，此時雖然消費者客戶端仍然代碼異常，但是消息仍然在隊列中：

然後我們刪除掉消費者客戶端中的異常代碼，重新啟動消費者客戶端，發現消息消費成功了，但是消息一直未Ack：

手動停掉消費者客戶端，發現消息又到了Ready狀態，準備重新投遞：

之所以消費掉消息，卻一直還是Unacked狀態，是因為我們沒在代碼中添加顯式的Ack代碼：

String message = new String(body, "UTF-8");
//int result = 1 / 0;
System.out.println("Received Message " + message + "");

long deliveryTag = envelope.getDeliveryTag();
channel.basicAck(deliveryTag, false);

deliveryTag可以看做消息的編號，它是一個64位的長整形值。

此時運行消費者客戶端，發現消息消費成功，並且在隊列中被移除：

本文的重點是你有沒有收穫與成長，其餘的都不重要，希望讀者們能謹記這一點。同時我經過多年的收藏目前也算收集到了一套完整的學習資料，包括但不限於：分散式架構、高可擴展、高性能、高並發、Jvm性能調優、Spring，MyBatis，Nginx源碼分析，Redis，ActiveMQ、、Mycat、Netty、Kafka、Mysql、Zookeeper、Tomcat、Docker、Dubbo、Nginx等多個知識點高級進階乾貨，希望對想成為架構師的朋友有一定的參考和幫助

喜歡這篇文章的可以點個贊，也可以關注我的專欄java經驗分享,專欄頂部有免費獲取資料方式

下面是部分資料截圖，誠意滿滿，感興趣的可以後台私信「資料」即可免費獲取


推薦閱讀：
相关文章