在 Spring 中消费 Kafka 消息

Question

通过关注tutorial，我能够创建一个简单的生产者-消费者示例。在我的示例中，只有 1 个主题，而我正在听那个主题。因此，ReceiverConfig 中的代码是有意义的。特别是围绕GROUP_ID_CONFIG 的点，即我创建主题topic_name，然后在此配置中对其进行配置。现在我的问题是，如果我有超过 1 个主题怎么办。假设我有topic_1、topic_2 等等？我应该为每个单独的主题创建ReceiverConfig 吗？

@EnableKafka
@Configuration
public class ReceiverConfig {
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(GROUP_ID_CONFIG, "topic_name");
        props.put(AUTO_OFFSET_RESET_CONFIG, "earliest");

        return props;
    }

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

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

Answer 1

简短的回答是否定的，您不需要为每个主题创建多个配置。

在继续之前，我认为最好指定 groupId 是 Consumer 进程所属的组，而 topic 将被 Consumer 进程消费是两个不同的东西。

通过下面的句子，您将告诉消费者它属于 topic_name 组，仅此而已。

props.put(GROUP_ID_CONFIG, "topic_name");

如果您希望消费者从多个主题读取数据，有一个subscribe 方法接收一个集合作为参数，这样您就可以指定所有主题来读取数据，而无需为每个主题创建新配置话题。

请查看这个例子，你会看到我提到的方法

// Subscribe to the topic.
consumer.subscribe(Collections.singletonList(TOPIC));