使用MQ的时候,怎么确保消息100%不丢失?

1.案例背景

我们以京东系统为例,用户在购买商品时,通常会选择用京豆抵扣一部分的金额,在这个过程中,交易服务和京豆服务通过 MQ 消息队列进行通信。
在下单时,交易服务发送“扣减账户 X 100 个京豆”的消息给 MQ 消息队列,而京豆服务则在消费端消费这条命令,实现真正的扣减操作。
使用MQ的时候,怎么确保消息100%不丢失?_第1张图片

在实现以上操作的过程中,我们会遇到哪些问题呢?

2.案例分析

首先我们需要先知道我们引入 MQ 消息中间件最直接的目的是:做系统解耦合流量控制,追其根源还是为了解决互联网系统的高可用和高性能问题。

  • 系统解耦:用 MQ 消息队列,可以隔离系统上下游环境变化带来的不稳定因素,比如京豆服务的系统需求无论如何变化,交易服务不用做任何改变,即使当京豆服务出现故障,主交易流程也可以将京豆服务降级,实现交易服务和京豆服务的解耦,做到了系统的高可用。
  • 流量控制:遇到秒杀等流量突增的场景,通过 MQ 还可以实现流量的“削峰填谷”的作用,可以根据下游的处理能力自动调节流量。

不过引入 MQ 虽然实现了系统解耦合流量控制,也会带来其他问题。
引入 MQ 消息中间件实现系统解耦,会影响系统之间数据传输的一致性。 在分布式系统中,如果两个节点之间存在数据同步,就会带来数据一致性的问题。同理,在这一讲你要解决的就是:消息生产端和消息消费端的消息数据一致性问题(也就是如何确保消息不丢失)

引入 MQ 消息中间件解决流量控制, 会使消费端处理能力不足从而导致消息积压,这也是我们需要要解决的问题。

通过上面的分析,其实我们不难发现问题与问题之间是环环相扣的,那面对“在使用 MQ 消息队列时,如何确保消息不丢失”这个问题时,我们应该怎么去解决,通过什么方式去解决呢?

首先,我们可以做一个简单的分析:

  • 如何知道有消息丢失?
  • 哪些环节可能丢消息?
  • 如何确保消息不丢失?

网络中的数据传输不可靠,想要解决如何不丢消息的问题,首先要知道哪些环节可能丢消息,以及我们如何知道消息是否丢失了,最后才是解决方案(而不是上来就直接说自己的解决方案)。就好比“架构设计”“架构”体现了架构师的思考过程,而“设计”才是最后的解决方案,两者缺一不可。

3.解决过程

我们首先来看消息丢失的环节,一条消息从生产到消费完成这个过程,可以划分三个阶段,分别为消息生产阶段,消息存储阶段和消息消费阶段。
使用MQ的时候,怎么确保消息100%不丢失?_第2张图片

  • 消息生产阶段: 从消息被生产出来,然后提交给 MQ 的过程中,只要能正常收到 MQ Broker 的 ack 确认响应,就表示发送成功,所以只要处理好返回值和异常,这个阶段是不会出现消息丢失的。
  • 消息存储阶段: 这个阶段一般会直接交给 MQ 消息中间件来保证,但是你要了解它的原理,比如 Broker 会做副本,保证一条消息至少同步两个节点再返回 ack。
  • 消息消费阶段: 消费端从 Broker 上拉取消息,只要消费端在收到消息后,不立即发送消费确认给 Broker,而是等到执行完业务逻辑后,再发送消费确认,也能保证消息的不丢失。

方案看似万无一失,每个阶段都能保证消息的不丢失,但在分布式系统中,故障不可避免,作为消息生产端,你并不能保证 MQ 是不是弄丢了你的消息,消费者是否消费了你的消息,所以,本着 Design for Failure 的设计原则,你还是需要一种机制,来 Check 消息是否丢失了。
所以我们下面要思考的就是:怎么进行消息检测,总体的解决方案思路也很简单:在消息生产端,给每个发出的消息都指定一个全局唯一 ID,或者附加一个连续递增的版本号,然后在消费端做对应的版本校验。

那么我们具体怎么取实现呢?拦截机制,在生产端发送消息之前,通过拦截器将消息版本号注入消息中(版本号可以采用连续递增的 ID 生成,也可以通过分布式全局唯一 ID生成)。然后在消费端收到消息后,再通过拦截器检测版本号的连续性或消费状态,这样实现的好处是消息检测的代码不会侵入到业务代码中,可以通过单独的任务来定位丢失的消息,做进一步的排查。

这里需要我们注意的是:如果同时存在多个消息生产端和消息消费端,通过版本号递增的方式就很难实现了,因为不能保证版本号的唯一性,此时只能通过**全局唯一 ID **的方案来进行消息检测,具体的实现原理和版本号递增的方式一致。

解决了消息不丢失,我们又迎来了另一个挑战:怎么解决消息被重复消费的问题?
在消息消费的过程中,如果出现失败的情况,通过补偿的机制发送方会执行重试,重试的过程就有可能产生重复的消息,那么如何解决这个问题?
其实这个问题我们也可以理解为:如何解决消费端幂等性问题(幂等性,就是一条命令,任意多次执行所产生的影响均与一次执行的影响相同),只要消费端具备了幂等性,那么重复消费消息的问题也就解决了。
具体如何解决MQ重复消费,以及如何保证幂等性操作,由于内容较多,我单独拎出来讲解了一下,感兴趣的可以移步:

  • 《1.MQ幂等、去重的解决方案》
  • 《2.rabbitmq实现幂等性操作——实战》

4.总结

  • 1.如何确保消息不会丢失?

你要知道一条消息从发送到消费的每个阶段,是否存在丢消息,以及如何监控消息是否丢失,最后才是如何解决问题,方案可以基于“ MQ 的可靠消息投递 ”的方式。

  • 2.如何保证消息不被重复消费?

在进行消息补偿的时候,一定会存在重复消息的情况,所以我们如何控制重新消费保证幂等性。

  • 3.如何处理消息积压问题?

其本质就是如何通过 MQ 实现真正的高性能。思路是,本着解决线上异常为最高优先级,然后通过监控和日志进行排查并优化业务逻辑,最后是扩容消费端和分片的数量。

你可能感兴趣的