MQ的可靠性
MQ的可靠性
在分布式系统中，消息队列（MQ）的可靠性是确保系统稳定运行的关键。本文将深入探讨MQ的可靠性问题，并提供具体的解决方案，包括数据持久化、消费者确认机制、消息失败处理策略以及业务幂等性等关键概念。通过本文，读者将能够掌握如何构建一个可靠的消息队列系统，以应对各种复杂的业务场景。
MQ的可靠性
在默认情况下，RabbitMQ会将接收到的信息保存在内存中以降低消息收发的延迟。这样会导致两个问题：
- 一旦MQ宕机，内存中的消息会丢失。
- 内存空间有限，当消费者故障或者处理过慢时，会导致消息积压，引发MQ阻塞。
解决方案
- 数据持久化
在早期MQ3.6之前,会采用这种模式,RabbitMQ实现数据持久化包括3个方面：
- 交换机的持久化
Transient属于临时的交换机，重启后不存在。默认是持久化创建
- 队列的持久化
和交换机的持久化类似
- 消息的持久化
发送消息时，delivery_mode设置为2表示持久化消息。
- Lazy Queue
要设置一个队列为惰性队列，只需要在声明队列时，指定x-queue-mode属性为lazy即可：
public Queue lazyQueue(){
return QueueBuilder.durable("lazy.queue")
.lazy() //开启lazy模式
.build();
}
@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);
}
消费者确认机制
SpringAMQP已经实现了消息确认功能。并允许我们通过配置文件选择ACK处理方式，有三种方式：
- none: 不处理。即消息投递给消费者后立刻ack,消息会立刻从MQ删除。非常不安全，不建议使用
- manual: 手动模式。需要自己在业务代码中调用api,发送ack或rehect,存在业务入侵，但更灵活
- auto: 自动模式。SpringAMQP利用AOP对我们的消息处理逻辑做了环绕增强，当业务正常执行时则自动返回ack.
当业务出现异常时，根据异常判断返回不同结果 - 如果是业务异常，会自动返回nack
- 如果是消息处理或校验异常，会自动返回reject
spring:
rabbitmq:
listener:
simple:
prefetch: 1
acknowledge-mode: none
消息失败处理
当消费者出现异常后，消息会不断requeue(重新入队)到队列，再重新发送给消费者，然后再次异常，再次requeue,无限循环，导致mq的消息处理飙升，带来不必要的压力。我们可以利用spring的retry机制，在消费者出现异常时利用本地重试，而不是无限制的requeue到mq队列：
spring:
rabbitmq:
listener:
simple:
prefetch: 1
retry:
enable: true #开启消费者失败重试
initial-interval: 1000ms #初始的失败等待时长为 1秒
multiplier: 1 #下次失败的等待时长倍数，下次等待时长=multiplier * last-interval
max-attempts： 3 #最大重试次数
stateless: true # true无状态；false有状态，如果业务中包含事务，这里改为false
失败消息处理策略
在开启重试模式后，重试次数耗尽，如果消息依然失败，则需要有MessageRecoverer接口来处理，它包含三种不同的实现：
- RejectAndDontRequeueRecoverer: 重试耗尽后，直接reject,丢弃消息。默认就是这种方式
- ImmediateRequeueMessageRecoverer:重试耗尽后，返回nack,消息重新入队
- RepublishMessageRecoverer: 重试耗尽后，将失败消息投递到指定的交换机。
将失败处理策略改为RepublishMessageRecoverer:
①首先，定义接受失败消息的交换机，队列及其绑定关系，此处略；
②然后，定义ReplishMessageRecoverer:
@Configuration
@ConditionalOnProperty(name = "spring.rabbitmq.listener.simple.retry.enabled",havingValue = "true")
public class ErrorConfiguration {
@Bean
public DirectExchange errorExchange(){
return new DirectExchange("error.direct");
}
@Bean
public Queue errorQueue(){
return new Queue("error.queue");
}
@Bean
public Binding directQueueBindingerror(Queue errorQueue, DirectExchange errorExchange){
return BindingBuilder.bind(errorQueue).to(errorExchange).with("error");
}
@Bean
public MessageRecoverer recoverer(RabbitTemplate rabbitTemplate){
return new RepublishMessageRecoverer(rabbitTemplate,"error.direct","error");
}
通过配置RepublishMessageRecoverer来设置本地重试后发往收集异常信息的交换机。同时可以通过@ConditionalOnProperty设置是否开启
总结
消费者如何保证消息一定被消费？
- 开启消费者确认机制为auto,由spring确认消息处理成功后返回ack,异常时返回nack
- 开启消费者失败重试机制，并设置MessageRecoverer,多次重试失败后将消息投递到异常交换机，交由人工处理。
业务幂等性
幂等是一个数学概念，用函数表达式来描述是这样的：f(x)=f(f(x))。在程序开发中，则是指同一个业务，执行一次或多次对业务状态的影响是一致的。
解决方案：
唯一消息id
方案一，是给每个消息都设置一个唯一id，利用id区分是否是重复消息：
① 每一条消息都生成一个唯一的id，与消息一起投递给消费者。
② 消费者接收到消息后处理自己的业务，业务处理成功后将消息ID保存到数据库。
③ 如果下次又收到相同的消息，去数据库查询判断是否存在，存在则为重复消息放弃处理。
@Bean
public MessageConverter messageConverter(){
Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
//配置自动创建消息id,用于识别不同消息，也可以在业务中基于ID判断是否是重复消息
jackson2JsonMessageConverter.setCreateMessageIds(true);
return jackson2JsonMessageConverter;
}
配置消息转换器时，可以设置自动写入唯一id
方案二，是结合业务逻辑，基于业务本身做判断。以我们的业务为例：我们要在支付后修改订单状态为已支付，应该在修改订单状态前先查询订单状态，判断状态是否是未支付。只有未支付订单才需要修改，其他状态不做处理。
问题：
1.如何保证支付服务和交易服务之间的订单状态一致性
- 首先，支付服务会在用户支付成功后利用MQ消息通知交易服务，完成订单状态同步。
- 其次，为了保证MQ消息的可靠性，我们采用了生产者确认机制，消费者确认，消费者失败重试等策略，确保消息投递和处理的可靠性。同时也开启了MQ的持久化，避免因为服务宕机导致的消息丢失。
- 最后，我们还在交易服务更新订单状态时做了业务幂等性判断，避免因为消息重复消费导致订单状态异常。
2.如果交易服务消息处理失败，有没有什么兜底方案？
- 我们可以在交易服务设置定时任务，定期查询订单支付状态。这样即使MQ通知失败，还可以利用定时任务作为兜底方案，确保订单支付状态的最终一致性。