Kafka 实现动态goupId 实现广播消息
有些场景下 我们需要广播消息, kafka在partition上的每条消息, 便宜想是按照group为单位做副本的, 所以针对每条消息,每组groupId对应的消费者都会分别消费一次, 这也是我们通过动态groupId来实现 多副本jar的广播消费的原理
·
有些场景下 我们需要广播消息, 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
}
}
更多推荐
已为社区贡献3条内容
所有评论(0)