【Rabbitmq】Rabbitmq高级特性-发送者可靠性
Rabbitmq发送者可靠性
- 发送者重连
- 发送者确认
- 1.开启确认机制
- 2.ReturnCallback
- 3.ConfirmCallback
- MQ的可靠性
- 数据持久化
- 交换机持久化
- 队列持久化
- 消息持久化
- Lazy Queue
- 总结
- 其他文章
Rabbitmq提供了两种发送来保证发送者的可靠性，第一种叫发送者重连，第二种叫发送者确认。
发送者重连
有时候由于网络波动，可能会出现发送者连接MQ失败的情况，通过配置可以开启连接失败后的重连机制：
spring:
rabbitmq:
connection-timeout: 1s #设置MQ的连接超时时间
template:
retry:
enabled: true #开启超时重试机制
initial-interval: 1000ms # 失败后的初次等待时间
multiplier: 1 #失败后下次等待时长倍数，下次等待时长 = initial-interval * multiplier
max-attempts: 3 #最大重试次数
注：
当网络不稳定的时候，利用重试机制可以有效提高消息发送的成功率。不过SpringAMQP提供的重试机制是阻塞式的重试，也就是说多次重试等待的过程中，当前线程是被阻塞的，会影响业务性能。
如果对于业务性能有要求，建议禁用重试机制。如果一定要使用，请合理配置等待时长和重试次数，当然也可以考虑使用异步线程来执行发送消息的代码
发送者确认
SpringAMQP提供了Publisher Confirm和Publisher Return两种确认机制。开启确机制认后，当发送者发送消息给MQ后，MQ会返回确认结果给发送者。返回的结果有以下几种情况：
- 消息投递到了MQ，但是路由失败。此时会通过PublisherReturn返回路由异常原因，然后返回ACK，告知投递成功
- 临时消息投递到了MQ，并且入队成功，返回ACK，告知投递成功
- 持久消息投递到了MQ，并且入队完成持久化，返回ACK ，告知投递成功
- 其它情况都会返回NACK，告知投递失败
1.开启确认机制
在publisher这个微服务的application.yml中添加配置：
spring:
rabbitmq:
publisher-confirm-type: correlated # 开启publisher confirm机制，并设置confirm类型
publisher-returns: true # 开启publisher return机制
注：
这里publisher-confirm-type有三种模式可选：
- none：关闭confirm机制
- simple：同步阻塞等待MQ的回执消息
- correlated：MQ异步回调方式返回回执消息
2.ReturnCallback
每个RabbitTemplate只能配置一个ReturnCallback，因此需要在项目启动过程中配置：
@Slf4j
@Configuration
@RequiredArgsConstructor
public class MqConfig {
private final RabbitTemplate rabbitTemplate;
@PostConstruct
public void init(){
rabbitTemplate.setReturnsCallback(returned -> {
//业务处理
log.error("触发return callback,");
log.debug("exchange: {}", returned.getExchange());
log.debug("routingKey: {}", returned.getRoutingKey());
log.debug("message: {}", returned.getMessage());
log.debug("replyCode: {}", returned.getReplyCode());
log.debug("replyText: {}", returned.getReplyText());
});
}
}
3.ConfirmCallback
发送消息，指定消息ID、消息ConfirmCallback
@Test
public void SendConfirmCallBack() throws InterruptedException {
// 1.创建CorrelationData
CorrelationData cd = new CorrelationData(UUID.randomUUID().toString());
// 2.给Future添加ConfirmCallback
cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
@Override
public void onFailure(Throwable ex) {
// 2.1.Future发生异常时的处理逻辑，基本不会触发
log.error("handle message ack fail", ex);
}
@Override
public void onSuccess(CorrelationData.Confirm result) {
// 2.2.Future接收到回执的处理逻辑，参数中的result就是回执内容
if (result.isAck()){// result.isAck()，boolean类型，true代表ack回执，false 代表 nack回执
log.error("发送消息成功，收到 ack!");
}else {
// result.getReason()，String类型，返回nack时的异常描述
log.error("发送消息失败，收到 nack, reason : {}", result.getReason());
}
}
});
String exchangeName = "amq.topic";
String message = "Hello topic china";
this.rabbitTemplate.convertAndSend(exchangeName,".11news11", message,cd);
Thread.sleep(3000); // 测试方法用于回显日志
}
注：
发送者确认机制需要跟MQ进行通讯和确认会影响发送的效率，开启根据实际需求考虑，开以后要注意重试次数不要无限重试，注意重试次数，否则对性能影像严重。
MQ的可靠性
在默认情况下，RabbitMQ会将接收到的信息保存在内存中以降低消息收发的延迟。这样会导致两个问题：
- 一旦MQ宕机，内存中的消息会丢失
- 内存空间有限，当消费者故障或处理过慢时，会导致消息积压，引发MQ阻塞
数据持久化
RabbitMQ实现数据持久化包括3个方面：
交换机持久化
交换机默认持久化
Durable 持久
Transient 临时
队列持久化
队列默认持久化
Durable 持久
Transient 临时
消息持久化
消息持久化需要手动指定
Non-persistent 非持久化
Persistent 持久化
代码实现
MessageBuilder.withBody("案例".getBytes(StandardCharsets.UTF_8)).setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();
注:
由于Mq默认持久化,所以一般不需要修改
Lazy Queue
从RabbitMQ的3.6.0版本开始，就增加了Lazy Queue的概念，也就是惰性队列。
惰性队列的特征如下：
- 接收到消息后直接存入磁盘，不再存储到内存
- 消费者要消费消息时才会从磁盘中读取并加载到内存（可以提前缓存部分消息到内存，最多2048条）
在3.12版本后，所有队列都是Lazy Queue模式，无法更改。
要设置一个队列为惰性队列，只需要在声明队列时，指定x-queue-mode属性为lazy即可：
代码实现：
注解式
@RabbitListener(queuesToDeclare = @Queue(
name = "lazy.queue",
durable = "true",
arguments = @Argument(name = "x-queue-mode", value = "lazy")
))
public void listenLazyQueue(String msg){
log.info("接收到 lazy.queue的消息：{}", msg);
}
声明式
@Bean
public Queue lazyQueue(){
return QueueBuilder
.durable("lazy.queue")
.lazy() // 开启Lazy模式
.build();
}
总结
- 首先通过配置可以让交换机、队列、以及发送的消息都持久化。这样队列中的消息会持久化到磁盘，MQ重启消息依然存在。
- RabbitMQ在3.6版本引入了LazyQueue，并且在3.12版本后会称为队列的默认模式。LazyQueue会将所有消息都持久化
- 开启持久化和发送者确认时， RabbitMQ只有在消息持久化完成后才会给发送者返回ACK回执
其他文章
发送者可靠性
消费者可靠性
延迟信息