当前位置：首页 > news >正文

RabbitMQ多角度可靠性分析/基于Java代码深度解析

news 2025/6/20 5:31:48

上一篇我介绍了RabbitMQ的基本交换机以及队列，不了解可以再回顾一下。

那么这就带来一个问题，我们在程序中添加一个中间件，把我们的消息依托给MQ，那么我们如何保证我们的消息在这个过程不会丢失。保证它的的可靠性，目前主流从三个角度保证：生产者可靠性，MQ可靠性，消费者可靠性。

生产者可靠性：

所谓生产者可靠性就是我们如何保证将我们的消息从生产者将消息发送到队列或者消息的可靠性

1. 重试机制

针对网络异常，连接中断，卡顿的自动重发。(萌芽阶段，与客户端的连接，还没有到交换机/d队列)

spring:rabbitmq:/....connection-timeout: 1s template:retry:initial-interval: 1000msmultiplier: 2max-attempts: 3

参数讲解

connetion-timeout	连接时长，如果在这个时长内依旧没有连接成功就认为连接失败
initial-interval	等待时长，第一次连接失败后多久重新重试
mutiplier	等待倍数，结合initial-interval，计算下次等待时长,initial-interval=initial-interval*mutiplier
max-attempts	最大重连次数

重连机制似乎看起来我们每个程序都需要配置一下，但实际上我们真是企业还是要慎重，因为我们的不断重连是阻塞式的。所以说在追求极度的高并发情况下，需要慎重使用，如果必须使用，我们应该优化参数值，比如连接次数，连接时长等参数值调整。

2. 回调机制

确认消息是否到投递到交换机以及是都正确路由(成长阶段，到达交换机/队列)

Confirm Callback策略

确认消息是否成功到达交换机

spring:rabbitmq:/...publisher-confirm-type: correlate/none/simple

参数解释

publisher-confirm-type		Spring AMQP 启用监听RabbitMQ 退回消息的机制
	none	关闭confirm机制
	simple	同步阻塞等待MQ回执消息
	correlated	MQ异步回调方式返回回执消息

//创建关联数据
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
// 使用Lambda设置确认回调
rabbitTemplate.setMandatory(true);
rabbitTemplate.setConfirmCallback((correlationData1, ack, cause) -> {
if (ack) {System.out.println("[消息确认] 成功 - ID: " + correlationData.getId());
} else {System.out.println("[消息确认] 失败 - ID: " + correlationData.getId() + ", 原因: " + cause);
}
});

Return Callback策略

确认消息是否被成功路由（异步等待回执消息）

spring:rabbitmq:/...publisher-returns: true

rabbitTemplate.setMandatory(true);
rabbitTemplate.setReturnsCallback(returnedMessage -> {//对于需要的接口泛型 接受一个参数 在{}体内写方法体System.err.println("消息投递失败");//...处理逻辑/});

MQ可靠性：

交换机持久化

队列持久化

消息持久化

在spring-amqp中消息默认持久化

发送非持久化消息

Message message = MessageBuilder.withBody("hello".getBytes(StandardCharsets.UTF_8)).setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT) //PERSISTENT.build();

消息的持久化影响了我的MQ的吞吐量，同时保证了消息的可靠性。为了平衡可靠性性和性能，

RabbitMQ从 3.6版本 引入了惰性队列(lazy queue)。

让我们看一下为什么持久化是怎么影响性能的

如果我们的消息是持久化的，消息依旧会先存放在内存中，当我们的信息量达到一定程度后，服务器page out将消息扇出到磁盘中。这个时刻队列暂时无法接受消息，处于阻塞状态。