配置选项
本节包含 Apache Kafka 绑定器使用的配置选项。
有关绑定器的常见配置选项和属性，请参阅核心文档中的绑定属性。
Kafka Binder 属性
spring.cloud.stream.kafka.binder.brokers
Kafka绑定器连接到的代理列表。
默认值：localhost。
spring.cloud.stream.kafka.binder.defaultBrokerPort
brokers 允许指定带或不带端口信息的主机（例如，host1,host2:port2）。当代理列表中未配置端口时，此设置将作为默认端口。
默认值：9092。
spring.cloud.stream.kafka.binder.configuration
传递给由绑定器创建的所有客户端的客户端属性（生产者和消费者）的键/值映射。由于这些属性同时被生产者和消费者使用，因此应仅限于通用属性——例如安全设置。通过此配置提供的未知Kafka生产者或消费者属性将被过滤掉，不允许传播。此处的属性会覆盖在引导程序中设置的任何属性。
默认值：空映射。
spring.cloud.stream.kafka.binder.consumerProperties
任意 Kafka 客户端消费者属性的键/值映射。除了支持已知的 Kafka 消费者属性外，此处也允许使用未知的消费者属性。此处设置的属性将覆盖在启动时以及上述 configuration 属性中设置的任何属性。
默认值：空映射。
spring.cloud.stream.kafka.binder.headers
由 binder 传输的自定义标头列表。仅在与旧版本应用程序（⇐ 1.3.x）且 kafka-clients 版本 < 0.11.0.0 进行通信时需要。新版本原生支持标头。
默认值：空。
spring.cloud.stream.kafka.binder.healthTimeout
获取分区信息的等待时间，单位为秒。如果此计时器超时，则健康报告显示为关闭。
默认值：60。
spring.cloud.stream.kafka.binder.requiredAcks
代理上所需的确认数量。请参阅 Kafka 文档中关于生产者 acks 属性的说明。
默认值：1。
spring.cloud.stream.kafka.binder.minPartitionCount
仅在设置了 autoCreateTopics 或 autoAddPartitions 时生效。绑定器在其生产和消费数据的主题上配置的全局最小分区数。该值可被生产者的 partitionCount 设置或生产者的 instanceCount * concurrency 设置值（如果任一值更大）所覆盖。
默认值：1。
spring.cloud.stream.kafka.binder.producerProperties
任意 Kafka 客户端生产者属性的键/值映射。除了支持已知的 Kafka 生产者属性外，此处也允许使用未知的生产者属性。此处设置的属性将覆盖在启动时及上述 configuration 属性中设置的任何属性。
默认值：空映射。
spring.cloud.stream.kafka.binder.replicationFactor
如果启用了 autoCreateTopics，则自动创建主题的复制因子。可以在每个绑定上进行覆盖。
如果您使用的是 2.4 版本之前的 Kafka 代理，那么此值应至少设置为 1。从 3.0.8 版本开始，绑定器使用 -1 作为默认值，这表示将使用代理的 default.replication.factor 属性来确定副本数量。请咨询您的 Kafka 代理管理员，确认是否有要求最小复制因子的策略。如果有，通常 default.replication.factor 会与该值匹配，此时应使用 -1，除非您需要的复制因子大于该最小值。
默认值：-1。
spring.cloud.stream.kafka.binder.autoCreateTopics
如果设置为 true，绑定器会自动创建新主题。如果设置为 false，绑定器将依赖已配置的主题。在后一种情况下，如果主题不存在，绑定器将无法启动。
此设置独立于代理的 auto.create.topics.enable 设置，且不会对其产生影响。如果服务器设置为自动创建主题，它们可能会作为元数据检索请求的一部分被创建，并使用代理的默认设置。
默认值：true。
spring.cloud.stream.kafka.binder.autoAddPartitions
如果设置为 true，binder 会在需要时创建新的分区。如果设置为 false，binder 将依赖已配置的主题分区大小。如果目标主题的分区数量小于预期值，binder 将无法启动。
默认值：false。
spring.cloud.stream.kafka.binder.transaction.transactionIdPrefix
在绑定器中启用事务。请参阅Kafka文档中的 transaction.id 以及 spring-kafka 文档中的事务。启用事务后，单个 producer 属性将被忽略，所有生产者都将使用 spring.cloud.stream.kafka.binder.transaction.producer.* 属性。
默认值为 null（无事务）
spring.cloud.stream.kafka.binder.transaction.producer.*
事务绑定器中生产者的全局生产者属性。请参阅 spring.cloud.stream.kafka.binder.transaction.transactionIdPrefix 和 Kafka 生产者属性 以及所有绑定器支持的通用生产者属性。
默认值：请参阅各个生产者属性。
spring.cloud.stream.kafka.binder.headerMapperBeanName
用于将 spring-messaging 消息头与 Kafka 消息头相互映射的 KafkaHeaderMapper 的 bean 名称。例如，如果您希望在使用 JSON 反序列化处理消息头的 BinderHeaderMapper bean 中自定义受信任的包，可以使用此属性。如果未通过此属性使此自定义 BinderHeaderMapper bean 对 binder 可用，则 binder 将查找名为 kafkaBinderHeaderMapper 且类型为 BinderHeaderMapper 的消息头映射器 bean，然后才会回退到由 binder 创建的默认 BinderHeaderMapper。
默认值：无。
spring.cloud.stream.kafka.binder.considerDownWhenAnyPartitionHasNoLeader
设置标志，当发现主题上的任何分区（无论哪个消费者正在从中接收数据）没有领导者时，将绑定器健康状态标记为 down。
默认值：true。
spring.cloud.stream.kafka.binder.certificateStoreDirectory
当 truststore 或 keystore 证书位置被指定为非本地文件系统资源（由 org.springframework.core.io.Resource 支持，例如 CLASSPATH、HTTP 等）时，绑定器会将该资源从路径（可转换为 org.springframework.core.io.Resource）复制到文件系统上的某个位置。这适用于代理级别的证书（ssl.truststore.location 和 ssl.keystore.location）以及用于 schema registry 的证书（schema.registry.ssl.truststore.location 和 schema.registry.ssl.keystore.location）。请注意，truststore 和 keystore 位置路径必须在 spring.cloud.stream.kafka.binder.configuration… 下提供。例如，spring.cloud.stream.kafka.binder.configuration.ssl.truststore.location、spring.cloud.stream.kafka.binder.configuration.schema.registry.ssl.truststore.location 等。文件将被复制到此属性值指定的位置，该位置必须是文件系统上运行应用程序的进程可写的现有目录。如果未设置此值且证书文件是非本地文件系统资源，则它将被复制到由 System.getProperty("java.io.tmpdir") 返回的系统临时目录。如果设置了此值，但在文件系统上找不到该目录或该目录不可写，同样适用此规则。
默认值：无。
spring.cloud.stream.kafka.binder.metrics.defaultOffsetLagMetricsEnabled
当设置为 true 时，每次访问指标时都会计算每个消费者主题的偏移量滞后指标。当设置为 false 时，仅使用定期计算的偏移量滞后。
默认值：true
spring.cloud.stream.kafka.binder.metrics.offsetLagMetricsInterval
计算每个消费者主题偏移量滞后的时间间隔。当 metrics.defaultOffsetLagMetricsEnabled 被禁用或其计算耗时过长时，将使用此值。
默认值：60 秒
spring.cloud.stream.kafka.binder.enableObservation
在此绑定器的所有绑定上启用 Micrometer 观测注册表。
默认值：false
spring.cloud.stream.kafka.binder.healthIndicatorConsumerGroup
KafkaHealthIndicator 元数据消费者 group.id。该消费者由 HealthIndicator 用于查询正在使用的主题的元数据。
默认值：无。
Kafka Consumer 属性
以下属性仅适用于 Kafka 消费者，且必须以 spring.cloud.stream.kafka.bindings.<channelName>.consumer. 为前缀。
为避免重复配置，Spring Cloud Stream 支持为所有通道设置属性值，格式为 spring.cloud.stream.kafka.default.consumer.<property>=<value>。
admin.configuration
自 2.1.1 版本起，此属性已被弃用，建议使用 topic.properties，未来版本将移除对该属性的支持。
admin.replicas-assignment
自 2.1.1 版本起，此属性已被弃用，建议使用 topic.replicas-assignment 替代，未来版本将移除对该属性的支持。
admin.replication-factor
自 2.1.1 版本起，此属性已被弃用，推荐使用 topic.replication-factor，未来版本将移除对其的支持。
autoRebalanceEnabled
当设置为 true 时，主题分区会在消费者组的成员之间自动重新平衡。当设置为 false 时，每个消费者会根据 spring.cloud.stream.instanceCount 和 spring.cloud.stream.instanceIndex 被分配一组固定的分区。这要求在每个启动的实例上正确设置 spring.cloud.stream.instanceCount 和 spring.cloud.stream.instanceIndex 属性。在这种情况下，spring.cloud.stream.instanceCount 属性的值通常必须大于 1。
默认值：true。
ackEachRecord
当 autoCommitOffset 为 true 时，此设置决定是否在每条记录处理完成后提交偏移量。默认情况下，偏移量会在 consumer.poll() 返回的一批记录中的所有记录处理完成后提交。每次轮询返回的记录数量可通过 Kafka 属性 max.poll.records 控制，该属性通过消费者 configuration 属性设置。将此设置为 true 可能会导致性能下降，但这样做可以减少故障发生时记录被重新投递的可能性。另请参阅绑定器的 requiredAcks 属性，该属性也会影响提交偏移量的性能。此属性自 3.1 版本起已弃用，建议使用 ackMode。如果未设置 ackMode 且未启用批处理模式，则将使用 RECORD ackMode。
默认值：false。
autoCommitOffset
从版本 3.1 开始，此属性已弃用。有关替代方案的更多详细信息，请参阅 ackMode。该属性用于确定在处理消息时是否自动提交偏移量。如果设置为 false，入站消息中将包含一个键为 kafka_acknowledgment、类型为 org.springframework.kafka.support.Acknowledgment 的标头。应用程序可以使用此标头来确认消息。有关详细信息，请参阅示例部分。当此属性设置为 false 时，Kafka 绑定器会将确认模式设置为 org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode.MANUAL，应用程序负责确认记录。另请参阅 ackEachRecord。
默认值：true。
ackMode
指定容器的确认模式。此模式基于 Spring Kafka 中定义的 AckMode 枚举。如果 ackEachRecord 属性设置为 true 且消费者未处于批处理模式，则将使用 RECORD 确认模式；否则，将使用通过此属性提供的确认模式。
autoCommitOnError
在可轮询消费者中，如果设置为 true，则始终在出错时自动提交。如果未设置（默认值）或设置为 false，则不会在可轮询消费者中自动提交。请注意，此属性仅适用于可轮询消费者。
默认值：未设置。
resetOffsets
默认值：false。
startOffset
新消费者组的起始偏移量。允许的值：earliest 和 latest。如果为消费者 'binding' 显式设置了消费者组（通过 spring.cloud.stream.bindings.<channelName>.group），则 'startOffset' 设置为 earliest。否则，对于 anonymous 消费者组，它设置为 latest。有关此属性的更多信息，请参阅 reset-offsets。
默认值：null（等同于 earliest）。
enableDlq
当设置为 true 时，将为消费者启用 DLQ 行为。默认情况下，导致错误的消息会被转发到名为 error.<destination>.<group> 的主题。DLQ 主题名称可以通过设置 dlqName 属性或定义类型为 DlqDestinationResolver 的 @Bean 来配置。当错误数量相对较少且重新处理整个原始主题可能过于繁琐时，这为更常见的 Kafka 重放场景提供了一个替代选项。更多信息请参阅 kafka dlq 处理。从 2.0 版本开始，发送到 DLQ 主题的消息会通过以下标头进行增强：x-original-topic、x-exception-message 和 x-exception-stacktrace（作为 byte[]）。默认情况下，失败记录会被发送到 DLQ 主题中与原始记录相同的分区号。有关如何更改此行为，请参阅 dlq 分区选择。当 destinationIsPattern 为 true 时不允许使用。
默认值：false。
dlqPartitions
当 enableDlq 为 true 且未设置此属性时，将创建一个与主主题分区数相同的死信主题。通常，死信记录会被发送到死信主题中与原记录相同的分区。此行为可以更改；请参阅 死信分区选择。如果此属性设置为 1 且没有 DqlPartitionFunction bean，所有死信记录将被写入分区 0。如果此属性大于 1，则必须提供一个 DlqPartitionFunction bean。请注意，实际分区数受绑定器的 minPartitionCount 属性影响。
默认值：none
配置
包含通用Kafka消费者属性的键/值对映射。除了Kafka消费者属性外，其他配置属性也可以在此处传递。例如应用程序所需的一些属性，如 spring.cloud.stream.kafka.bindings.input.consumer.configuration.foo=bar。bootstrap.servers 属性不能在此处设置；如果需要连接到多个集群，请使用多绑定器支持。
默认值：空映射。
dlqName
接收错误消息的DLQ主题名称。
默认值：null（如果未指定，导致错误的消息将被转发到名为 error.<destination>.<group> 的主题）。
dlqProducerProperties
通过此配置，可以设置DLQ专用的生产者属性。所有可通过Kafka生产者属性设置的属性均可通过此属性进行配置。当消费者启用原生解码时（即设置 useNativeDecoding: true），应用程序必须为DLQ提供相应的键/值序列化器。这需要通过 dlqProducerProperties.configuration.key.serializer 和 dlqProducerProperties.configuration.value.serializer 的形式进行配置。
Default: 默认的 Kafka 生产者属性。
standardHeaders
指示入站通道适配器填充了哪些标准头部。允许的值：none、id、timestamp 或 both。在使用原生反序列化且第一个接收消息的组件需要 id（例如配置为使用 JDBC 消息存储的聚合器）时非常有用。
默认值：none
converterBeanName
一个实现了 RecordMessageConverter 的 bean 的名称。用于入站通道适配器中，以替换默认的 MessagingMessageConverter。
默认值：null
idleEventInterval
表示最近没有收到消息的事件之间的间隔时间，以毫秒为单位。使用 ApplicationListener<ListenerContainerIdleEvent> 来接收这些事件。有关使用示例，请参阅 pause-resume。
默认值：30000
destinationIsPattern
当设置为 true 时，目标被视为一个正则表达式 Pattern，用于由代理匹配主题名称。当设置为 true 时，不会预配主题，并且不允许设置 enableDlq，因为绑定器在预配阶段不知道主题名称。请注意，检测匹配模式的新主题所需的时间由消费者属性 metadata.max.age.ms 控制，该属性（在撰写本文时）默认为 300,000 毫秒（5 分钟）。这可以通过上面的 configuration 属性进行配置。
默认值：false
topic.properties
一个用于配置新主题时使用的Kafka主题属性Map——例如，spring.cloud.stream.kafka.bindings.input.consumer.topic.properties.message.format.version=0.9.0.0
默认值：无。
topic.replicas-assignment
一个副本分配的 Map<Integer, List<Integer>>，其中键为分区，值为分配方案。在配置新主题时使用。请参阅 kafka-clients jar 中的 NewTopic Javadocs。
默认值：无。
topic.replication-factor
创建主题时使用的复制因子。覆盖绑定器范围的设置。如果存在 replicas-assignments，则忽略此设置。
默认值：无（使用绑定器范围的默认值 -1）。
pollTimeout
轮询消费者中用于轮询的超时时间。
默认值：5 秒。
transactionManager
用于覆盖此绑定的绑定器事务管理器的 KafkaAwareTransactionManager 的 Bean 名称。通常，如果您想使用 ChainedKafkaTransactionManager 将另一个事务与 Kafka 事务同步，则需要此配置。为了实现记录的恰好一次消费和生产，消费者和生产者绑定都必须配置相同的事务管理器。
默认值：无。
txCommitRecovered
在使用事务性绑定器时，默认情况下，已恢复记录（例如，当重试次数耗尽且记录被发送到死信主题时）的偏移量将通过新事务提交。将此属性设置为 false 可禁止提交已恢复记录的偏移量。
默认值：true。
commonErrorHandlerBeanName
每个消费者绑定所使用的 CommonErrorHandler bean 名称。当存在时，用户提供的这个 CommonErrorHandler 将优先于绑定器定义的任何其他错误处理器。如果应用程序不想使用 ListenerContainerCustomizer，然后检查目标/组组合来设置错误处理器，这是一种便捷的表达错误处理器的方式。
默认值：无。
Kafka 生产者属性
以下属性仅适用于 Kafka 生产者，并且必须以 spring.cloud.stream.kafka.bindings.<channelName>.producer. 为前缀。
为避免重复，Spring Cloud Stream 支持为所有通道设置值，格式为 spring.cloud.stream.kafka.default.producer.<property>=<value>。
admin.configuration
自 2.1.1 版本起，此属性已被弃用，建议使用 topic.properties，未来版本将移除对该属性的支持。
admin.replicas-assignment
自 2.1.1 版本起，此属性已被弃用，推荐使用 topic.replicas-assignment，未来版本将移除对其的支持。
admin.replication-factor
自 2.1.1 版本起，此属性已被弃用，推荐使用 topic.replication-factor，未来版本将移除对其的支持。
bufferSize
Kafka 生产者尝试在发送前批量处理的数据上限（以字节为单位）。
默认值：16384。
sync
生产者是否为同步。
默认值：false。
sendTimeoutExpression
一个针对出站消息进行求值的SpEL表达式，用于在启用同步发布时评估等待确认的超时时间——例如，headers['mySendTimeout']。超时值的单位为毫秒。在3.0之前的版本中，除非使用原生编码，否则无法使用负载（payload），因为当此表达式被求值时，负载已经转换为byte[]形式。现在，表达式会在负载转换之前进行求值。
默认值：none。
batchTimeout
生产者等待多长时间以允许更多消息在同一个批次中累积，然后再发送消息。（通常，生产者根本不会等待，而是直接发送在前一次发送进行期间累积的所有消息。）非零值可能会以延迟为代价提高吞吐量。
默认值：0。
messageKeyExpression
一个针对出站消息求值的SpEL表达式，用于填充所生成Kafka消息的键——例如 headers['myKey']。在3.0之前的版本中，除非使用原生编码，否则无法使用负载数据，因为在此表达式求值时，负载数据已转换为 byte[] 形式。现在，该表达式会在负载数据转换之前进行求值。对于常规处理器（Function<String, String> 或 Function<Message<?>, Message<?>>），如果生成的键需要与来自主题的传入键相同，可以按如下方式设置此属性：spring.cloud.stream.kafka.bindings.<output-binding-name>.producer.messageKeyExpression: headers['kafka_receivedMessageKey']。对于响应式函数，有一个重要的注意事项需要牢记。在这种情况下，应用程序需要手动将传入消息的标头复制到出站消息。您可以设置标头，例如 myKey，并如上所述使用 headers['myKey']，或者为了方便起见，只需设置 KafkaHeaders.MESSAGE_KEY 标头，这样您就完全不需要设置此属性。
默认值：none。
headerPatterns
一个以逗号分隔的简单模式列表，用于匹配需要映射到 ProducerRecord 中 Kafka Headers 的 Spring 消息头。模式可以以通配符（星号）开头或结尾。通过在模式前添加 ! 可以表示否定匹配。匹配在首次匹配（肯定或否定）后停止。例如，!ask,as* 会传递 ash 但不会传递 ask。id 和 timestamp 永远不会被映射。
默认值：*（所有标题 - 除了 id 和 timestamp）
配置
包含通用 Kafka 生产者属性的键/值对映射。bootstrap.servers 属性不能在此处设置；如果需要连接到多个集群，请使用多绑定器支持。
默认值：空映射。
topic.properties
一个用于配置新主题时使用的Kafka主题属性Map——例如，spring.cloud.stream.kafka.bindings.output.producer.topic.properties.message.format.version=0.9.0.0
topic.replicas-assignment
一个 Map<Integer, List<Integer>> 类型的副本分配映射，其中键为分区，值为分配方案。在配置新主题时使用。请参阅 kafka-clients jar 包中的 NewTopic Javadocs。
默认值：无。
topic.replication-factor
创建主题时使用的复制因子。覆盖绑定器范围的设置。如果存在 replicas-assignments，则忽略此设置。
默认值：无（使用绑定器范围的默认值 -1）。
useTopicHeader
设置为 true 以使用出站消息中 KafkaHeaders.TOPIC 消息头的值覆盖默认绑定目标（主题名称）。如果该消息头不存在，则使用默认绑定目标。
默认值：false。
recordMetadataChannel
用于发送成功发送结果的 MessageChannel 的 bean 名称；该 bean 必须存在于应用程序上下文中。发送到通道的消息是已发送的消息（经过转换后，如果有的话），并带有一个额外的头部 KafkaHeaders.RECORD_METADATA。该头部包含由 Kafka 客户端提供的 RecordMetadata 对象；它包含了记录写入主题的分区和偏移量信息。
ResultMetadata meta = sendResultMsg.getHeaders().get(KafkaHeaders.RECORD_METADATA, RecordMetadata.class)
发送失败的消息会进入生产者错误通道（如果已配置）；请参阅 Kafka 错误通道。
默认值：null。
Kafka绑定器使用生产者的partitionCount设置作为提示，以创建具有指定分区数量的主题（结合minPartitionCount，两者中的较大值将被使用）。在同时为绑定器配置minPartitionCount和为应用程序配置partitionCount时需谨慎，因为会使用较大的值。如果主题已存在且分区数量较少，并且autoAddPartitions被禁用（默认情况），绑定器将无法启动。如果主题已存在且分区数量较少，并且autoAddPartitions被启用，则会添加新的分区。如果主题已存在的分区数量大于（minPartitionCount或partitionCount）的最大值，则使用现有的分区数量。
压缩
设置 compression.type 生产者属性。支持的值为 none、gzip、snappy、lz4 和 zstd。如果按照 Spring for Apache Kafka 文档 中的讨论，将 kafka-clients jar 覆盖到 2.1.0（或更高版本），并希望使用 zstd 压缩，请使用 spring.cloud.stream.kafka.bindings.<binding-name>.producer.configuration.compression.type=zstd。
默认值：none。
transactionManager
用于覆盖此绑定的绑定器事务管理器的 KafkaAwareTransactionManager 的 Bean 名称。通常，如果您想使用 ChainedKafkaTransactionManager 将另一个事务与 Kafka 事务同步，则需要此配置。为了实现记录的恰好一次消费和生产，消费者和生产者绑定都必须配置相同的事务管理器。
默认值：无。
closeTimeout
关闭生产者时等待的超时时间（以秒为单位）。
默认值：30
allowNonTransactional
通常，如果当前没有正在进行的事务，与事务性绑定器关联的所有输出绑定都会在新事务中发布。此属性允许您覆盖该行为。如果设置为 true，发布到此输出绑件的记录将不会在事务中运行，除非已有一个事务正在进行中。
默认值：false