实际业务处理 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 存储msgId 与msg （包含 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）