apache-kafka - 如何使用 spring-kafka 为监听器传递多个引导服务器

Question

我有一个监听器，需要从多个具有相同主题的 kafka 服务器读取，这些服务器都配置在一个 zookeeper 下。我如何从这些多台服务器中读取数据。你能帮忙解决这个问题吗？

我可以指向zookeeper而不是Kafka服务器吗？

Answer 1

@KafkaListener需要 KafkaListenerContainerFactory @Bean ，而后者又基于 ConsumerFactory .和 DefaultKafkaConsumerFactory接受 Map<String, Object>消费者配置:

@Configuration
@EnableKafka
public class KafkaConfig {

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

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

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, ...);
        ...
        return props;
    }
}

https://docs.spring.io/spring-kafka/docs/1.2.2.RELEASE/reference/html/_reference.html#__kafkalistener_annotation

哪里 ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG完全是标准的 Apache Kafka bootstrap.servers property :

A list of host/port pairs to use for establishing the initial connection to the Kafka cluster. The client will make use of all servers irrespective of which servers are specified here for bootstrapping—this list only impacts the initial hosts used to discover the full set of servers. This list should be in the form host1:port1,host2:port2,.... Since these servers are just used for the initial connection to discover the full cluster membership (which may change dynamically), this list need not contain the full set of servers (you may want more than one, though, in case a server is down).

不，你不能点Zookeeper地址。 Kafka 不再支持这一点。