我们可以在一个应用程序属性文件中有两个不同的spring-kafka使用者属性吗

Question

Consumer1配置详细信息(用于批处理(

spring.kafka.consumer.[0].bootstrap-servers = ${bootstrap.servers1}
spring.kafka.consumer.[0].enable-auto-commit = false
spring.kafka.consumer.[0].auto-offset-reset = latest
spring.kafka.consumer.[0].max-poll-records = 100
spring.kafka.consumer.[0].key-deserializer= org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.[0].value-deserializer= org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.[0].properties.allow.auto.create.topics = false
spring.kafka.consumer.[0].properties.max.poll.interval.ms = 300000
spring.kafka.listener.[0].ack-mode = MANUAL
spring.kafka.listener.[0].concurrency = 1

Consumer2配置详细信息(针对单个记录(

spring.kafka.consumer.[1].bootstrap-servers = ${bootstrap.servers2}
spring.kafka.consumer.[1].enable-auto-commit = false
spring.kafka.consumer.[1].auto-offset-reset = latest
spring.kafka.consumer.[1].key-deserializer= org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.[1].value-deserializer= org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.[1].properties.allow.auto.create.topics = false
spring.kafka.consumer.[1].properties.max.poll.interval.ms = 300000
spring.kafka.listener.[1].ack-mode = MANUAL
spring.kafka.listener.[1].concurrency = 1

Answer 1

否；要配置多组基础设施，您需要手动定义bean，而不是使用自动配置。

但是，您可以直接覆盖@KafkaListener上的各个属性：

/**
* Kafka consumer properties; they will supersede any properties with the same name
* defined in the consumer factory (if the consumer factory supports property overrides).
* <p>
* <b>Supported Syntax</b>
* <p>The supported syntax for key-value pairs is the same as the
* syntax defined for entries in a Java
* {@linkplain java.util.Properties#load(java.io.Reader) properties file}:
* <ul>
* <li>{@code key=value}</li>
* <li>{@code key:value}</li>
* <li>{@code key value}</li>
* </ul>
* {@code group.id} and {@code client.id} are ignored.
* @return the properties.
* @since 2.2.4
* @see org.apache.kafka.clients.consumer.ConsumerConfig
* @see #groupId()
* @see #clientIdPrefix()
*/
String[] properties() default {};

例如@KafakListener( ..., properties = "bootstrap.servers=${bootstrap.servers2}")