有些场景下 我们需要广播消息, 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

    }


}
Logo

华为开发者空间,是为全球开发者打造的专属开发空间,汇聚了华为优质开发资源及工具,致力于让每一位开发者拥有一台云主机,基于华为根生态开发、创新。

更多推荐