MQ的消息丢失/重复/积压的问题解决

  • A+
所属分类:消息队列

在我们实际的开发过程中,我们肯定会用到MQ中间件,常见的MQ中间件有kafka,RabbitMQ,RocketMQ。在使用的过程中,我们必须要考虑这样一个问题,在使用MQ的时候,我们怎么确保消息100%不丢失?

案例背景

以我们熟悉的淘宝系统为例子,在用户下订单的时候,通常会给客户发放一下优惠劵。在整个过程中,交易服务和发优惠劵服务就是通过MQ消息队列进行通信。在交易服务完成后,交易服务可以发送“发一个满100减5的优惠劵”的消息给MQ。优惠劵服务则在消费端消费这个消息,从而实现真正的优惠劵的发放。

MQ的作用

在实际工作中,引入MQ消息最直接的目的就是让系统解耦合流量控制,从而实现系统的高可用和高性能。

  • 系统解耦:用MQ可以隔离系统上下游环境变换带来的不稳定因素。比如无论优惠劵服务需求如何变化,交易服务不用做任何改变。即使优惠劵服务出现故障,交易流程也不会受到影响。从而使交易服务和优惠劵服务达到了解耦的目的。从而使整个系统高可用。

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

但是,引入了MQ虽然实现了系统解耦和流量控制,同时引入引入了新的问题。

  1. 引入MQ实现系统解耦,会影响系统之间的数据传输一致性。在分布式系统中,如果两个节点之间存在数据同步,就会带来数据一致性问题。同理,在使用MQ时,我们也要解决消息生产端和消息消费端的数据一致性问题。

  2. 引入MQ实现流量控制,会使消费端处理能力不足,从而导致消息积压。

在使用MQ时,如何确保消息不丢失?
我们要从如下几个方面分析:如何知道有消息丢失?哪些环节可能丢失消息?如何确保消息不丢失?

网络中传输数据是不可靠的。要想解决如何不丢失消息的问题。首先,我们要知道哪些环节可能丢失消息。

哪些环节可能出现消息的丢失

一条消息从从生产到消费,主要可划为三个阶段:消息的生产阶段,消息的存储阶段,消息的消费阶段。
MQ的消息丢失/重复/积压的问题解决

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

  2. 消息的存储阶段:这个阶段一般交由MQ来保证。但是我们需要知道它的原理。比如,Broker会做副本,保证一条消息至少同步给两个节点再返回ack

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

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

如何check消息是否丢失?
总体的方案:在消息生产端,给发出的每一个消息指定一个全局唯一的ID,在消费端做校验。
具体实现:我们可以使用拦截器。在生产端发消息之前,通过拦截器将消息的全局唯一ID注入消息中,然后,在消费端收到消息之后,在通过拦截器检测消息ID或者消费状态。这样实现的好处就是消息的检测不会侵入业务代码中,可以通过ID找到具体丢失的消息,进行进一步的排查。

如何解决消息重复消费的问题

例如,在消息生产过程中,如果出现失败的情况,通过补偿机制会执行重试,重试就可能产生重复的消息,那么我们应该如何解决这个问题?这其实就是消费端幂等性问题(幂等性:就是一条命令执行任意多次所产生的影响和执行一次的影响相同)。只要消费者具备了幂等性,那么重复消费消息的问题也就解决了。
MQ的消息丢失/重复/积压的问题解决
最简单的方案就是,在数据库中建一个消息日志表,这个表记录消息ID和消息执行状态。这个我们消费消息的逻辑变为:在消息日志中增加一个消息记录,再根据消息记录,执行发送优惠劵业务。我们每次都会在插入之前检查该消息是否已存在。这样就不会出现一条消息被多次执行的情况。这里的数据库也可以使用redis/memcache来实现唯一约束方案。

如何解决消息积压问题

消息积压反应的是性能问题。因为消息发送之后才会出现积压,所以这个和消息生产者没有关系。绝大部分MQ单节点每秒几万的处理能力,相对比业务逻辑来说,性能一般不会在MQ的存储上。所以这个问题,我们主要是从消费端入手解决。
如果是线上的突发问题,要临时扩容,增加消费端的数量,与此同时,降级一些非核心业务。通过扩容和降级承担流量。
然后,需要排查异常问题。通过监控/日志等手段分析消费者的业务代码是否出现了问题。
最后,如果是消费端处理能力不足,可以通过水平扩容来提高消费端的并发处理能力。在扩容消费者实例数的时候,必须同步扩容Topic的分区数量,确保消费者实例数和分区数对等。如果只增加消费者数量,不增加分区数。由于分区是单线程消费的,这样扩容没有效果。

比如在Kafka中,一个Topic可以配置多个Partition。数据会被写入多个Partition中,但是kafka约定一个分区只能被一个消费者消费,Topic的partition数量也决定了最大消费者的数量。

除此之外,还有

  • 如何选型消息中间件?

  • 消息中间件中的队列模型与发布订阅模型的区别?

  • 为什么消息队列能实现高吞吐?

  • 序列化、传输协议,以及内存管理等问题?
    等问题。我们下一篇文章再讨论

发表评论

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen: