实际业务处理 Kafka 消息丢失、重复消费和顺序消费的问题

关于 Kafka 消息丢失、重复消费和顺序消费的问题

消息丢失,消息重复消费,消息顺序消费等问题是我们使用 MQ 时不得不考虑的一个问题,下面我结合实际的业务来和你分享一下解决方案。

消息丢失问题

比如我们使用 Kakfa 时,以下场景都会发生消息丢失:

  • producer -> broker (生产者生产消息)
  • broker -> broker (集群环境,broker 同步给其他 broker)
  • broker -> consumer (消费者消费消息)

解决方案也很简单,设置 acks(消息确认机制)retries(重试机制)factor(设置 partition 数量)…

一般来说,最常见的消息丢失场景就是:consumer 消费消息

要保证 consumer 消费消息时不丢失消息,必须使用手动提交 ack

我们业务是这样实现的:

  1. Kafka 拉取消息(一次批量拉取 100条)
  2. 为每条消息分配一个 msgId(递增)
  3. msgId 存入内存队列(sortSet)
  4. 使用 Map 存储 msgIdmsg (包含 offset)的映射关系
  5. 当业务处理完消息后,获取当前消息的 msgId,然后从 sortSet 中删除该 msgId(表示该消息已经处理过了)
  6. ack 时,如果当前 msgId <= sortSet(msgId 在 sortSet 中是从小到大排列) ,就提交当前 offset
  7. 就算 consumer 在处理消息时挂了,下次重启时就会从 sortSet 队首的消息开始拉取,实现至少处理一次语义。
  8. 步骤 7 存在一个问题:当消息处理完后,还没从 sortSet 中删除该 msgId,系统就挂了,当系统重启时,又会重新处理一次刚刚已处理过的消息,这就引出消息重复消费的问题了。

消息重复消费

要解决消息重复消费,也就是要实现幂等(幂等就是:多次请求,但结果保持不变,举一个例子你就明白了:在 http 中,你发送同一个 get 请求,无论发送多少次,返回结果都是一样的

回到我们的业务场景上,我以处理订单消息为例:

  • 幂等Key 由我们的订单Id + 订单状态组成(一笔订单的状态只会处理一次)

  • 在处理之前,我们首先会去 Redis 查询是否存在这个 Key

    如果存在,说明我们已经处理过了,直接丢掉;

    ​ 如果不存在,说明没处理过,继续往下处理;

  • 最终的逻辑是:将处理过的数据存到DB上,再把 幂等Key 存到 Redis

显然一般场景下 Redis 是无法保证幂等的

所以Redis只是一个前置处理,最终的幂等性依赖 DB唯一Key(订单Id+订单状态)

总的来说就是:通过 Redis 做前置处理,DB 唯一索引做最终保证实现幂等性

消息顺序消费

消息的顺序性很好理解,还是以订单处理为例

订单的状态有:支付、确认收货、完成等等,而订单下还有计费、退款的消息报

理论上来说:支付的消息肯定要比退款的消息先到。

但是程序处理的过程就不一定了,所以我们处理消息顺序消费的流程如下:

  • 宽表:创建一张宽表,唯一索引是 订单Id,将订单的每个状态拆分为一个列,当消息来了,只更新对应的字段就好,消息只会存在短暂的状态不一致问题,但是最终状态是一致的
  • 消息补偿机制
  • 把相同的 userID/orderId 发送到相同的 partition(因为一个 consumer 消费一个 partition)
页面下部广告