如何使用spring-kafka为侦听器传递多个引导程序服务器

Question

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

取而代之的是卡夫卡服务器，我可以指向zookeeper吗？

Answer 1

该@KafkaListener要求KafkaListenerContainerFactory@Bean，而这又是基于ConsumerFactory。而DefaultKafkaConsumerFactory接受消费者CONFIGS的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卡夫卡bootstrap.serversproperty：

主机/端口对用于建立名单与Kafka集群的初始连接。客户端将使用所有服务器，而不管在此指定哪些服务器用于引导 - 此列表仅影响用于发现全套服务器的初始主机。此列表应采用host1：port1，host2：port2，...的形式。由于这些服务器仅用于初始连接以发现完整集群成员资格（可能会动态更改），因此此列表不需包含完整集合的服务器（不过，如果服务器停机，您可能需要多个服务器）。

不，您不能指向Zookeeper地址。卡夫卡不再支持这一点。