一段解决kafka消息处理异常的经典对话

有一天，卡尔维护的购买系统发生了一个奇怪的异常，从日志里看到，购买后的任务处理竟然先于购买任务执行了。“不可能啊，按照代码的顺序，一定是先执行购买流程，再发送消息到kafka，最后消费端接收到消息后执行购买后的一些善后任务。从A到B到C，顺序清清楚楚。” 于是，他请教了马克，马克眯着眼睛细看了一会，道:"问题是不是出在这段@Transaction注解上？"

伪代码：

马克说道：“这kafka消息鬼的很，它没准在事务提交之前就发送出去了，而消费者在fetch消息执行业务流程的时候这段事务仍然没有提交，这就导致了数据上的乱序，看上去就像购买后任务先于购买任务执行。”

“那该怎么改呢？把kafkaTemplete.sendMdg()这段移出方法，等事务提交了再发送消息？但我把消息发送这步写在事务注解的方法内部，就是为了在消息发送失败的时候能够实现回滚。如果移出来，而消息发送的时候失败，那怎么办？” 卡尔问道。

“可以考虑使用本地消息表。” 马克说着在白板上写下了一段伪代码：

producer：

consumer：

"主要是通过时效性高的MQ，自动触发事件；万一消息发送失败，也可以通过数据表的消息记录轮询来保证。不过关系数据库的吞吐量和性能存在瓶颈，频繁的读写消息会给数据库造成压力，考虑当前场景，稳定性要求较高，而并发量还没有上来。可以考虑这种方法。” 马克道。

卡尔改完后，测试发布，之后再也没出现乱序了，但消息有时会莫名地丢失或者重复消费，卡尔不得不经常查看线上日志，手工修复一些数据问题，费时费力，只能在晚上加班加点开发新的业务。

卡尔道：“是不是自动提交会带来一些不可控因素？”

马克道：“对，当我们的配置是自动提交的时候，消费者的消息投递保证有可能是at least once，或者at most once。当到达提交时间间隔，触发Kafka自动提交上次的偏移量时，就可能发生at most once的情况， 在这段时间，如果消费者还没完成消息的处理进程就崩溃了， 消费者进程重新启动时，它开始接收上次提交的偏移量之后的消息，实际上消费者可能会丢失几条消息；而当消费者处理完消息并将消息提交到持久化存储系统，而消费者进程崩溃时，会发生at least once的情况。 在此期间，kafka没有向broker提交offset，因为自动提交时间间隔没有过去。 当消费者进程重新启动时，会收到从上次提交的偏移量开始的一些旧消息。”

“正是这个导致消息丢失或者重复消费现象，那你想怎么改呢？” 卡尔道。

“这防止了消息丢失，但消息重复问题该怎么解决？”

“至于你说的消费重复问题，主要解决思路是在服务层实现幂等性。让接收端支持消息去重的功能。比如在上面的伪代码中，record中放一个唯一键字段，消费时根据唯一键查询这条消息，判断是否消费过。也可以通过redis缓存来实现类似的机制。”

卡尔道：“真是这样子的吗？”

“尽信书不如无书，尤其是技术，是需要经过长时间的时间检验的，你对此有所怀疑的话可以在本地开发环境优化试试看。” 马克道。