写点什么

消息队列的丢失、重复与积压问题

作者:Xiao8
  • 2022 年 6 月 23 日
  • 本文字数:1825 字

    阅读完需:约 6 分钟

作者:Damon

博客:http://www.damon8.cn

交个朋友之猿天地 | 微服务 | 容器化 | 自动化


我们常常在用到 MQ 时,经常基本都会抛出一个问题:在使用 MQ 的时候,怎么确保消息 100% 不丢失?


这个问题在实际工作中很常见,既能考察候选者对于 MQ 中间件技术的掌握程度,又能很好地区分候选人的能力水平。接下来,我们就从这个问题出发,探讨你应该掌握的基础知识。

案例背景

以京东系统为例,用户在购买商品时,通常会选择用京豆抵扣一部分的金额,在这个过程中,交易服务和京豆服务通过 MQ 消息队列进行通信。在下单时,交易服务发送“扣减账户 X 100 个京豆”的消息给 MQ 消息队列,而京豆服务则在消费端消费这条命令,实现真正的扣减操作。



那在这个过程中你会遇到什么问题呢?

案例分析

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


  • 系统解耦:用 MQ 消息队列,可以隔离系统上下游环境变化带来的不稳定因素,比如京豆服务的系统需求无论如何变化,交易服务不用做任何改变,即使当京豆服务出现故障,主交易流程也可以将京豆服务降级,实现交易服务和京豆服务的解耦,做到了系统的高可用。

  • 流量控制:遇到秒杀等流量突增的场景,通过 MQ 还可以实现流量的“削峰填谷”的作用,可以根据下游的处理能力自动调节流量。


不过引入 MQ 虽然实现了系统解耦合流量控制,也会带来其他问题。


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


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


所以你能发现,问题与问题之间往往是环环相扣的,面试官会借机考察你解决问题思路的连贯性和知识体系的掌握程度。


那面对“在使用 MQ 消息队列时,如何确保消息不丢失”这个问题时,你要怎么解决呢?首先,你要分析其中有几个考点,比如:


  • 如何知道有消息丢失

  • 哪些环节可能丢消息

  • 如何确保消息不丢失


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

案例解答

我们首先来看消息丢失的环节,一条消息从生产到消费完成这个过程,可以划分三个阶段,分别为消息生产阶段,消息存储阶段和消息消费阶段。



  • 消息生产阶段: 从消息被生产出来,然后提交给 MQ 的过程中,只要能正常收到 MQ Broker 的 ack 确认响应,就表示发送成功,所以只要处理好返回值和异常,这个阶段是不会出现消息丢失的。

  • 消息存储阶段: 这个阶段一般会直接交给 MQ 消息中间件来保证,但是你要了解它的原理,比如 Broker 会做副本,保证一条消息至少同步两个节点再返回 ack(这里涉及数据一致性原理,我在 04 讲中已经讲过,在面试中,你可以灵活延伸)。

  • 消息消费阶段: 消费端从 Broker 上拉取消息,只要消费端在收到消息后,不立即发送消费确认给 Broker,而是等到执行完业务逻辑后,再发送消费确认,也能保证消息的不丢失。


方案看似万无一失,每个阶段都能保证消息的不丢失,但在分布式系统中,故障不可避免,作为消费生产端,你并不能保证 MQ 是不是弄丢了你的消息,消费者是否消费了你的消息,所以,本着 Design for Failure 的设计原则,你还是需要一种机制,来 Check 消息是否丢失了。


紧接着,我们来看怎么进行消息检测? 总体方案解决思路为:在消息生产端,给每个发出的消息都指定一个全局唯一 ID,或者附加一个连续递增的版本号,然后在消费端做对应的版本校验。


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


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

发布于: 刚刚阅读数: 4
用户头像

Xiao8

关注

God bless the fighters. 2020.03.11 加入

欢迎关注公众号:程序猿Damon,长期从事Java开发,研究Springcloud的微服务架构设计。目前主要从事基于K8s云原生架构研发的工作,Golang开发,长期研究边缘计算框架KubeEdge、调度框架Volcano、容器云KubeSphere研究

评论

发布
暂无评论
消息队列的丢失、重复与积压问题_6月月更_Xiao8_InfoQ写作社区