V2EX = way to explore
V2EX 是一个关于分享和探索的地方
现在注册
已注册用户请  登录
NoKey
V2EX  ›  Java

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

  •  1
     
  •   NoKey · 131 天前 · 2518 次点击
    这是一个创建于 131 天前的主题,其中的信息可能已经有所发展或是发生改变。
    在 springboot 里面写 kafka 消费代码
    为了增加消费速度,写了一个线程池,用来异步处理消费的消息
    那么,请教一下各位大佬
    如果,异步处理的时候,出现问题了,怎么重复消费这条消息呢?
    处理和消费是分开的,处理这里出问题了,消费那里无感知
    请各位大佬指点,谢谢
    30 条回复    2023-10-19 10:26:58 +08:00
    maocat
        1
    maocat  
       131 天前
    消息 ID ,加日志
    totoro52
        2
    totoro52  
       131 天前
    消费确认
    totoro52
        3
    totoro52  
       131 天前
    改为手动确认模式,当消费者消费完成后手动触发确认,MQ 才会删除这条消息。
    hidemyself
        4
    hidemyself  
       131 天前
    后处理呗,存本地表,定时任务扫描
    oneisall8955
        5
    oneisall8955  
       131 天前 via Android
    手动 ACK
    tramm
        6
    tramm  
       130 天前
    那就不要异步呗...
    手动提交的话,那就得等结果出来,跟同步处理没啥区别吧
    xiaofan2
        7
    xiaofan2  
       130 天前
    很明显你是没有开启手动提交 offset?
    正常来说,你消息扔到线程池后应该返回一个 Future ,然后你需要等待你这个消费处理完成,再给 broker 返回提交成功
    apisces
        8
    apisces  
       130 天前
    有个疑问,如果手动 ack 的话,那也需要等到线程池的 Future 返回才提交,这样和同步的区别在哪里呢
    NoKey
        9
    NoKey  
    OP
       130 天前
    @xiaofan2 这样是不是无法实现不断的处理消息的目的,总是要等待这一批消息处理完成,才可以继续处理下一批?
    WashFreshFresh
        10
    WashFreshFresh  
       130 天前
    @NoKey 处理完后手动 ack 比较好 没有 ack 的消息是会重发的
    lx0319
        11
    lx0319  
       130 天前
    去重机制。 或者设计容忍重复的方式,用于重复消费时,保持幂等
    可以处理后记录 offset 和 partition ,真出现问题,从故障 offset 重取就好 kafka 重置 offset 也可以,另开一个 group ,单独消费这条数据也行。
    venomD
        12
    venomD  
       130 天前   ❤️ 1
    1. 弄个死信队列,处理异常的直接丢到死信队列里
    2. 丢到数据库里,mysql ,redis ?定时任务扫描处理,成功后删除
    fengpan567
        13
    fengpan567  
       130 天前
    线程池异步处理有问题,消息如果在线程池的队列中,重启服务的时候数据不就丢了。还不如写到中间表,定时去捞未处理数据
    lsk569937453
        14
    lsk569937453  
       130 天前
    @fengpan567 重启服务的时候内存的数据是没有了,但是因为消息没有 commit ,所以可以从 mq 再次消费没有 commit 的消息。
    4kingRAS
        15
    4kingRAS  
       130 天前
    @apisces 不阻塞 kafka 线程啊,同步的话 kafka 线程会阻塞到你提交为止
    julyclyde
        16
    julyclyde  
       130 天前
    消费和处理分开,这设计是不是有点问题啊?

    是不是应该改成各线程分别消费并就地处理?
    jfds
        17
    jfds  
       130 天前
    不用线程池消息会堆积么,干嘛要增加消费速度。MQ 已经是异步的了,一般对 rt 不敏感,为了这一点速度搞异步链路引出新的问题不值得。
    zhhmax
        18
    zhhmax  
       130 天前 via iPad
    办法很多,比如说消息 ID 放到 redis ,处理成功就去更新一下 redis 这个消息的自动过期时间等自动过期就行,一直没过期的就是处理失败的,如何确定一直失败呢,消费一次计数器加一即可。这样也不影响其他线程消费队列导致阻塞。失败到一定次数还可以加入其他逻辑人工干预。
    Luckyshot
        19
    Luckyshot  
       130 天前
    消息表+定时任务
    rainbowStay
        20
    rainbowStay  
       130 天前
    @zhhmax 即使消息 ID 放到 redis ,那消费失败了消息耶没法重发啊?
    rainbowStay
        21
    rainbowStay  
       130 天前
    @totoro52 #3 提个问题,如果不手动触发确认,MQ 是会阻塞还是会继续处理后面的其他消息?
    zhhmax
        22
    zhhmax  
       130 天前 via iPad
    @rainbowStay 不用重发了,因为消息已经发成功了,有 ID 就能通过其他方式拿到完整消息的,要是消息再发一遍就重复了,这里和消费已经没关系了。楼主的异步处理消息的方式已经决定了不能再有重发消息的步骤,不然处理消息的那部分逻辑还得加上消息是否是处理失败的消息,增加代码复杂度。
    w4n9hu1
        23
    w4n9hu1  
       130 天前 via iPhone
    加个 retry queue 和 dlq? 之前总结过一篇文章 https://w4n9hu1.dev/2023/05/26/exception-handling-in-event-driven-architectures/
    rainbowStay
        24
    rainbowStay  
       130 天前
    @zhhmax #22 根据 ID 能怎么拿到完整消息啊。。。那我理解要么就上游保留了关联 id 的消息信息,要么就是楼主在消费信息时自己保存一遍关联 id 的消息信息,这样不是更加大了复杂性吗?
    zhhmax
        25
    zhhmax  
       130 天前 via iPad
    @rainbowStay 我们可以讨论更严谨一点,这个 kafka 如果连接的是两个不同的系统,那么你说的无法通过 ID 得到完整消息确实是个问题,保存 ID 的时候可以做到把这条失败的完整消息再保存到其他地方而不用再考虑重发问题。如果是上游的消息生产者也是自己内部系统,只能在消息队列中才能得知完整的消息内容而无法通过 ID 再从其他途径得到同样的内容在我个人看来是属于重大的设计缺陷,即便是这样,那也可以换个方式,新建一个队列把失败的消息放进去让消费原来消息队列的任务也监听一下这个队列就可以了,那么新的问题又来了,这样做到重发了,但是如果任务一直处理失败会不会放大数据量引发其他问题,比如说某段时间某批数据永远无法处理成功而一直重发会不会影响到其他批次正常消息的处理效率问题。
    liprais
        26
    liprais  
       130 天前
    消费成功了再去更新 offset,不成功就重试呗
    你不更新 offset 永远读的是这一条就完了
    rainbowStay
        27
    rainbowStay  
       130 天前
    给楼主找了下 Kafka 的文档: https://kafka.apache.org/20/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html ,直接搜”consume a batch of records“定位到关键部分
    fkdog
        28
    fkdog  
       130 天前
    假设前提是,你的 kafka topic 允许乱序消费。
    1. 一次从 kafka 里取 n 条数据,比如 100 条。
    2. 然后 100 条丢异步处理。手动 ack 从第一条开始消息到第一条失败消息区间里最大的那条 offset 作为 ack 提交。比如 1 ~ 33 是成功的,34 是失败的,那么 ack 33.
    3. 消费端做幂等处理。防止消息重复消费。
    4. 对于重复发生错误的消息,做 retry n 次还是失败的话,单独记录人工介入。
    NoKey
        29
    NoKey  
    OP
       130 天前
    @fkdog 感谢,感觉这说的很全面了,但是,这个依然是阻塞式的,就是这 100 条会阻塞着处理完成后,才可能进行下一组,然后,异步处理 100 条,找到成功处理的 offset ,这个也是要等到这 100 个全部完成了才知道,中途是不确定的。可能还是楼上说的,要异步处理可能需要手动解决失败的信息,放重试队列或者放数据库里。
    fkdog
        30
    fkdog  
       129 天前
    @NoKey
    所以你说的“阻塞式”是有什么问题吗?
    不想丢消息,你还是要用到 ack offset 。
    只要你是在本地开了线程池异步去消费消息,因为线程池也是乱序的,那么你还是需要记录每条消息的处理结果,然后合并计算连续成功区间内的最后一条 offset 。

    感觉你的想法是来一条就自动 offset 一条,然后本地保存这条记录。
    那还不如直接任务存数据库+定时轮训扫描。
    关于   ·   帮助文档   ·   博客   ·   API   ·   FAQ   ·   我们的愿景   ·   实用小工具   ·   820 人在线   最高记录 6543   ·     Select Language
    创意工作者们的社区
    World is powered by solitude
    VERSION: 3.9.8.5 · 33ms · UTC 21:53 · PVG 05:53 · LAX 13:53 · JFK 16:53
    Developed with CodeLauncher
    ♥ Do have faith in what you're doing.