请问一下,多节点消费 kafka 如何做到串行消费 - V2EX
V2EX = way to explore
V2EX 是一个关于分享和探索的地方
现在注册
已注册用户请  登录
NoKey
V2EX    Kafka

请问一下,多节点消费 kafka 如何做到串行消费

  •  
  •   NoKey 2024-08-27 22:46:30 +08:00 3529 次点击
    这是一个创建于 411 天前的主题,其中的信息可能已经有所发展或是发生改变。
    就是有个场景,有个任务要处理,任务必须串行执行
    为了保证任务下发后不丢失
    打算借助 kafka 的持久化,把任务丢到 kafka 中
    后台服务有多个节点,也就是多个节点在消费 kafka
    怎么样才能做到串行消费 kafka 的消息呢?
    就是说,多个节点排队消费,第一个节点拿到消息后,如果没处理完,其他节点不能继续消费
    搞了好久都没实现
    又不想用数据库存储然后定时读取的方式(这种方式肯定没问题)
    想通过这个功能,学会 kakfa 的使用
    请假一下各位大佬,基于 kafka 能不能实现这个想法呢?谢谢
    第 1 条附言    2024-08-28 16:36:17 +08:00
    统一回复一下,应该是我调试的时候搞错了,今天我重新梳理了一下配置,当前验证的场景如下:
    把消息发送到 topic 下的同一个分区,多个同 groupID 的消费者去消费,只有一个消费者可以拿到消息,且,这个消费者没处理完消息之前,其他消息都会被阻塞,等待这条消息处理完,才会继续消费下一条消息。已满足当前使用。
    谢谢大家的回复
    31 条回复    2025-03-12 13:14:27 +08:00
    Trim21
        1
    Trim21  
       2024-08-27 22:48:56 +08:00
    只用一个节点消费?
    kechx
        2
    kechx  
       2024-08-27 22:51:04 +08:00
    进队列前对任务进行一个 hash ,有时序性的扔在一起
    7911364440
        3
    7911364440  
       2024-08-27 22:51:43 +08:00
    给任务加个字增 id ,每个消费服务执行任务前校验一下顺序
    crysislinux
        4
    crysislinux  
       2024-08-27 22:55:34 +08:00 via Android
    kafka 一个分区不就是顺序执行的么。你还想要啥
    MeiJiayun
        5
    MeiJiayun  
       2024-08-27 23:02:18 +08:00 via iPhone
    指定 key ,key 相同会推送到相同分区,同一分区的消息有顺序
    dddd1919
        6
    dddd1919  
       2024-08-27 23:08:55 +08:00
    给串行的每个节点定义一个 topic 和 topic 的串行顺序,每个节点执行器执行完成后把消息推送到下一个 topic
    kur0d3s
        7
    kur0d3s  
       2024-08-28 08:56:01 +08:00
    要求单实例顺序执行了, 那消费者多实例的意义是啥...
    Wh1t3zZ
        8
    Wh1t3zZ  
       2024-08-28 09:04:05 +08:00
    "多个节点排队消费,第一个节点拿到消息后,如果没处理完,其他节点不能继续消费"

    没看懂多个节点消费的意义是什么
    NoKey
        9
    NoKey  
    OP
       2024-08-28 09:08:01 +08:00
    @Wh1t3zZ 因为服务自己是多节点部署的,启动起来,kafka 就订阅消费了。。。
    NoKey
        10
    NoKey  
    OP
       2024-08-28 09:09:30 +08:00
    @crysislinux 消费节点是多个,我测试的,即便多个消费者在一个消费者组,第一个消费者拿到消息处理的过程中,第二条消息到来,也会被消费,这个时候,第一条消息和第二条消息就是并行在处理了
    NoKey
        11
    NoKey  
    OP
       2024-08-28 09:10:51 +08:00
    @Trim21 主要是服务是多节点部署,启动起来自然开始订阅消费
    zt5b79527
        12
    zt5b79527  
       2024-08-28 09:14:00 +08:00   2
    单个实例去消费不就完了,如果必须是多个实例,就去抢分布式锁,拿到锁的才能去消费
    frank1256
        13
    frank1256  
       2024-08-28 09:18:27 +08:00
    多节点抢锁,抢不到锁的 commit offset 要回退的,因为你第二个节点虽然拿不到锁,但已经消费到数据了
    crysislinux
        14
    crysislinux  
       2024-08-28 09:24:02 +08:00 via Android
    @NoKey 消费者组和分区不是一个概念吧,你搞错了。
    echoZero
        15
    echoZero  
       2024-08-28 09:35:08 +08:00
    放同一个分区。之所有会出现多条消息并行处理,因为是批量拉取消费消息,这个是你的消费逻辑问题,看一下是不是消费逻辑并发处理拉取的消息了
    zdt3476
        16
    zdt3476  
       2024-08-28 09:50:31 +08:00
    你给这个 topic 只分配一个 partition 就好了,这样多消费者就不起作用了
    awalkingman
        17
    awalkingman  
       2024-08-28 10:00:48 +08:00
    @zt5b79527 正解
    MoYi123
        18
    MoYi123  
       2024-08-28 10:14:46 +08:00
    kafka 在这起什么作用? rpc 调用就会导致任务丢失了?
    Marinaaaa
        19
    Marinaaaa  
       2024-08-28 10:33:00 +08:00   3
    1. topic 单分区 ,多消费者, 这样再多消费者也就只有一个在工作。

    2. 多分区,多消费者,任务投递的指定 key ,这样同样 key 的任务在同一个分区,也是被顺序消费的。

    3. 分布式锁 , 抢到锁的去消费
    NoKey
        20
    NoKey  
    OP
       2024-08-28 10:54:23 +08:00
    @Marinaaaa 我现在用的是第二种,指定生成者往指定分区发消息,消费者就算只有一个,也有问题:我在消费方法中加了 sleep 来模拟处理数据耗时,第一条消息还在 sleep 的时候,第二条消息到来,也会进入处理阶段,也就是消费者相当于多线程的,不断的在消费。不知道这种情况如何处理
    wwhontheway
        21
    wwhontheway  
       2024-08-28 10:56:39 +08:00
    没看懂,你说的串行是什么? task 按顺序一个一个消费?
    ------------------------
    就是说,多个节点排队消费,第一个节点拿到消息后,如果没处理完,其他节点不能继续消费
    ------------------------
    这又是啥?这不是串行呀。这是 task lock 保证的
    zhhmax
        22
    zhhmax  
       2024-08-28 11:17:37 +08:00
    @NoKey #20 你手动提交每条消息的 offset ,然后把消费的方法改成阻塞式,这样每次消费完再手动提交之后才会消费下一条消息。
    NoKey
        23
    NoKey  
    OP
       2024-08-28 11:21:17 +08:00
    @wwhontheway 就是想知道 springboot+kafka ,能不能做成一个简单的串行处理队列
    wwhontheway
        24
    wwhontheway  
       2024-08-28 11:59:11 +08:00
    @NoKey 那就是单实例呀,多实例串行,工作只有一个,其他实例闲着,何必多实例?
    R4rvZ6agNVWr56V0
        25
    R4rvZ6agNVWr56V0  
       2024-08-28 12:23:32 +08:00
    创建一个专门用于处理这些串行任务的 Kafka 主题,然后:
    使用单一消费者实例,属于自己的消费者组
    使用 TransactionalId 来跟踪每个任务的进度
    实现一个轮询机制,定期检查任务是否已经完成,如果完成则提交 offset 并删除该任务
    可以考虑使用 Kafka Streams 或 KSQL 来简化实现这个业务
    csys
        26
    csys  
       2024-08-28 13:09:03 +08:00
    你手动指定分区或者使用相同的消息 key ,这样所有消息都会被发送到同一个分区,而一个分区只会被分配给一个消费者,相当于你只有一个订阅到此分区的消费实例在消费这些消息,其它消费者都在空跑
    guojiawei0203
        27
    guojiawei0203  
       2024-08-28 13:16:03 +08:00
    没啥量直接单分区,多分区根据 key 去 hash
    wysnxzm
        28
    wysnxzm  
       2024-08-28 13:47:48 +08:00
    将任务分组保证同一组的任务被同一个消费者消费,该消费者保证自身所有任务串行执行
    这样可以在多消费者的同时保证串行
    vaynecv
        29
    vaynecv  
       2024-08-28 14:47:32 +08:00
    启动的时候指定某一个实例进行消费,其他示例忽略此消息
    night98
        30
    night98  
       2024-08-28 18:49:07 +08:00
    你是想 1 到 10 总共十条消息,无论如何都要保证按照 1 到 10 按顺序消费,不能出现 1 ,2 同时消费,然后再 3 ,4 同时消费的情况,是这个业务逻辑是吗?

    spring kafka 限制单次消费拉取的数据量,然后就是楼上那些说的,一起弄一下就 ok 了

    实际逻辑就是一次必须只能有一个人持有消息,持有后必须按顺序消费。
    tangproxy
        31
    tangproxy  
       214 天前 via Android
    @vaynecv 赞同,后台服务多实例,初始化的时候通过分布式锁确定订阅 topic 的实例。然后,补充一点,防止这个订阅了 topic 的实例宕机,后台其他实例监听分布式锁过期的订阅,redission 那个,再进行重新抢锁,订阅 topic 。如果过期时间比较长(默认 10 分钟)无法接受的话,可以再加个补偿,举个例子,管理后台服务的程序(运管)可能最先知道后台实例挂了,通知其他实例这个情况。
    关于     帮助文档     自助推广系统     博客     API     FAQ     Solana     5899 人在线   最高记录 6679       Select Language
    创意工作者们的社区
    World is powered by solitude
    VERSION: 3.9.8.5 42ms UTC 03:19 PVG 11:19 LAX 20:19 JFK 23:19
    Do have faith in what you're doing.
    ubao snddm index pchome yahoo rakuten mypaper meadowduck bidyahoo youbao zxmzxm asda bnvcg cvbfg dfscv mmhjk xxddc yybgb zznbn ccubao uaitu acv GXCV ET GDG YH FG BCVB FJFH CBRE CBC GDG ET54 WRWR RWER WREW WRWER RWER SDG EW SF DSFSF fbbs ubao fhd dfg ewr dg df ewwr ewwr et ruyut utut dfg fgd gdfgt etg dfgt dfgd ert4 gd fgg wr 235 wer3 we vsdf sdf gdf ert xcv sdf rwer hfd dfg cvb rwf afb dfh jgh bmn lgh rty gfds cxv xcv xcs vdas fdf fgd cv sdf tert sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf sdf shasha9178 shasha9178 shasha9178 shasha9178 shasha9178 liflif2 liflif2 liflif2 liflif2 liflif2 liblib3 liblib3 liblib3 liblib3 liblib3 zhazha444 zhazha444 zhazha444 zhazha444 zhazha444 dende5 dende denden denden2 denden21 fenfen9 fenf619 fen619 fenfe9 fe619 sdf sdf sdf sdf sdf zhazh90 zhazh0 zhaa50 zha90 zh590 zho zhoz zhozh zhozho zhozho2 lislis lls95 lili95 lils5 liss9 sdf0ty987 sdft876 sdft9876 sdf09876 sd0t9876 sdf0ty98 sdf0976 sdf0ty986 sdf0ty96 sdf0t76 sdf0876 df0ty98 sf0t876 sd0ty76 sdy76 sdf76 sdf0t76 sdf0ty9 sdf0ty98 sdf0ty987 sdf0ty98 sdf6676 sdf876 sd876 sd876 sdf6 sdf6 sdf9876 sdf0t sdf06 sdf0ty9776 sdf0ty9776 sdf0ty76 sdf8876 sdf0t sd6 sdf06 s688876 sd688 sdf86