1. 配置RabbitMQ
# 發(fā)送確認
spring.rabbitmq.publisher-confirms=true
# 發(fā)送回調(diào)
spring.rabbitmq.publisher-returns=true
# 消費手動確認
spring.rabbitmq.listener.simple.acknowledge-mode=manual
2. 生產(chǎn)者發(fā)送消息確認機制
- 其實這個也不能叫確認機制，只是起到一個監(jiān)聽的作用，監(jiān)聽生產(chǎn)者是否發(fā)送消息到exchange和queue。
- 生產(chǎn)者和消費者代碼不改變。
- 新建配置類 MQProducerAckConfig.java 實現(xiàn)ConfirmCallback和ReturnCallback接口，@Component注冊成組件。
- ConfirmCallback只確認消息是否到達exchange，已實現(xiàn)方法confirm中ack屬性為標準，true到達，反之進入黑洞。
- ReturnCallback消息沒有正確到達隊列時觸發(fā)回調(diào)，如果正確到達隊列不執(zhí)行。
package com.fzb.rabbitmq.config;
import org.apache.commons.lang3.SerializationUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
/**
* @Description 消息發(fā)送確認
* <p>
* ConfirmCallback 只確認消息是否正確到達 Exchange 中
* ReturnCallback 消息沒有正確到達隊列時觸發(fā)回調(diào)，如果正確到達隊列不執(zhí)行
* <p>
* 1. 如果消息沒有到exchange,則confirm回調(diào),ack=false
* 2. 如果消息到達exchange,則confirm回調(diào),ack=true
* 3. exchange到queue成功,則不回調(diào)return
* 4. exchange到queue失敗,則回調(diào)return
* @Author jxb
* @Date 2019-04-04 16:57:04
*/
@Component
public class MQProducerAckConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
@Autowired
private RabbitTemplate rabbitTemplate;
@PostConstruct
public void init() {
rabbitTemplate.setConfirmCallback(this); //指定 ConfirmCallback
rabbitTemplate.setReturnCallback(this); //指定 ReturnCallback
}
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
if (ack) {
System.out.println("消息發(fā)送成功" + correlationData);
} else {
System.out.println("消息發(fā)送失敗:" + cause);
}
}
@Override
public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
// 反序列化對象輸出
System.out.println("消息主體: " + SerializationUtils.deserialize(message.getBody()));
System.out.println("應(yīng)答碼: " + replyCode);
System.out.println("描述：" + replyText);
System.out.println("消息使用的交換器 exchange : " + exchange);
System.out.println("消息使用的路由鍵 routing : " + routingKey);
}
}
3. 消費者消息手動確認
- SpringBoot集成RabbitMQ確認機制分為三種：none、auto(默認)、manual
Auto：
1. 如果消息成功被消費（成功的意思是在消費的過程中沒有拋出異常），則自動確認
2. 當(dāng)拋出 AmqpRejectAndDontRequeueException 異常的時候，則消息會被拒絕，且 requeue = false（不重新入隊列）
3. 當(dāng)拋出 ImmediateAcknowledgeAmqpException 異常，則消費者會被確認
4. 其他的異常，則消息會被拒絕，且 requeue = true，此時會發(fā)生死循環(huán)，可以通過 setDefaultRequeueRejected（默認是true）去設(shè)置拋棄消息
如設(shè)置成manual手動確認，一定要對消息做出應(yīng)答，否則rabbit認為當(dāng)前隊列沒有消費完成，將不再繼續(xù)向該隊列發(fā)送消息。
channel.basicAck(long,boolean); 確認收到消息，消息將被隊列移除，false只確認當(dāng)前consumer一個消息收到，true確認所有consumer獲得的消息。
channel.basicNack(long,boolean,boolean); 確認否定消息，第一個boolean表示一個consumer還是所有，第二個boolean表示requeue是否重新回到隊列，true重新入隊。
channel.basicReject(long,boolean); 拒絕消息，requeue=false 表示不再重新入隊，如果配置了死信隊列則進入死信隊列。
當(dāng)消息回滾到消息隊列時，這條消息不會回到隊列尾部，而是仍是在隊列頭部，這時消費者會又接收到這條消息，如果想消息進入隊尾，須確認消息后再次發(fā)送消息。
channel.basicPublish(message.getMessageProperties().getReceivedExchange(),
message.getMessageProperties().getReceivedRoutingKey(),
MessageProperties.PERSISTENT_TEXT_PLAIN,
message.getBody());
- 延續(xù)上一章direct類型隊列為例，當(dāng)消息出現(xiàn)異常，判斷是否回滾過消息，如否則消息從新入隊，反之拋棄消息。其中一個消費者模擬一個異常。
@RabbitListener(bindings = {@QueueBinding(value = @Queue(value = "direct.queue"), exchange = @Exchange(value = "direct.exchange"), key = "HelloWorld")})
public void getDirectMessage(User user, Channel channel, Message message) throws IOException {
try {
// 模擬執(zhí)行任務(wù)
Thread.sleep(1000);
// 模擬異常
String is = null;
is.toString();
// 確認收到消息，false只確認當(dāng)前consumer一個消息收到，true確認所有consumer獲得的消息
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
} catch (Exception e) {
if (message.getMessageProperties().getRedelivered()) {
System.out.println("消息已重復(fù)處理失敗,拒絕再次接收" + user.getName());
// 拒絕消息，requeue=false 表示不再重新入隊，如果配置了死信隊列則進入死信隊列
channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
} else {
System.out.println("消息即將再次返回隊列處理" + user.getName());
// requeue為是否重新回到隊列，true重新入隊
channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
}
//e.printStackTrace();
}
}
@RabbitListener(queues = "direct.queue")
public void getDirectMessageCopy(User user, Channel channel, Message message) throws IOException {
try {
// 模擬執(zhí)行任務(wù)
Thread.sleep(1000);
System.out.println("--jxb--MQConsumer--getDirectMessageCopy：" + user.toString());
// 確認收到消息，false只確認當(dāng)前consumer一個消息收到，true確認所有consumer獲得的消息
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
} catch (Exception e) {
if (message.getMessageProperties().getRedelivered()) {
System.out.println("消息已重復(fù)處理失敗,拒絕再次接收！");
// 拒絕消息，requeue=false 表示不再重新入隊，如果配置了死信隊列則進入死信隊列
channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
} else {
System.out.println("消息即將再次返回隊列處理！");
// requeue為是否重新回到隊列，true重新入隊
channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
}
e.printStackTrace();
}
}
從執(zhí)行結(jié)果來看，三條消息都調(diào)用了confirm方法，說明消息發(fā)送到了exchange，且沒有調(diào)用return方法，說明消息成功到達相應(yīng)隊列。
getDirectMessageCopy方法成功消費掉“張三”這條消息，由于getDirectMessage方法模擬異常，所以第一次把“李四”從新入隊，此時getDirectMessageCopy繼續(xù)消費“王五”成功，getDirectMessage方法因李四已經(jīng)從新入隊過，再次發(fā)生異常則拋棄消息。
進一步挖掘你會發(fā)現(xiàn)，開始一共3條消息，有一條回滾消息總數(shù)變成了4條，每個消費者消費2條，所以兩個消費者是輪詢分配的。
- 工作隊列有兩種工作方式：輪詢分發(fā)(默認)、公平分發(fā)即當(dāng)某個消費者沒有消費完成之前不用再分發(fā)消息。
- 修改配置文件
# 消費者每次從隊列獲取的消息數(shù)量。此屬性當(dāng)不設(shè)置時為：輪詢分發(fā)，設(shè)置為1為：公平分發(fā)
spring.rabbitmq.listener.simple.prefetch=1
將第一個消費者模擬執(zhí)行5秒，然后向數(shù)據(jù)庫增加一條數(shù)據(jù)，執(zhí)行結(jié)果為：
可以看到，getDirectMessageCopy執(zhí)行了4次，getDirectMessage執(zhí)行了1次，根據(jù)他們的消費能力來公平分發(fā)消息。
如果可以，我想回到10年前...
