在一个应用程序中使用多个不同的kafka集群

Question

这可能不是典型的设置，但由于更高的决策，我们最终在一个应用程序中有多个 kafka 集群，每个集群有多个主题，每个集群可能具有不同的序列化策略。Json/avro.avro可能与融合的架构注册表一起使用，或者使用单个对象编码。

好吧，我通过构建自己的抽象和注册中心，分析配置并手动创建大部分内容，以某种方式实现了它，但我觉得我需要在几个地方多次重复主题名称、模式注册url等内容，以便创建所有需要的bean。丑得要命。

我想问一下，是否有更好的方法和支持，我可能忽略了这一点。

我需要创建kafka集群的N个表示，配置一次。为给定的kafka集群配置相应的主题，为主题配置合流模式注册表（如果适用）等，以便我可以创建Avro模式文件的实例，将其发送到KafkaTemplate，它就会工作。

Answer 1

这取决于复杂性和配置的差异程度，以及这是否会有所帮助，但是您可以在@KafkaListener和每个KafkaTemplate中覆盖单个Kafka属性（例如引导服务器，反序列化程序等）。

例如

@KafkaListener(id = "two", topics = "two",
        properties = "value.deserializer:org.apache.kafka.common.serialization.ByteArrayDeserializer")
public void listen2(byte[] in) {
    System.out.println("2: " + new String(in));
}

编辑

覆盖可以外部化-如下所示：

@KafkaListener(id = "so67959209", topics = "so67959209",
        properties = "${consumer.overrides.one}")
public void listen(byte[] in) {

}

和

consumer.overrides.one=bootstrap.servers:localhost:9092\n \
 key.deserializer:org.apache.kafka.common.serialization.ByteArrayDeserializer\n \
 value.deserializer:org.apache.kafka.common.serialization.ByteArrayDeserializer

重要的是覆盖是原始的Kafka属性名称，而不是引导版本-例如bootstrap.servers与引导的引导服务器。