RabbitMQ确保消息可靠性
|总字数:4.2k|阅读时长:14分钟|浏览量:|
消息丢失的可能性
支付服务先扣减余额和更新支付状态（这俩是同步调用），然后通过RabbitMq异步调用支付服务更新订单状态。但是有些情况下，可能订单已经支付 ，但是更新订单状态却失败了，这就出现了消息丢失。
发送者在发送的过程中出现了网络故障RabbitMQ在发送消息的过程中出现了问题消费者在更新订单状态的时候出现了问题
发送者的可靠性
发送者确认机制需要与MQ进行通信和确认，会影响消息发送的效率且一般出现的概率极低，所以一般不用这个。
方法1. 发送者重连
确保发送者与MQ之间连接的可靠性。有的时候由于网络波动，可能出现发送者连接MQ失败的情况，这个配置是关闭的，可以开启连接失败后的重连机制：
spring:
rabbitmq:
connection-timeout: 1s
template:
retry:
enabled: true
initial-interval: 1000ms
multiplier: 1
max-attempts: 3
|
【注】：当网络不稳定时，利用重试机制可以提高消息发送的成功率，但是SpringAMQP提供的重试机制是阻塞式的重试，如果需要多次重试等待，当前线程被阻塞，会影响性能。
如果对业务性能有要求，建议禁用重试机制，如果一定要使用，要合理的配置等待时常和重试次数，或使用异步线程来执行发送消息的代码。
方法2. 发送者确认
确保消息发送的可靠性。SpringAMQP提供了Publisher Confirm和Publisher Return两种机制，开启确认机制后，当发送者发送消息给MQ后，MQ会返回确认结果给发送者，返回的结果有以下几种情况：
- 消息投递到MQ，但是路由失败，此时通过PublisherReturn返回路由异常信息，然后返回
ACK，告知投递成功。例如：
- 临时消息【不需要往磁盘做持久化的消息】投递到MQ，并入队成功，返回
ACK，告知投递成功。
- 持久消息投递到MQ，并入队完成持久化，返回
ACK，告知投递成功。
- 其他情况都会返回
NACK，告知投递失败。
步骤：
- 在发送方publisher所在的微服务的application.yml中配置：
spring:
rabbitmq:
publisher-confirm-type: correlated
publisher-returns: true
|
publisher-confirm-type有三种模式：
- none：关闭confirm机制
- simple：同步阻塞等待MQ回执消息
- correlated：MQ异步回调方式返回回执消息（常用）
- 开启回调机制：每个RabbitTemplate只能配置一个ReturnCallback，在发送者publisher所在的项目启动时配置即可。
@Configuration
@RequiredArgsConstructor
@Slf4j
public class MqConfig {
private final RabbitTemplate rabbitTemplate;
@PostConstruct
public void init() {
rabbitTemplate.setReturnsCallback(returnedMessage -> {
log.error("监听到了消息return callback");
log.debug("exchange: {}", returnedMessage.getExchange());
log.debug("routingKey: {}", returnedMessage.getRoutingKey());
log.debug("message: {}", returnedMessage.getMessage());
log.debug("replyCode: {}", returnedMessage.getReplyCode());
log.debug("replyText: {}", returnedMessage.getReplyText());
});
}
}
|
- 开启消息确认机制：发送消息、指定消息ID、每次发送消息都需要配置一个ConfirmCallback
public void testConfirmCallback() {
CorrelationData cd = new CorrelationData(UUID.randomUUID().toString());
cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
@Override
public void onFailure(Throwable ex) {
log.error("spring ampq处理确认结果异常", ex);
}
@Override
public void onSuccess(CorrelationData.Confirm result) {
if(result.isAck()) {
log.debug("收到ACK，消息发送成功");
}else {
log.debug("收到NACK，消息发送失败，失败原因：{}",result.getReason());
}
}
});
rabbitTemplate.convertAndSend("hmall.direct", "blue", "hello world", cd);
}
|
MQ的可靠性
RabbitMQ一般会将收到的信息保存到内存（速度快）中，降低消息收发的延迟，这样会导致：
- MQ宕机，内存中的消息会丢失。
- 内存空间有限，消费者故障或处理过慢，会导致消息积压，引发MQ阻塞。
【案例】发送者往MQ发消息，MQ会把数据保存到内存中，如果内存满了，MQ就会把一部分数据迁移到磁盘中暂时进行持久化存储，移动到磁盘的这段时间发送者发送的消息就会产生丢失。
方法1. 数据持久化
数据持久化就是把数据持久化到磁盘，但是不是向上边那个案例，等满了再去持久化（被动），而是提前进行持久化。
- 交换机的持久化（默认开启的）
- 队列的持久化（默认开启的）
- 消息持久化（默认是非持久的）
在发送消息的时候设定的
【案例】：比较一下持久化和非持久化的性能。
发100w条消息给MQ：
这是非持久化的方式：使用纯内存的方式存储，每次内存满之后，MQ就会把消息写到磁盘中，此时就会出现阻塞状态，处理速度降低到0
【问题】可能出现消息丢失和MQ阻塞
【解决办法】使用持久化的方式：
public void testSendPersistentMsg() {
Message msg = MessageBuilder.withBody("hello world".getBytes(StandardCharsets.UTF_8))
.setDeliveryMode(MessageDeliveryMode.PERSISTENT)
.build();
for (int i = 0; i < 100000; ++i) {
rabbitTemplate.convertAndSend("simple.queue", msg);
}
}
|
Mq并没有阻塞，每发一条消息就赶紧把它存到磁盘中，和纯内存方式相比，不会有个中断的过程。
方法2. Lazy Queue（推荐）
【问题】：由于使用了消息持久化的方式，发到MQ的消息不仅要到内存，还要在磁盘中写一份，这会导致整体的并发能力下降
【特征】：
- 接收到消息后直接入磁盘，不再存储到内存
- 在写磁盘的时候也对写入磁盘的操作进行一些优化，比传统的写操作高很多
- 消费者要消费消息时，才会从磁盘中读取并加载到内存
- 【
问题】：可能会影响消费者处理消息的速度
- 【
解决】：可以提前缓存部分消息到内存，最多2048条
控制台声明Lazy Queue队列
Java代码添加
声明Bean
@Bean
public Queue lazyQueue(){
return QueueBuilder
.durable("lazy.queue")
.lazy()
.build();
}
|
@RabbitListener注解
@RabbitListener(queuesToDeclare = @Queue(
name = "lazy.queue",
durable = "true",
arguments = @Argument(name = "x-queue-mode", value = "lazy") // 开启Lazy模式
))
public void listenLazyQueue(String msg){
log.info("接收到 lazy.queue的消息：{}", msg);
}
|
消费者的可靠性
消费者确认机制
为了确认消费者是否成功处理消息，当消费者处理消息结束后，应该向MQ发送一个回执，告知MQ自己的消息处理状态。有如下几种消息处理状态：
- ack：处理消息成功，RabbitMQ从队列中删除该消息
- nack：消息处理失败，RabbitMQ需要再次投递消息
- reject：消息处理失败并拒绝该消息，RabbitMQ从队列中删除该消息【在处理的过程中，发现消息的内容有问题，没有重试的必要，直接拒绝就行】
【注意】：不管是哪种情况，都应该等消息处理完后得到结果再返回，不要一拿到消息就返回
返回消息处理状态的过程，类似于处理事务，事务处理成功，返回ACK；处理失败，返回NACK
SpringAMQP允许通过在消费者的配置文件选择ACK的处理方式，有三种：
- none：不处理，消息投递给消费者后立刻ack，消息会立刻从MQ中删除，别用
- manual：手动模式，需要在业务代码中调用api，发送ack或reject，存在业务入侵，但是更灵活。
- auto：自动模式，利用AOP对消息处理逻辑进行了环绕增强
- 业务处理正常：自动返回
ack
- 业务处理异常：自动返回
nack
- 消息处理或校验异常【MessageConversionException】：自动返回
reject
spring:
rabbitmq:
listener:
simple:
acknowledge-mode: auto# 不做处理
|
失败重试策略
在消费者出现异常时，利用本地重试，而不是无限的重新入队到mq，可以在消费者的yaml文件中添加配置来开启重试机制。
spring:
rabbitmq:
listener:
simple:
retry:
enabled: true
initial-interval: 1000ms
multiplier: 1
max-attempts: 3
stateless: true
|
【问题】：在开启重试模式后，重试次数耗尽，如果消息仍然失败，默认会把消息进行丢弃。
【解决】：因此需要有MessageRecoverer接口来处理，包含三种不同的实现：
- RejectAndDontRequeueRecoverer（默认）：重试耗尽后，直接reject，丢弃消息。
- ImmediateRequeueMessageRecoverer：重试耗尽后，返回nack，消息重新入队。
- RepublishMessageRecoverer：重试耗尽后，将失败消息投递到指定的交换机 。
修改失败重试策略为RepublishMessageRecoverer
- 定义接收失败的交换机、队列、
- 定义RepublishMessageRecoverer
@Configuration
public class ErrorMessageConfiguration {
@Bean
public DirectExchange errorExchange() {
return new DirectExchange("error.direct");
}
@Bean
public Queue errorQueue() {
return new Queue("error.queue");
}
@Bean
public Binding errorQueueBinding() {
return BindingBuilder.bind(errorQueue())
.to(errorExchange())
.with("error");
}
@Bean
public MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate) {
return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
}
}
|
业务幂等性
f(x) = f(f(x))，指同一个业务，执行一次或多次对业务状态的影响是一致的。
- 幂等业务：查询业务、删除业务
- 非幂等业务：用户下单需要扣减库存、用户退款业务需要恢复余额
方案1. 唯一消息id
给每个消息设置一个唯一id，利用id区分是否是重复消息：
- 每条消息都生成一个唯一id，与消息一起投递给消费者
- 消费者接收到消息后处理自己的业务，业务处理成功后将消息id保存到数据库中
- 如果下次又收到相同消息，去数据库查询判断是否存在，存在则视为重复消息放弃处理
- 在发送方配置Bean用来自动创建消息id
@Configuration
public class MqConfig {
@Bean
public MessageConverter messageConverter() {
Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter();
converter.setCreateMessageIds(true);
return converter;
}
}
|
- 在接收方接收消息id
@RabbitListener(queues = "simple.queue")
public void listenSimpleQueue(Message msg) {
log.info("监听到simple.queue的消息：{}", msg);
log.info("消息id：{}", msg.getMessageProperties().getMessageId());
}
|
方案2. 业务判断（常用）
结合业务逻辑，基于业务本身做判断。
【案例】：当用户下单成功后，通过MQ通知交易服务来修改订单状态为已支付（这里记作消息1），修改成功后交易服务返回ACK给MQ，此时出现了网络的故障，MQ没有收到交易服务发送的ACK，MQ认为交易服务宕机，消息又重新入队。
就在此刻，用户点击了申请退款，直接向交易服务修改订单状态为退款中（这个操作没有走MQ，此时订单状态是退款中，但是消息1还在消息队列中）。
此时网络恢复了，MQ又将消息1发送给交易服务，此时交易服务又把订单状态标记为已支付（订单申请退款中的状态又被覆盖了）。
【解决】：通知来的时候，先判断订单的状态，再进行操作。
@Component
@RequiredArgsConstructor
public class PayStatusListener {
private final IOrderService orderService;
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "trade.pay.success.queue", durable = "true"),
exchange = @Exchange(name = "pay.direct"),
key = "pay.success"
))
public void listenPaySuccess(Long orderId) {
Order order = orderService.getById(orderId);
if(order == null || order.getStatus() != 1) {
return;
}
orderService.markOrderPaySuccess(orderId);
}
}
|
延迟消息
延迟消息：发送者发送消息时指定一个时间，消费者不会立刻收到消息，而是在指定时间之后才收到消息。
延迟任务：设置在一定时间后才执行的任务。
方案1. 死信交换机
当一个队列中的消息满足下列情况之一的，就会成为死信：
- 消费者使用basic.reject或basic.nack声明消费失败，并且消息的requeue参数设置为false。
- 消息是一个过期消息（达到队列设置的过期时间 或 消息本身设置的过期时间），超时无人消费。
- 要投递的队列消息堆积满了，最早的消息可能成为死信。
队列通过dead-letter-exchange属性指定了一个交换机，那么该队列中的死信就会投递到这个交换机中，这个交换机就叫做死信交换机（DLX）。
- 声明死信队列、死信交换机、它们之间的绑定关系：
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "dlx.queue", durable = "true"), // 死信队列
exchange = @Exchange(name = "dlx.direct", type = ExchangeTypes.DIRECT), // 死信交换机
key = {"hi"}
))
public void listenDlxQueue(String msg) {
log.info("消费者监听到dlx.queue的消息: " + msg);
}
|
- 声明普通队列、普通交换机、它们之间的绑定关系，并把队列绑定到死信交换机上（此时就不需要把它绑定消费者了）：
@Configuration
public class NormalConfiguration {
@Bean
public DirectExchange normalExchange() {
return ExchangeBuilder.directExchange("normal.direct").build();
}
@Bean
public Queue normalQueue() {
return QueueBuilder
.durable("normal.direct")
.deadLetterExchange("dlx.direct")
.build();
}
@Bean
public Binding normalQueueBinding(Queue normalQueue, DirectExchange normalExchange) {
return BindingBuilder
.bind(normalQueue)
.to(normalExchange)
.with("hi");
}
}
|
- 发送延迟消息：
@Test
public void testSendDelayMsg() {
rabbitTemplate.convertAndSend("normal.direct", "hi", "hello world", message -> {
message.getMessageProperties().setExpiration("10000");
return message;
});
}
|
【注】：normal.direct和normal.queue之间绑定的BindingKey 与 dlx.direct和dlx.queue之间绑定的BindingKey要一致
方案2. 延迟消息插件DelayExchange（推荐）
这个插件可以将普通交换机改造为支持延迟消息功能的交换机，当消息投递到交换机后，可以暂存一段时间，到后期再投递到队列。
一、安装插件
- 插件下载地址：DelayExchange
- 需要把插件放在RabbitMQ插件目录对应的数据卷下
docker volume inspect mq-plugins
|
- 执行命令，安装插件
docker exec -it rabbitmq rabbitmq-plugins enable rabbitmq_delayed_message_exchange
|
二、使用插件
- 声明延迟交换机：只要设置delay的属性为true即可
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "delay.queue", durable = "true"),
exchange = @Exchange(name = "delay.direct", delayed = "true", type = ExchangeTypes.DIRECT), // 只要设置一个delayed属性为true即可
key = {"hi"}
))
public void listenDelayQueue(String msg) {
log.info("消费者监听到delay.queue的消息: " + msg);
}
|
- 发送延迟消息：通过消息头x-delay来设置过期时间
@Test
public void testSendDelayMsgByPlugin() {
rabbitTemplate.convertAndSend("delay.direct", "delay", "hello world", message -> {
message.getMessageProperties().setDelay(10000);
return message;
});
}
|
延迟消息的实现需要记录消息的过期时间，计时的时钟需要依赖cpu，是个cpu密集型任务。因此使用延迟消息时，需要避免同一时刻在mq里存在大量的延迟消息（尽可能地让延迟消息的延迟时间不要太长）。
