kafka 集群实例消费同一个topic 动态设置消费组

背景：我服务A部署集群 3个实例。

场景：前端WS连接A服务的实例。服务端往客户端推送数据。

触发点：另一个系统通过kafka通知过来。

遇到的问题：按kafka的消费规则。这三个集群实例在同一个group中，及groupId是相同的。这时候只会有一个实例收到kafka消息。只会有一个客户端收到服务端推送的消息。另外两个实例是收不到kafka消息，就不会往前端推送数据。

 

目标：这三个集群实例都收到kafka消息。

 

解决方案：在实例启动时候动态创建自己实例 消费kafka的groupId。这样三个集群实例不在一个消费组中，自然都能收到kafka消息了。

 

代码实现：

 @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> instanceChangeContainerFactory() {
        Map<String, Object> properties = new ConcurrentHashMap<>(delayContainerFactory.getConsumerFactory().getConfigurationProperties());

        // 每个实例的groupId动态改成ip
        String nodeIp = System.getenv(Constants.ENV_CONTAINER_IP);
        String ip = StringUtils.isNotBlank(nodeIp) ? nodeIp : NetUtils.localIP();

        String groupId = ObjectUtils.isEmpty(properties.get(ConsumerConfig.GROUP_ID_CONFIG)) ? ip : properties.get(ConsumerConfig.GROUP_ID_CONFIG) + "_" + ip;
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);

        ConsumerFactory<String, String> consumerFactory = new DefaultKafkaConsumerFactory<>(properties);

        ConcurrentKafkaListenerContainerFactory<String, String> instanceChangeContainerFactory = new ConcurrentKafkaListenerContainerFactory<>();
        instanceChangeContainerFactory.setAutoStartup(kafkaOpenListener);
        instanceChangeContainerFactory.setConsumerFactory(consumerFactory);

        return instanceChangeContainerFactory;
    }