Kafka 实现动态goupId 实现广播消息

Posted 2022-07-29 李昊轩的博客

有些场景下 我们需要广播消息, kafka在partition上的每条消息, 便宜想是按照group为单位做副本的, 所以针对每条消息, 每组groupId对应的消费者都会分别消费一次, 这也是我们通过动态groupId来实现 多副本jar的广播消费的原理

注意: 这里有一个坑, 一直变化的groupid 是记不住偏移量进度的, 所以需要设置kafka的新groupid拉取为latest 而不是其他

具体代码如下:
简单说下, @kafkaListener 有个属性叫做 beanRef 这个属性可以动态引入一个bean作为他自身的配置, 我们只需要注册一个具有随机groupId的配置bean即可.

@Component
public class DynamicGroupConsumer 

    @Autowired
    KafkaTemplate<String, Object> kafkaTemplate;

    @Bean("__listener2")
    public Listener listener() 
        String randomGroupName = UUID.randomUUID().toString();
        return new Listener(Collections.singletonList(TopicConstant.REFRESH_EVENTS), randomGroupName);
    

    /**
     * 刷新消费者
     */
    @KafkaListener(beanRef = "__listener2", topics = "#__listener2.topicList", groupId = "#__listener2.group")
    public void executeRefresh(String jsonEvents) 
     // anything