springboot 里面异步处理消费到的 kafka 消息，如何保证不丢消息呢

[NoKey]

在 springboot 里面写 kafka 消费代码
为了增加消费速度，写了一个线程池，用来异步处理消费的消息
那么，请教一下各位大佬
如果，异步处理的时候，出现问题了，怎么重复消费这条消息呢？
处理和消费是分开的，处理这里出问题了，消费那里无感知
请各位大佬指点，谢谢

[rainbowStay]

@totoro52 #3 提个问题，如果不手动触发确认，MQ 是会阻塞还是会继续处理后面的其他消息？

[zhhmax]

@rainbowStay 不用重发了，因为消息已经发成功了，有 ID 就能通过其他方式拿到完整消息的，要是消息再发一遍就重复了，这里和消费已经没关系了。楼主的异步处理消息的方式已经决定了不能再有重发消息的步骤，不然处理消息的那部分逻辑还得加上消息是否是处理失败的消息，增加代码复杂度。

[w4n9hu1]

加个 retry queue 和 dlq? 之前总结过一篇文章 https://w4n9hu1.dev/2023/05/26/exception-handling-in-event-driven-architectures/

[rainbowStay]

@zhhmax #22 根据 ID 能怎么拿到完整消息啊。。。那我理解要么就上游保留了关联 id 的消息信息，要么就是楼主在消费信息时自己保存一遍关联 id 的消息信息，这样不是更加大了复杂性吗？

[zhhmax]

@rainbowStay 我们可以讨论更严谨一点，这个 kafka 如果连接的是两个不同的系统，那么你说的无法通过 ID 得到完整消息确实是个问题，保存 ID 的时候可以做到把这条失败的完整消息再保存到其他地方而不用再考虑重发问题。如果是上游的消息生产者也是自己内部系统，只能在消息队列中才能得知完整的消息内容而无法通过 ID 再从其他途径得到同样的内容在我个人看来是属于重大的设计缺陷，即便是这样，那也可以换个方式，新建一个队列把失败的消息放进去让消费原来消息队列的任务也监听一下这个队列就可以了，那么新的问题又来了，这样做到重发了，但是如果任务一直处理失败会不会放大数据量引发其他问题，比如说某段时间某批数据永远无法处理成功而一直重发会不会影响到其他批次正常消息的处理效率问题。

[liprais]

消费成功了再去更新 offset,不成功就重试呗
你不更新 offset 永远读的是这一条就完了

[rainbowStay]

给楼主找了下 Kafka 的文档： https://kafka.apache.org/20/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html ，直接搜”consume a batch of records“定位到关键部分

[fkdog]

假设前提是，你的 kafka topic 允许乱序消费。
1. 一次从 kafka 里取 n 条数据，比如 100 条。
2. 然后 100 条丢异步处理。手动 ack 从第一条开始消息到第一条失败消息区间里最大的那条 offset 作为 ack 提交。比如 1 ～ 33 是成功的，34 是失败的，那么 ack 33.
3. 消费端做幂等处理。防止消息重复消费。
4. 对于重复发生错误的消息，做 retry n 次还是失败的话，单独记录人工介入。

[NoKey]

@fkdog 感谢，感觉这说的很全面了，但是，这个依然是阻塞式的，就是这 100 条会阻塞着处理完成后，才可能进行下一组，然后，异步处理 100 条，找到成功处理的 offset ，这个也是要等到这 100 个全部完成了才知道，中途是不确定的。可能还是楼上说的，要异步处理可能需要手动解决失败的信息，放重试队列或者放数据库里。

[fkdog]

@NoKey
所以你说的“阻塞式”是有什么问题吗？
不想丢消息，你还是要用到 ack offset 。
只要你是在本地开了线程池异步去消费消息，因为线程池也是乱序的，那么你还是需要记录每条消息的处理结果，然后合并计算连续成功区间内的最后一条 offset 。

感觉你的想法是来一条就自动 offset 一条，然后本地保存这条记录。
那还不如直接任务存数据库+定时轮训扫描。