消費者 ack 以及 生產者 confirm
https://my.oschina.net/u/3523423/blog/1620885
這篇文章主要講 RabbitMQ 中 消費者 ack 以及 生產者 confirms。
如上圖，生產者把訊息傳送到 RabbitMQ，然後 RabbitMQ 再把訊息投遞到消費者。
生產者和 RabbitMQ，以及 RabbitMQ 和消費者都是通過 TCP 連線，但是他們之間是通過通道（Channel）傳遞資料的。多個執行緒共享一個連線，但是每個執行緒擁有獨自的通道。
消費者 ack
-
問題：怎麼保證 RabbitMQ 投遞的訊息被成功投遞到了消費者？
RabbitMQ 投遞的訊息，剛投遞一半，產生了網路抖動，就有可能到不了消費者。
-
解決辦法：
RabbitMQ 對消費者說：“如果你成功接收到了訊息，給我說確認收到了，不然我就當你沒有收到，我還會重新投遞”
在 RabbitMQ 中，有兩種 acknowledgement 模式。
自動 acknowledgement 模式
這也稱作發後即忘模式。
在這種模式下，RabbitMQ 投遞了訊息，在投遞成功之前，如果消費者的 TCP 連線 或者 channel 關閉了，這條訊息就會丟失。
會有丟失訊息問題。
手動 acknowledgement 模式
在這種模式下，RabbitMQ 投遞了訊息，在投遞成功之前，如果消費者的 TCP 連線 或者 channel 關閉了，導致這條訊息沒有被 acked，RabbitMQ 會自動把當前訊息重新入隊，再次投遞。
會有重複投遞訊息的問題，所以消費者得準備好處理重複訊息的問題，就是所謂的：冪等性。
為了啟用 手動 ack 模式，消費者需要實現 ChannelAwareMessageListener 介面。
@Component public class Consumer implements ChannelAwareMessageListener { @Autowired private MessageConverter messageConverter; @Override public void onMessage(Message message, Channel channel) throws Exception { MessageProperties messageProperties = message.getMessageProperties(); // 代表投遞的識別符號，唯一標識了當前通道上的投遞，通過 deliveryTag ，消費者就可以告訴 RabbitMQ 確認收到了當前訊息，見下面的方法 long deliveryTag = messageProperties.getDeliveryTag(); // 如果是重複投遞的訊息，redelivered 為 true Boolean redelivered = messageProperties.getRedelivered(); // 獲取生產者傳送的原始訊息 Object originalMessage = messageConverter.fromMessage(message); Console.log("consume message = {} , deliveryTag = {} , redelivered = {}" , originalMessage, deliveryTag, redelivered); // 代表消費者確認收到當前訊息，第二個引數表示一次是否 ack 多條訊息 channel.basicAck(deliveryTag, false); // 代表消費者拒絕一條或者多條訊息，第二個引數表示一次是否拒絕多條訊息，第三個引數表示是否把當前訊息重新入隊 // channel.basicNack(deliveryTag, false, false); // 代表消費者拒絕當前訊息，第二個引數表示是否把當前訊息重新入隊 // channel.basicReject(deliveryTag,false); } }
-
channel.basicAck
代表消費者確認收到當前訊息，語義上表示消費者成功處理了當前訊息。
-
channel.basicNack
代表消費者拒絕一條或者多條訊息。basicNack 算是 basicReject 的一個擴充套件，因為 basicReject 不能一次拒絕多條訊息。
-
channel.basicReject
代表消費者拒絕這條訊息，語義上表示消費者沒有處理當前訊息。
對於 basicNack 和 basicReject ，如果引數
boolean requeue傳入false，訊息還是會從佇列裡面刪除。這三個方法只是語義上的不同。 -
deliveryTag
deliveryTag 是 64 bit long 值，從 1 開始，不停的遞增 1。不同的 channel 有獨立的 deliveryTag。比如有兩個消費者，你會發現，都是從 1 開始遞增，互不影響。
由於上面建立的消費者，沒有指明監聽那個佇列，所以還需要建立一個 MessageListenerContainer。
@Bean
public SimpleMessageListenerContainer container(ConnectionFactory connectionFactory, ChannelAwareMessageListener listener) {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
// 指定消費者
container.setMessageListener(listener);
// 指定監聽的佇列
container.setQueueNames(QUEUE_NAME);
// 設定消費者的 ack 模式為手動確認模式
container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
container.setPrefetchCount(300);
return container;
}
這樣就開啟了消費者手動 ack 模式。
注意
如果開啟了消費者手動 ack 模式，但是又沒有呼叫手動確認方法（比如：channel.basicAck），那問題就大了，RabbitMQ 會在當前 channel 上一直阻塞，等待消費者 ack。
生產者 confirms
-
問題：怎麼保證生產者傳送的訊息被 RabbitMQ 成功接收？
生產者傳送的訊息，剛傳送一半，產生了網路抖動，就有可能到不了 RabbitMQ。
-
解決辦法：
生產者對 RabbitMQ 說：“如果你成功接收到了訊息，給我說確認收到了，不然我就當你沒有收到”
自定義訊息元資料
/**
* 自定義訊息元資料
*/
@NoArgsConstructor
@Data
public class RabbitMetaMessage implements Serializable{
/**
* 是否是 returnCallback
*/
private boolean returnCallback;
/**
* 承載原始訊息資料資料
*/
private Object payload;
public RabbitMetaMessage(Object payload) {
this.payload = payload;
}
}
- returnCallback 標記當前訊息是否觸發了 returnCallback（後面會解釋）
- payload 儲存原始訊息資料
生產者
先把訊息儲存到 redis，再發送到 rabbitmq
@RestController
public class ProducerController {
@Autowired
private RabbitTemplate rabbitTemplate;
@Autowired
private RedisTemplate redisTemplate;
@Autowired
private DefaultKeyGenerator keyGenerator;
@GetMapping("/sendMessage")
public Object sendMessage() {
new Thread(() -> {
HashOperations hashOperations = redisTemplate.opsForHash();
for (int i = 0; i < 1; i++) {
String id = keyGenerator.generateKey() + "";
String value = "message " + i;
RabbitMetaMessage rabbitMetaMessage = new RabbitMetaMessage(value);
// 先把訊息儲存到 redis
hashOperations.put(RedisConfig.RETRY_KEY, id, rabbitMetaMessage);
Console.log("send message = {}", value);
// 再發送到 rabbitmq
rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME, RabbitMQConfig.ROUTING_KEY, value, (message) -> {
message.getMessageProperties().setMessageId(id);
return message;
}, new CorrelationData(id));
}
}).start();
return "ok";
}
}
配置 ConnectionFactory
@Bean
public ConnectionFactory connectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory("192.168.238.132", 5672);
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
// 設定 生產者 confirms
connectionFactory.setPublisherConfirms(true);
// 設定 生產者 Returns
connectionFactory.setPublisherReturns(true);
return connectionFactory;
}
配置 RabbitTemplate
@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
// 必須設定為 true，不然當 傳送到交換器成功，但是沒有匹配的佇列，不會觸發 ReturnCallback 回撥
// 而且 ReturnCallback 比 ConfirmCallback 先回調，意思就是 ReturnCallback 執行完了才會執行 ConfirmCallback
rabbitTemplate.setMandatory(true);
// 設定 ConfirmCallback 回撥
rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
Console.log("ConfirmCallback , correlationData = {} , ack = {} , cause = {} ", correlationData, ack, cause);
// 如果傳送到交換器都沒有成功（比如說刪除了交換器），ack 返回值為 false
// 如果傳送到交換器成功，但是沒有匹配的佇列（比如說取消了繫結），ack 返回值為還是 true （這是一個坑，需要注意）
if (ack) {
String messageId = correlationData.getId();
RabbitMetaMessage rabbitMetaMessage = (RabbitMetaMessage) redisTemplate.opsForHash().get(RedisConfig.RETRY_KEY, messageId);
Console.log("rabbitMetaMessage = {}", rabbitMetaMessage);
if (!rabbitMetaMessage.isReturnCallback()) {
// 到這一步才能完全保證訊息成功傳送到了 rabbitmq
// 刪除 redis 裡面的訊息
redisTemplate.opsForHash().delete(RedisConfig.RETRY_KEY, messageId);
}
}
});
// 設定 ReturnCallback 回撥
// 如果傳送到交換器成功，但是沒有匹配的佇列，就會觸發這個回撥
rabbitTemplate.setReturnCallback((message, replyCode, replyText,
exchange, routingKey) -> {
Console.log("ReturnCallback unroutable messages, message = {} , replyCode = {} , replyText = {} , exchange = {} , routingKey = {} ", message, replyCode, replyText, exchange, routingKey);
// 從 redis 取出訊息，設定 returnCallback 設定為 true
String messageId = message.getMessageProperties().getMessageId();
RabbitMetaMessage rabbitMetaMessage = (RabbitMetaMessage) redisTemplate.opsForHash().get(RedisConfig.RETRY_KEY, messageId);
rabbitMetaMessage.setReturnCallback(true);
redisTemplate.opsForHash().put(RedisConfig.RETRY_KEY, messageId, rabbitMetaMessage);
});
return rabbitTemplate;
}
ReturnCallback 回撥
必須 rabbitTemplate.setMandatory(true)，不然當 傳送到交換器成功，但是沒有匹配的佇列，不會觸發 ReturnCallback 回撥。而且 ReturnCallback 比 ConfirmCallback 先回調。
如何模擬 傳送到交換器成功，但是沒有匹配的佇列，先把專案啟動，然後再把佇列解綁，再發送訊息，就會觸發 ReturnCallback 回撥，而且發現訊息也丟失了，沒有到任何佇列。
這樣就解綁了。
執行專案，然後開啟瀏覽器，輸入 http://localhost:9999/sendMessage
控制檯打出如下日誌
這樣就觸發了 ReturnCallback 回撥 ，從 redis 取出訊息，設定 returnCallback 設定為 true。你會發現 ConfirmCallback 的 ack 返回值還是 true。
ConfirmCallback 回撥
這裡有個需要注意的地方，如果傳送到交換器成功，但是沒有匹配的佇列（比如說取消了繫結），ack 返回值為還是 true （這是一個坑，需要注意，就像上面那種情況！！！）。所以不能單靠這個來判斷訊息真的傳送成功了。這個時候會先觸發 ReturnCallback 回撥，我們把 returnCallback 設定為 true，所以還得判斷 returnCallback 是否為 true，如果為 ture，表示訊息傳送不成功，false 才能完全保證訊息成功傳送到了 rabbitmq。
如何模擬 ack 返回值為 false，先把專案啟動，然後再把交換器刪除，就會發現 ConfirmCallback 的 ack 為 false。
執行專案，然後開啟瀏覽器，輸入 http://localhost:9999/sendMessage
控制檯打出如下日誌
你會發現 ConfirmCallback 的 ack 返回值才是 false。
注意
不能單單依靠 ConfirmCallback 的 ack 返回值為 true，就斷定當前訊息傳送成功了。
原始碼地址
參考資料
Consumer Acknowledgements and Publisher Confirms
結語
由於本人知識和能力有限，文中如有沒說清楚的地方，希望大家能在評論區指出，以幫助我將博文寫得更好。