配置选项
本节包含 Kafka Streams 绑定器使用的配置选项。
有关binder的常见配置选项和属性，请参阅核心文档。
Kafka Streams Binder 属性
以下属性在绑定器级别可用，且必须添加前缀 spring.cloud.stream.kafka.streams.binder.。任何在 Kafka Streams 绑定器中重复使用的 Kafka 绑定器提供的属性，必须使用 spring.cloud.stream.kafka.streams.binder 作为前缀，而不是 spring.cloud.stream.kafka.binder。此规则的唯一例外是在定义 Kafka 引导服务器属性时，此时使用任一前缀均可。
配置
包含与 Apache Kafka Streams API 相关属性的键/值对映射。此属性必须以 spring.cloud.stream.kafka.streams.binder. 为前缀。以下是使用此属性的一些示例。
spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000
有关流配置中可能包含的所有属性的更多信息，请参阅 Apache Kafka Streams 文档中的 StreamsConfig JavaDocs。所有可以通过 StreamsConfig 设置的配置都可以通过此属性设置。使用此属性时，由于这是绑定器级别的属性，因此适用于整个应用程序。如果应用程序中有多个处理器，所有处理器都将获取这些属性。对于像 application.id 这样的属性，这可能会带来问题，因此您必须仔细检查 StreamsConfig 中的属性如何通过此绑定器级别的 configuration 属性进行映射。
functions.<function-bean-name>.applicationId
仅适用于函数式风格处理器。这可用于在应用程序中为每个函数设置应用程序ID。在多个函数的情况下，这是一种设置应用程序ID的便捷方式。
functions.<function-bean-name>.configuration
仅适用于函数式风格处理器。包含与Apache Kafka Streams API相关属性的键/值对映射。这类似于上文描述的绑定器级别configuration属性，但此级别的configuration属性仅针对命名函数进行限制。当您拥有多个处理器并希望根据特定函数限制配置访问时，可能需要使用此属性。所有StreamsConfig属性均可在此处使用。
代理
Broker URL
默认值：localhost
zkNodes
Zookeeper URL
默认值：localhost
deserializationExceptionHandler
反序列化错误处理器类型。此处理器在绑定器级别应用，因此适用于应用程序中的所有输入绑定。有一种方法可以在消费者绑定级别以更细粒度的方式控制它。可能的值为 - logAndContinue、logAndFail、skipAndContinue 或 sendToDlq
默认值：logAndFail
applicationId
在绑定器级别全局设置Kafka Streams应用程序的application.id的便捷方式。如果应用程序包含多个函数，则应设置不同的应用程序ID。请参阅上文详细讨论设置应用程序ID的部分。
默认值：应用程序将生成一个静态应用程序ID。有关详细信息，请参阅应用程序ID部分。
stateStoreRetry.maxAttempts
尝试连接到状态存储的最大尝试次数。
默认值：1
stateStoreRetry.backoffPeriod
在尝试重连状态存储时的退避周期。
默认值：1000 毫秒
consumerProperties
在绑定器级别设置任意消费者属性。
producerProperties
在绑定器级别设置任意生产者属性。
includeStoppedProcessorsForHealthCheck
当通过执行器停止处理器的绑定时，默认情况下该处理器将不参与健康检查。将此属性设置为 true 可为所有处理器启用健康检查，包括当前通过绑定执行器端点停止的处理器。
默认值：false
Kafka Streams 生产者属性
以下属性仅适用于 Kafka Streams 生产者，且必须以 spring.cloud.stream.kafka.streams.bindings.<binding name>.producer. 为前缀。为方便起见，如果有多个输出绑定且它们都需要一个共同的值，可以通过使用前缀 spring.cloud.stream.kafka.streams.default.producer. 进行配置。
keySerde
使用的键序列化/反序列化器
默认值：请参阅上文关于消息序列化/反序列化的讨论
valueSerde
使用的值序列化/反序列化器
默认值：请参阅上文关于消息序列化/反序列化的讨论
useNativeEncoding
启用/禁用原生编码的标志
默认值：true。
streamPartitionerBeanName
自定义出站分区器 Bean 名称，供消费者使用。应用程序可以提供自定义的 StreamPartitioner 作为 Spring Bean，并将此 Bean 的名称提供给生产者使用，以替代默认分区器。
默认值：请参阅上文关于出站分区支持的讨论。
producedAs
处理器正在向其生成数据的接收器组件的自定义名称。
默认值：none（由 Kafka Streams 生成）
Kafka Streams 消费者属性
以下属性适用于 Kafka Streams 消费者，且必须以 spring.cloud.stream.kafka.streams.bindings.<binding-name>.consumer. 为前缀。为方便起见，如果存在多个输入绑定且它们都需要相同的值，可以通过使用前缀 spring.cloud.stream.kafka.streams.default.consumer. 进行配置。
applicationId
为每个输入绑定设置 application.id。
默认值：见上文。
keySerde
使用的键序列化/反序列化器
默认值：请参阅上文关于消息序列化/反序列化的讨论
valueSerde
使用的值序列化/反序列化器
默认值：请参阅上面关于消息序列化/反序列化的讨论
materializedAs
在使用传入的 KTable 类型时进行物化的状态存储
默认值：none。
cachingDisabled
禁用物化KTable的缓存。当设置为true时，在Materialized对象上调用withCachingDisabled()。当设置为false时，在Materialized对象上调用withCachingEnabled()。
默认值：false。
loggingDisabled
禁用物化 KTable 的日志记录。当设置为 true 时，在 Materialized 对象上调用 withLoggingDisabled()。
默认值：false。
useNativeDecoding
启用/禁用原生解码的标志
默认值：true。
dlqName
DLQ 主题名称。
默认值：请参阅上文关于错误处理和死信队列的讨论。
startOffset
如果没有已提交的偏移量可供消费，则从此处开始消费。这通常用于消费者首次消费某个主题时。Kafka Streams 使用 earliest 作为默认策略，绑定器也采用相同的默认设置。可以通过此属性将其覆盖为 latest。
默认值：earliest。
注意：在消费者上使用 resetOffsets 对 Kafka Streams binder 没有任何影响。与基于消息通道的 binder 不同，Kafka Streams binder 不会按需定位到起始或结束位置。
deserializationExceptionHandler
反序列化错误处理器类型。此处理器按消费者绑定应用，与之前描述的绑定器级别属性不同。可能的值为 - logAndContinue、logAndFail、skipAndContinue 或 sendToDlq
默认值：logAndFail
timestampExtractorBeanName
消费者端使用的特定时间戳提取器 Bean 名称。应用程序可以将 TimestampExtractor 作为 Spring Bean 提供，并且可以将此 Bean 的名称提供给消费者使用，以替代默认的提取器。
默认值：请参阅上文关于时间戳提取器的讨论。
eventTypes
此绑定支持的事件类型列表，以逗号分隔。
默认值：none
eventTypeHeaderKey
通过此绑定传入的每条记录的事件类型标头键。
默认值：event_type
consumedAs
处理器所消费的源组件的自定义名称。
默认值：none（由 Kafka Streams 生成）
关于并发性的特别说明
在Kafka Streams中，您可以通过num.stream.threads属性来控制处理器可以创建的线程数量。您可以使用上述在绑定器、函数、生产者或消费者级别描述的各种configuration选项来实现这一点。您也可以使用核心Spring Cloud Stream为此目的提供的concurrency属性。使用此属性时，需要在消费者上设置。当您有多个输入绑定时，请在第一个输入绑定上设置此属性。例如，设置spring.cloud.stream.bindings.process-in-0.consumer.concurrency时，它将被绑定器转换为num.stream.threads。如果您有多个处理器，并且一个处理器定义了绑定级别的并发性，但其他处理器没有定义，那么那些没有绑定级别并发性的处理器将默认回退到通过spring.cloud.stream.kafka.streams.binder.configuration.num.stream.threads指定的绑定器范围内的属性。如果此绑定器配置不可用，则应用程序将使用Kafka Streams设置的默认值。