对于发送方发送消息到RabbitMQ的可靠性机制
引入：在持久化的消息正确存⼊RabbitMQ之后,还需要有⼀段时间(虽然很短,但是不可忽视)才能存⼊磁盘中.RabbitMQ并不会为每条消息都进⾏同步存盘(调⽤内核的fsync⽅法)的处理, 可能仅仅保存到操作系统缓存之中⽽不是物理磁盘之中. 如果在这段时间内RabbitMQ服务节点发⽣了宕机、重启等异常情况, 消息保存还没来得及落盘, 那么这些消息将会丢失
RabbitMQ为我们提供了两种解决⽅案来实现持久化:
a. 通过事务机制实现
b. 通过发送⽅确认(publisher confirm) 机制实现
事务比较消耗信能，因此通常使用发送⽅确认(publisher confirm) 机制实现，RabbitMQ为我们提供了两个⽅式来控制消息的可靠性投递
1. confirm确认模式
2. return退回模式
注：这两个模式是不互斥的，可以同时存在
1、producer->Broker (发送方确认）
1）producer->exchange : confirm 模式
2）exchange->queue : return 模式
3）队列溢出：死信等
2、Broker （持久化（交换机、队列、消息））
3、Broker->Consumer
自动确认：消息送达到consumer时，消息会自动删除
手动确认
1.confirm确认模式
Producer 在发送消息的时候, 对发送端设置⼀个ConfirmCallback的监听, ⽆论消息是否到达
Exchange, 这个监听都会被执⾏, 如果Exchange成功收到, ACK( Acknowledge character , 确认
字符)为true, 如果没收到消息, ACK就为false.
配置RabbitMQ
spring:
rabbitmq:
addresses: amqp://study:study@110.41.51.65:15673/bite
listener:
simple:
acknowledge-mode: manual #消息接收确认
publisher-confirm-type: correlated #消息发送确认
设置确认回调逻辑并发送消息
⽆论消息确认成功还是失败, 都会调⽤ConfirmCallback的confirm⽅法. 如果消息成功发送到Broker,ack为true.
如果消息发送失败, ack为false, 并且cause提供失败的原因.
@Bean("confirmRabbitTemplate")
public RabbitTemplate confirmRabbitTemplate(ConnectionFactory
connectionFactory){
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
@Override
public void confirm(CorrelationData correlationData, boolean ack,
String cause) {
System.out.printf("");
if (ack){
System.out.printf("消息接收成功, id:%s \n",
correlationData.getId());
}else {
System.out.printf("消息接收失败, id:%s, cause: %s",
correlationData.getId(), cause);
}
}
});
return rabbitTemplate;
}RabbitTemplate.ConfirmCallback 和 ConfirmListener 区别
在RabbitMQ中, ConfirmListener和ConfirmCallback都是⽤来处理消息确认的机制, 但它们属于不同
的客⼾端库, 并且使⽤的场景和⽅式有所不同.
1. ConfirmListener 是 RabbitMQ Java Client 库中的接⼝. 这个库是 RabbitMQ 官⽅提供的⼀个直
接与RabbitMQ服务器交互的客⼾端库. ConfirmListener 接⼝提供了两个⽅法: handleAck 和
handleNack, ⽤于处理消息确认和否定确认的事件.
2. ConfirmCallback 是 Spring AMQP 框架中的⼀个接⼝. 专⻔为Spring环境设计. ⽤于简化与
RabbitMQ交互的过程. 它只包含⼀个 confirm ⽅法，⽤于处理消息确认的回调
如果当前ConfirmCallback写在生产者上面，会出现问题: 第二次重启服务器：会报错 Only one ConfirmCallback is supported by each RabbitTemplate] with root cause
如果只重写RabbitTemplate中的ConfirmCallback ，会出现其他的生产者、消费者会受到影响，因此服务器中自己生成的RabbitTemplate也要重写
2、return退回模式
消息到达Exchange之后, 会根据路由规则匹配, 把消息放⼊Queue中. Exchange到Queue的过程, 如果⼀条消息⽆法被任何队列消费(即没有队列与消息的路由键匹配或队列不存在等), 可以选择把消息退回给发送者. 消息退回给发送者时, 我们可以设置⼀个返回回调⽅法, 对消息进⾏处理.
配置RabbitMQ
spring:
rabbitmq:
addresses: amqp://study:study@110.41.51.65:15673/bite
listener:
simple:
acknowledge-mode: manual #消息接收确认
publisher-confirm-type: correlated #消息发送确认
设置返回回调逻辑并发送消息
消息⽆法被路由到任何队列, 它将返回给发送者，这时setReturnCallback设置的回调将被触发
如果把setReturnCallback写在生产者里面就会出现和 ConfirmCallback一样的报错情况，因此也要重写到RabbitTemplate中
rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
@Override
public void returnedMessage(ReturnedMessage returned) {
System.out.printf("消息被退回: %s", returned);
});使⽤RabbitTemplate的setMandatory⽅法设置消息的mandatory属性为true(默认为false). 这个属性的作⽤是告诉RabbitMQ, 如果⼀条消息⽆法被任何队列消费, RabbitMQ应该将消息返回给发送者, 此时 ReturnCallback 就会被触发.
