mq是一种趋势,总体来说对我们的系统是利大于弊的,难道因为它会出现一些问题,我们就不用它了?

那么我们要如何解决这些问题呢?

4.1 重复消息问题

不管是由于生产者产生的重复消息,还是由于消费者导致的重复消息,我们都可以在消费者中这个问题。

这就要求消费者在做业务处理时,要做幂等设计,如果有不知道如何设计的朋友,可以参考《高并发下如何保证接口的幂等性?》,里面介绍得非常详情。

在这里我推荐增加一张消费消息表,来解决mq的这类问题。消费消息表中,使用messageId做唯一索引,在处理业务逻辑之前,先根据messageId查询一下该消息有没有处理过,如果已经处理过了则直接返回成功,如果没有处理过,则继续做业务处理。

image

4.2 数据一致性问题

我们都知道数据一致性分为:

  1. 强一致性
  2. 弱一致性
  3. 最终一致性

而mq为了性能考虑使用的是最终一致性,那么必定会出现数据不一致的问题。这类问题大概率是因为消费者读取消息后,业务逻辑处理失败导致的,这时候可以增加重试机制。

重试分为:同步重试 和 异步重试。

有些消息量比较小的业务场景,可以采用同步重试,在消费消息时如果处理失败,立刻重试3-5次,如何还是失败,则写入到记录表中。但如果消息量比较大,则不建议使用这种方式,因为如果出现网络异常,可能会导致大量的消息不断重试,影响消息读取速度,造成消息堆积。

image

而消息量比较大的业务场景,建议采用异步重试,在消费者处理失败之后,立刻写入重试表,有个job专门定时重试。

还有一种做法是,如果消费失败,自己给同一个topic发一条消息,在后面的某个时间点,自己又会消费到那条消息,起到了重试的效果。如果对消息顺序要求不高的场景,可以使用这种方式。

4.3 消息丢失问题

不管你是否承认有时候消息真的会丢,即使这种概率非常小,也会对业务有影响。生产者、mq服务器、消费者都有可能会导致消息丢失的问题。

为了解决这个问题,我们可以增加一张消息发送表,当生产者发完消息之后,会往该表中写入一条数据,状态status标记为待确认。如果消费者读取消息之后,调用生产者的api更新该消息的status为已确认。有个job,每隔一段时间检查一次消息发送表,如果5分钟(这个时间可以根据实际情况来定)后还有状态是待确认的消息,则认为该消息已经丢失了,重新发条消息。

image这样不管是由于生产者、mq服务器、还是消费者导致的消息丢失问题,job都会重新发消息。

4.4 消息顺序问题

消息顺序问题是我们非常常见的问题,我们以kafka消费订单消息为例。订单有:下单、支付、完成、退货等状态,这些状态是有先后顺序的,如果顺序错了会导致业务异常。

解决这类问题之前,我们先确认一下,消费者是否真的需要知道中间状态,只知道最终状态行不行?

image

其实很多时候,我真的需要知道的是最终状态,这时可以把流程优化一下:

image这种方式可以解决大部分的消息顺序问题。

但如果真的有需要保证消息顺序的需求。订单号路由到不同的partition,同一个订单号的消息,每次到发到同一个partition。image

4.5 消息堆积

如果消费者消费消息的速度小于生产者生产消息的速度,将会出现消息堆积问题。其实这类问题产生的原因很多,如果你想进一步了解,可以看看我的另一篇文章《我用kafka两年踩过的一些非比寻常的坑》。

那么消息堆积问题该如何解决呢?

这个要看消息是否需要保证顺序。

如果不需要保证顺序,可以读取消息之后用多线程处理业务逻辑。

image

这样就能增加业务逻辑处理速度,解决消息堆积问题。但是线程池的核心线程数和最大线程数需要合理配置,不然可能会浪费系统资源。

如果需要保证顺序,可以读取消息之后,将消息按照一定的规则分发到多个队列中,然后在队列中用单线程处理。

image

好了,今天先分享到这来,下期再见。我在这里只是抛砖引玉,其实mq相关的内容还有很多,比如:定时发送、延迟发送、私信队列、事务问题等等,后面会逐步完善相关内容。

最后修改:2026 年 06 月 05 日
如果觉得我的文章对你有用,请随意赞赏