Kafka 2消费者工厂监听器没有不断连接答案

【问题标题】：Kafka 2 consumer factories listeners are not connected constantlyKafka 2消费者工厂监听器没有不断连接
【发布时间】：2018-10-12 21:25:23
【问题描述】：

我们在项目中使用的是 Spring Kafka 2.1.4.RELEASE 版本，我们有以下配置：

@EnableKafka
public class KafkaConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Configuration
    class ProducerConfig {
        @Bean
        public Map<String, Object> producerConfigs() {
            Map<String, Object> props = new HashMap<>();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ASerializer.class);
            return props;
        }

        @Bean
        public ProducerFactory<String, A> producerFactory() {
            return new DefaultKafkaProducerFactory<>(producerConfigs());
        }

        @Bean
        public KafkaTemplate<String, A> kafkaTemplate() {
            return new KafkaTemplate<>(producerFactory());
        }
    }

    @Configuration
    class ConsumerConfig {
        @Value("${spring.kafka.consumer.group-id}")
        private String groupId;
        @Value("${spring.kafka.consumer.auto-offset-reset}")
        private String autoOffsetReset;
        @Value("${spring.kafka.consumer.enable-auto-commit}")
        private boolean enableAutoCommit;
        @Value("${spring.kafka.consumer.max-poll-records}")
        private Integer maxPollRecords;

        @Bean
        public Map<String, Object> firstConsumerConfig() {
            Map<String, Object> props = getCommonConsumerConfig();
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ADeserializer.class);
            return props;
        }

        @Bean
        public Map<String, Object> secondConsumerConfig() {
            Map<String, Object> props = getCommonConsumerConfig();
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, BDeserializer.class);
            return props;
        }

        @Bean
        public ConsumerFactory<String, A> firstConsumerFactory() {
            return new DefaultKafkaConsumerFactory<>(firstConsumerConfig());
        }

        @Bean
        public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, A>> firsttContainerFactory() {
            ConcurrentKafkaListenerContainerFactory<String, A> factory = new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(firstConsumerFactory());
            return factory;
        }

        @Bean
        public ConsumerFactory<String, B> secondConsumerFactory() {
            return new DefaultKafkaConsumerFactory<>(secondConsumerConfig());
        }

        @Bean
        public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, B>> outputTopicContainerFactory() {
            ConcurrentKafkaListenerContainerFactory<String, B> factory = new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(secondConsumerFactory());
            return factory;
        }


        private Map<String, Object> getCommonConsumerConfig() {
            Map<String, Object> props = new HashMap<>();
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
            props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
            props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
            props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
            return props;
        }
    }
}

如您所见，我们使用了 2 个消费者工厂。我们有以下消费类：

@ConfigurationProperties
@Service
public class Listener {

    @KafkaListener(containerFactory = "firstContainerFactory", topics = "someTopic")
    public void firstListener(@Payload A first) {
        //some logic
    }

    @KafkaListener(containerFactory = "secondContainerFactory", topics = {
     //topic list       
    })
    public void secondTopicListener(@Payload B second) {
       //some logic

    }


}

所以我们在启动这个应用程序时注意到的是，它并没有一直连接到这两个主题。有时它仅连接到第二个主题或仅连接到第一个主题，并且可能连接到第一个和第二个主题（这是正确的）。那么您能帮忙了解一下这里的配置错误吗？

【问题讨论】：

“连接”是什么意思？或许你只是苦于 Kafka 组管理？
您应该为每个使用不同的group.id；否则对一个的重新平衡会影响另一个。
您的 ProducerConfig 似乎不会被 Spring 拾取，因为它不是静态的。
它实际上是被拾取的，因为它有 Config 注释。

标签： java spring-boot kafka-consumer-api spring-kafka

【解决方案1】：

将每个侦听器放在不同的group.id 中通常是最佳实践（您可以使用@KafkaListener 上的groupId 属性覆盖消费者工厂）。否则，当第二个开始时，会在第一个上引起重新平衡。当前的 2.1.x 版本是 2.1.10。

【讨论】：

那么我可以为每个消费者分配不同的组 ID 吗？或者是一些 kafka 设置？
您可以在firstConsumerConfig 和secondConsumerConfig 中设置不同的group.id（而不是在getCommonConsumerConfig() 中设置单个值），也可以在@KafkaListener 中指定它们。跨度>
明白了，这意味着他们将被分配到不同的消费者群体，对吗？
正确并且（假设主题不同）一个听众的主题/分区的重新平衡不会影响另一个听众。
所以我为他们分配了不同的组 ID，问题仍然存在。所以基本上我只重命名了其中一个，而另一个没有更改。所以没有改变的那个是行不通的。但是，如果我重命名它，那么我会收到来自该主题的所有消息（包括旧消息），这对我不利。但是，如果我在一段时间内没有更改主题 - 在重新平衡消费者后连接到主题。为什么会这样？

【解决方案2】：

好的，经过进一步调查，我能够确定我身边发生了什么样的问题。所以基本上我们有一个有多个主题的消费者组。因此，在我的情况下，每个主题都有 0 个分区（据我所知，没有分区，我们使用 1 个主题队列进行操作）。因此，当我连接到该 kafka 实例时 - 所有消费者都连接到这些主题，但是当有人也连接到该主题（可能是我的同事）时，重新平衡正在发生，他开始听其中一个主题而不是我（由于事实上，每个分区只有一个消费者可以呈现）。

【讨论】：