以下属性仅适用于 Kafka 消费者，必须以spring.cloud.stream.kafka.bindings.<channelName>.consumer.为前缀。

当应用程序启动时，每个分配的分区的初始位置取决于两个属性startOffset 和 resetOffsets。如果resetOffsets 为false，则应用正常的 Kafka 消费者 auto.offset.reset 语义。即，如果绑定消费者的组没有为分区提交偏移量，则位置为earliest 或 latest。默认情况下，具有显式group 的绑定使用earliest，而匿名绑定（没有group）使用latest。这些默认值可以通过设置startOffset 绑定属性来覆盖。第一次使用特定group 启动绑定时，将没有提交的偏移量。另一个没有提交偏移量的情况是偏移量已过期。对于现代代理（从 2.1 开始）和默认代理属性，偏移量在最后一个成员离开组后 7 天过期。有关更多信息，请参阅 offsets.retention.minutes 代理属性。

当resetOffsets 为true 时，绑定会应用与代理上没有提交偏移量时类似的语义，就好像此绑定从未从主题中消费过一样；即，任何当前提交的偏移量都会被忽略。

以下是可能使用此功能的两种用例。

要更细致地控制主题偏移量，请参阅 使用 KafkaRebalanceListener；当提供监听器时，resetOffsets 不应设置为 true，否则会导致错误。 >>>>>>> 7bc90c10…​ GH-1084: 添加 txCommitRecovered 属性

从 3.0 版本开始，当 spring.cloud.stream.binding.<name>.consumer.batch-mode 设置为 true 时，从 Kafka Consumer 轮询接收到的所有记录将作为 List<?> 提供给监听器方法。否则，该方法将一次调用一个记录。批次的大小由 Kafka 消费者属性 max.poll.records、fetch.min.bytes、fetch.max.wait.ms 控制；有关更多信息，请参阅 Kafka 文档。

请注意，批处理模式不支持 @StreamListener - 它只适用于较新的函数式编程模型。

以下属性仅适用于 Kafka 生产者，并且必须以 spring.cloud.stream.kafka.bindings.<channelName>.producer. 为前缀。

ResultMetadata meta = sendResultMsg.getHeaders().get(KafkaHeaders.RECORD_METADATA, RecordMetadata.class)

发送失败将进入生产者错误通道（如果已配置）；请参阅 错误通道。默认值：null

+

在本节中，我们将展示在特定场景中使用上述属性的方法。

默认情况下，记录使用与原始记录相同的分区发布到死信主题。这意味着死信主题必须至少与原始记录具有相同数量的分区。

要更改此行为，请将 DlqPartitionFunction 实现作为 @Bean 添加到应用程序上下文。只能存在一个这样的 Bean。该函数将提供消费者组、失败的 ConsumerRecord 和异常。例如，如果您始终想要路由到分区 0，您可以使用

还可以为 DLQ 主题定义自定义名称。为此，请将 DlqDestinationResolver 的实现作为 @Bean 创建到应用程序上下文。当绑定器检测到这样的 Bean 时，它将优先使用它，否则它将使用 dlqName 属性。如果两者都找不到，它将默认为 error.<destination>.<group>。以下是如何将 DlqDestinationResolver 作为 @Bean 的示例。

在为 DlqDestinationResolver 提供实现时，需要注意的一点是，绑定器中的供应器不会自动为应用程序创建主题。这是因为绑定器无法推断实现可能发送到的所有 DLQ 主题的名称。因此，如果您使用此策略提供 DLQ 名称，则应用程序有责任确保这些主题事先创建。

由于框架无法预测用户希望如何处理死信消息，因此它不提供任何标准机制来处理它们。如果死信的原因是暂时的，您可能希望将消息路由回原始主题。但是，如果问题是永久性问题，则会导致无限循环。本主题中的示例 Spring Boot 应用程序展示了如何将这些消息路由回原始主题，但它会在三次尝试后将它们移动到“停车场”主题。该应用程序是另一个 spring-cloud-stream 应用程序，它从死信主题读取。当 5 秒内没有收到消息时，它将终止。

这些示例假设原始目标是 so8400out，消费者组是 so8400。

有几种策略需要考虑

以下代码清单显示了示例应用程序

从 Spring Cloud Stream 3.0.0 开始，Kafka Streams 绑定器允许应用程序使用 Java 8 中可用的函数式编程风格进行设计和开发。这意味着应用程序可以简洁地表示为类型 java.util.function.Function 或 java.util.function.Consumer 的 lambda 表达式。

让我们来看一个非常基本的例子。

虽然简单，但这是一个完整的独立 Spring Boot 应用程序，它利用 Kafka Streams 进行流处理。这是一个没有出站绑定的消费者应用程序，只有一个入站绑定。应用程序消费数据，并简单地将 KStream 键和值的信息记录到标准输出。应用程序包含 SpringBootApplication 注解和一个标记为 Bean 的方法。bean 方法的类型为 java.util.function.Consumer，它用 KStream 参数化。然后在实现中，我们返回一个本质上是 lambda 表达式的 Consumer 对象。在 lambda 表达式中，提供了处理数据的代码。

在这个应用程序中，有一个单一的输入绑定，类型为 KStream。绑定器为应用程序创建此绑定，名称为 process-in-0，即函数 bean 名称后跟一个连字符 (-)，然后是文字 in，再跟一个连字符，然后是参数的序数位置。您可以使用此绑定名称来设置其他属性，例如目标。例如，spring.cloud.stream.bindings.process-in-0.destination=my-topic。

构建为 uber-jar（例如，kstream-consumer-app.jar）后，您可以像下面这样运行上面的示例。

以下是一个完整的处理器示例，它包含输入和输出绑定。这是经典的词频统计示例，其中应用程序从主题接收数据，然后在滚动时间窗口中计算每个词的出现次数。

同样，这是一个完整的 Spring Boot 应用程序。与第一个应用程序的不同之处在于，bean 方法的类型为 java.util.function.Function。Function 的第一个参数化类型用于输入 KStream，第二个参数化类型用于输出。在方法体中，提供了一个 Function 类型的 lambda 表达式，作为实现，给出了实际的业务逻辑。与之前讨论的基于 Consumer 的应用程序类似，这里的输入绑定默认命名为 process-in-0。对于输出，绑定名称也会自动设置为 process-out-0。

构建为 uber-jar（例如，wordcount-processor.jar）后，您可以像下面这样运行上面的示例。

此应用程序将从 Kafka 主题 words 中消费消息，并将计算结果发布到输出主题 counts。

Spring Cloud Stream 将确保来自传入和传出主题的消息自动绑定为 KStream 对象。作为开发人员，您可以专注于代码的业务方面，即编写处理器中所需的逻辑。框架会自动处理 Kafka Streams 基础设施所需的 Kafka Streams 特定配置。

我们上面看到的两个示例都有一个 KStream 输入绑定。在这两种情况下，绑定都从单个主题接收记录。如果您想将多个主题多路复用到单个 KStream 绑定中，可以在下面提供以逗号分隔的 Kafka 主题作为目标。

spring.cloud.stream.bindings.process-in-0.destination=topic-1,topic-2,topic-3

此外，如果您想根据正则表达式匹配主题，还可以提供主题模式作为目标。

spring.cloud.stream.bindings.process-in-0.destination=input.*

虽然上面概述的函数式编程模型是首选方法，但如果您愿意，您仍然可以使用经典的 StreamListener 方法。

以下是一些示例。

以下是使用 StreamListener 的单词计数示例的等效代码。

如您所见，这有点冗长，因为您需要提供 EnableBinding 和其他额外的注释（如 StreamListener 和 SendTo）才能使其成为一个完整的应用程序。EnableBinding 是您指定包含绑定的绑定接口的地方。在本例中，我们使用的是库存 KafkaStreamsProcessor 绑定接口，它具有以下契约。

绑定器将为输入 KStream 和输出 KStream 创建绑定，因为您使用的是包含这些声明的绑定接口。

除了函数式风格提供的编程模型的明显差异之外，这里需要特别提到的一点是，绑定名称是您在绑定接口中指定的。例如，在上面的应用程序中，由于我们使用的是 KafkaStreamsProcessor，因此绑定名称为 input 和 output。绑定属性需要使用这些名称。例如 spring.cloud.stream.bindings.input.destination、spring.cloud.stream.bindings.output.destination 等。请记住，这与函数式风格有根本的不同，因为在函数式风格中，绑定器会为应用程序生成绑定名称。这是因为应用程序在使用 EnableBinding 的函数模型中没有提供任何绑定接口。

以下是一个具有两个输入的接收器的另一个示例。

以下是与上面看到的基于 BiFunction 的处理器相同的 StreamListener 等效代码。

最后，以下是具有三个输入和柯里化函数的应用程序的 StreamListener 等效代码。

您可能会注意到上面两个示例更加冗长，因为除了提供EnableBinding之外，您还需要编写自己的自定义绑定接口。使用函数模型，您可以避免所有这些仪式细节。

在我们继续研究 Kafka Streams 绑定器提供的通用编程模型之前，以下是多个输出绑定的StreamListener版本。

回顾一下，我们已经回顾了使用 Kafka Streams 绑定器时的各种编程模型选择。

绑定器为输入上的KStream、KTable和GlobalKTable提供绑定功能。KTable和GlobalKTable绑定仅在输入上可用。绑定器支持KStream的输入和输出绑定。

Kafka Streams 绑定器编程模型的优点是，绑定器为您提供了使用完全函数式编程模型或使用基于StreamListener的命令式方法的灵活性。

绑定器允许在单个 Spring Cloud Stream 应用程序中拥有多个 Kafka Streams 处理器。您可以拥有以下应用程序。

在这种情况下，绑定器将创建 3 个具有不同应用程序 ID 的独立 Kafka Streams 对象（下面将详细介绍）。但是，如果您在应用程序中有多个处理器，则必须告诉 Spring Cloud Stream 哪些函数需要激活。以下是如何激活函数。

spring.cloud.stream.function.definition: process;anotherProcess;yetAnotherProcess

如果您希望某些函数不要立即激活，可以从该列表中删除它们。

当您只有一个 Kafka Streams 处理器和其他类型的Function bean 在同一个应用程序中时，这也适用，该应用程序通过不同的绑定器处理（例如，基于常规 Kafka 消息通道绑定器的函数 bean）。

应用程序 ID 是您需要为 Kafka Streams 应用程序提供的必填属性。Spring Cloud Stream Kafka Streams 绑定器允许您通过多种方式配置此应用程序 ID。

如果您在应用程序中只有一个处理器或StreamListener，那么您可以使用以下属性在绑定器级别设置它。

spring.cloud.stream.kafka.streams.binder.applicationId.

为了方便起见，如果您只有一个处理器，您也可以使用spring.application.name作为属性来委托应用程序 ID。

如果您在应用程序中有多个 Kafka Streams 处理器，那么您需要为每个处理器设置应用程序 ID。在函数模型的情况下，您可以将其作为属性附加到每个函数。

例如，假设您有以下函数。

和

然后，您可以使用以下绑定器级别属性为每个应用程序设置应用程序 ID。

spring.cloud.stream.kafka.streams.binder.functions.process.applicationId

和

spring.cloud.stream.kafka.streams.binder.functions.anotherProcess.applicationId

在 StreamListener 的情况下，您需要在处理器的第一个输入绑定上设置此属性。

例如，假设您有两个基于 StreamListener 的处理器。

然后，您必须使用以下绑定属性为此设置应用程序 ID。

spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId

和

spring.cloud.stream.kafka.streams.bindings.anotherInput.consumer.applicationId

对于基于函数的模型，这种在绑定级别设置应用程序 ID 的方法也适用。但是，如果您使用的是函数模型，则在绑定器级别按函数设置应用程序 ID 会更容易，如上所述。

对于生产部署，强烈建议通过配置显式指定应用程序 ID。如果您要自动扩展应用程序，这一点尤其重要，在这种情况下，您需要确保每个实例都使用相同的应用程序 ID 部署。

如果应用程序未提供应用程序 ID，则绑定器将为您自动生成一个静态应用程序 ID。这在开发场景中很方便，因为它避免了显式提供应用程序 ID 的需要。以这种方式生成的应用程序 ID 在应用程序重启时将保持静态。在函数模型中，生成的应用程序 ID 将是函数 bean 名称后跟文字 applicationID，例如，如果 process 是函数 bean 名称，则为 process-applicationID。在 StreamListener 的情况下，生成的应用程序 ID 将使用包含类名后跟方法名后跟文字 applicationId，而不是使用函数 bean 名称。

默认情况下，绑定器在使用函数式风格时，会使用上面讨论的策略来生成绑定名称，即 <function-bean-name>-<in>|<out>-[0..n]，例如 process-in-0、process-out-0 等。如果您想覆盖这些绑定名称，可以通过指定以下属性来实现。

spring.cloud.stream.function.bindings.<default binding name>。默认绑定名称是绑定器生成的原始绑定名称。

例如，假设您有以下函数。

绑定器将生成名称为 process-in-0、process-in-1 和 process-out-0 的绑定。现在，如果您想将它们更改为完全不同的名称，也许是更具领域特色的绑定名称，那么您可以按照以下步骤进行操作。

spring.cloud.stream.function.bindings.process-in-0=users

spring.cloud.stream.function.bindings.process-in-0=regions

和

spring.cloud.stream.function.bindings.process-out-0=clicks

之后，您必须在这些新的绑定名称上设置所有绑定级别的属性。

请记住，在上面描述的函数式编程模型中，在大多数情况下，遵循默认绑定名称是有意义的。您可能仍然想要进行这种覆盖的唯一原因是，当您有大量配置属性时，您希望将绑定映射到更具领域友好的名称。

在运行 Kafka Streams 应用程序时，您必须提供 Kafka 代理服务器信息。如果您没有提供此信息，绑定器会期望您在默认的 localhost:9092 上运行代理。如果不是这种情况，则需要覆盖它。有几种方法可以做到这一点。

当涉及到绑定器级别属性时，无论您使用的是通过常规 Kafka 绑定器提供的代理属性 - spring.cloud.stream.kafka.binder.brokers 还是 Kafka Streams 绑定器，都没有关系。Kafka Streams 绑定器将首先检查是否设置了 Kafka Streams 绑定器特定的代理属性 (spring.cloud.stream.kafka.streams.binder.brokers)，如果未找到，则会查找 spring.cloud.stream.kafka.binder.brokers。

键始终使用原生 Serdes 反序列化。

对于值，默认情况下，Kafka 会在入站进行本机反序列化。请注意，这与之前版本的 Kafka Streams 绑定器中的默认行为有重大变化，在之前版本中，反序列化是由框架完成的。

Kafka Streams 绑定器将尝试通过查看 `java.util.function.Function|Consumer` 或 `StreamListener` 的类型签名来推断匹配的 `Serde` 类型。以下是它匹配 Serdes 的顺序。

如果上述策略均无效，则应用程序必须通过配置提供 `Serde`。这可以通过两种方式配置 - 绑定或默认。

首先，绑定器将查看是否在绑定级别提供了 `Serde`。例如，如果您有以下处理器，

那么，您可以使用以下方法提供绑定级别 `Serde`

如果您希望在入站反序列化时使用默认的键/值 Serdes，您可以在绑定器级别进行设置。

如果您不想使用 Kafka 提供的原生解码，您可以依靠 Spring Cloud Stream 提供的消息转换功能。由于原生解码是默认设置，为了让 Spring Cloud Stream 反序列化入站值对象，您需要显式禁用原生解码。

例如，如果您有与上面相同的 BiFunction 处理器，那么 spring.cloud.stream.bindings.process-in-0.consumer.nativeDecoding: false 您需要为所有输入单独禁用原生解码。否则，对于未禁用的输入，原生解码仍然会应用。

默认情况下，Spring Cloud Stream 将使用 application/json 作为内容类型，并使用相应的 json 消息转换器。您可以使用以下属性和相应的 MessageConverter bean 来使用自定义消息转换器。

出站序列化基本上遵循与入站反序列化相同的规则。与入站反序列化一样，Spring Cloud Stream 的先前版本的一个主要变化是，出站的序列化由 Kafka 本地处理。在 3.0 版本之前的绑定器中，这是由框架本身完成的。

出站的键始终由 Kafka 使用匹配的 Serde 进行序列化，该 Serde 由绑定器推断。如果它无法推断键的类型，则需要使用配置指定。

值 Serdes 使用与入站反序列化相同的规则进行推断。首先，它匹配以查看出站类型是否来自应用程序中提供的 bean。如果没有，它会检查它是否与 Kafka 公开的 Serde 匹配，例如 - Integer、Long、Short、Double、Float、byte[]、UUID 和 String。如果这不起作用，它将回退到 Spring Kafka 项目提供的 JsonSerde，但首先查看默认的 Serde 配置以查看是否有匹配项。请记住，所有这些都对应用程序透明。如果这些都不起作用，则用户必须通过配置提供要使用的 Serde。

假设您使用的是与上面相同的 BiFunction 处理器。然后，您可以配置出站键/值 Serdes，如下所示。

如果 Serde 推断失败，并且没有提供绑定级别 Serdes，则绑定器将回退到 JsonSerde，但会查看默认 Serdes 以进行匹配。

默认 Serdes 的配置方式与上面描述的反序列化中的配置方式相同。

spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde

如果您的应用程序使用分支功能并具有多个输出绑定，则必须为每个绑定配置这些绑定。同样，如果绑定程序能够推断出Serde类型，则无需进行此配置。

如果您不想使用 Kafka 提供的原生编码，而是想使用框架提供的消息转换，则需要显式禁用原生编码，因为原生编码是默认设置。例如，如果您具有与上述相同的 BiFunction 处理器，则spring.cloud.stream.bindings.process-out-0.producer.nativeEncoding: false。在分支的情况下，您需要为所有输出单独禁用原生编码。否则，原生编码仍将应用于您未禁用的那些输出。

当 Spring Cloud Stream 完成转换时，默认情况下，它将使用application/json作为内容类型，并使用相应的 json 消息转换器。您可以使用以下属性和相应的MessageConverter bean 来使用自定义消息转换器。

当禁用原生编码/解码时，绑定程序不会像原生 Serdes 那样进行任何推断。应用程序需要显式提供所有配置选项。因此，通常建议使用默认的序列化/反序列化选项，并在编写 Spring Cloud Stream Kafka Streams 应用程序时坚持使用 Kafka Streams 提供的原生序列化/反序列化。您必须使用框架提供的消息转换功能的一种情况是，当您的上游生产者使用特定的序列化策略时。在这种情况下，您希望使用匹配的反序列化策略，因为原生机制可能会失败。当依赖于默认的Serde机制时，应用程序必须确保绑定程序具有正确映射入站和出站的正确方法，并使用适当的Serde，否则可能会失败。

值得一提的是，上面概述的数据序列化/反序列化方法仅适用于处理器的边缘，即入站和出站。您的业务逻辑可能仍然需要调用显式需要Serde对象的 Kafka Streams API。这些仍然是应用程序的责任，必须由开发人员相应地处理。

Kafka Streams 绑定器允许使用以下属性指定上述反序列化异常处理程序。

或

除了上述两种反序列化异常处理程序之外，绑定器还提供第三种处理程序，用于将错误记录（毒丸）发送到 DLQ（死信队列）主题。以下是启用此 DLQ 异常处理程序的方法。

设置上述属性后，所有反序列化错误的记录将自动发送到 DLQ 主题。

您可以如下设置发布 DLQ 消息的主题名称。

您可以为 DlqDestinationResolver 提供实现，它是一个函数式接口。DlqDestinationResolver 以 ConsumerRecord 和异常作为输入，然后允许指定主题名称作为输出。通过访问 Kafka ConsumerRecord，可以在 BiFunction 的实现中内省标头记录。

以下是如何为 DlqDestinationResolver 提供实现的示例。

在为 DlqDestinationResolver 提供实现时，需要注意的一点是，绑定器中的供应器不会自动为应用程序创建主题。这是因为绑定器无法推断实现可能发送到的所有 DLQ 主题的名称。因此，如果您使用此策略提供 DLQ 名称，则应用程序有责任确保这些主题事先创建。

如果 DlqDestinationResolver 作为 bean 存在于应用程序中，则它具有更高的优先级。如果您不想遵循这种方法，而是使用配置提供静态 DLQ 名称，则可以设置以下属性。

如果设置了此属性，则错误记录将发送到主题 custom-dlq。如果应用程序没有使用上述任何策略，则它将创建一个名为 error.<input-topic-name>.<application-id> 的 DLQ 主题。例如，如果您的绑定的目标主题是 inputTopic，应用程序 ID 是 process-applicationId，则默认 DLQ 主题是 error.inputTopic.process-applicationId。如果您的意图是启用 DLQ，始终建议为每个输入绑定显式创建一个 DLQ 主题。

属性 spring.cloud.stream.kafka.streams.binder.deserializationExceptionHandler 适用于整个应用程序。这意味着，如果同一个应用程序中有多个函数或 StreamListener 方法，则此属性将应用于所有这些方法。但是，如果您在一个处理器中有多个处理器或多个输入绑定，则可以使用绑定器为每个输入消费者绑定提供的更细粒度的 DLQ 控制。

如果您有以下处理器，

并且您只想在第一个输入绑定上启用 DLQ，并在第二个绑定上启用 logAndSkip，则可以在消费者上执行以下操作。

spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.deserializationExceptionHandler: sendToDlq spring.cloud.stream.kafka.streams.bindings.process-in-1.consumer.deserializationExceptionHandler: logAndSkip

以这种方式设置反序列化异常处理程序比在绑定器级别设置具有更高的优先级。

默认情况下，记录使用与原始记录相同的分区发布到死信主题。这意味着死信主题必须至少与原始记录具有相同数量的分区。

要更改此行为，请将 DlqPartitionFunction 实现作为 @Bean 添加到应用程序上下文。只能存在一个这样的 bean。该函数将提供消费者组（在大多数情况下与应用程序 ID 相同）、失败的 ConsumerRecord 和异常。例如，如果您始终希望路由到分区 0，则可以使用

在使用 Kafka Streams 绑定器中的异常处理功能时，请牢记以下几点。

与上面描述的反序列化异常处理程序支持不同，绑定器没有为处理生产异常提供此类一流机制。但是，您仍然可以使用 StreamsBuilderFactoryBean 自定义程序配置生产异常处理程序，您可以在下面后续部分中找到有关此自定义程序的更多详细信息。

使用以下 API 方法检索与给定存储和键的组合关联的 KeyQueryMetadata 对象。

使用以下 API 方法检索与给定存储和键的组合关联的 KakfaStreams 对象。

如上所述，绑定器没有提供注册全局状态存储作为功能的第一类方法。为此，您需要使用自定义器。以下是如何操作。

再次，如果您有多个处理器，您需要通过使用上面概述的应用程序 ID 过滤掉其他 StreamsBuilderFactoryBean 对象，将全局状态存储附加到正确的 StreamsBuilder 上。

在错误处理部分，我们指出绑定器没有提供处理生产异常的一流方法。虽然情况如此，您仍然可以使用 StreamsBuilderFacotryBean 自定义器来注册生产异常处理程序。见下文。

再次，如果您有多个处理器，您可能希望将其适当地设置在正确的 StreamsBuilderFactoryBean 上。您也可以使用配置属性添加此类生产异常处理程序（有关详细信息，请参见下文），但这是一种选择，如果您选择使用编程方法。

以下属性在绑定器级别可用，必须以 spring.cloud.stream.kafka.streams.binder. 为前缀。

有关可能进入流配置的所有属性的更多信息，请参阅 Apache Kafka Streams 文档中的 StreamsConfig JavaDocs。您可以从 StreamsConfig 设置的所有配置都可以通过此设置。使用此属性时，它适用于整个应用程序，因为这是一个绑定器级别属性。如果您的应用程序中有多个处理器，它们都将获取这些属性。对于 application.id 之类的属性，这将变得很麻烦，因此您必须仔细检查如何使用此绑定器级别 configuration 属性映射来自 StreamsConfig 的属性。

以下属性仅适用于 Kafka Streams 生产者，并且必须以 spring.cloud.stream.kafka.streams.bindings.<绑定名称>.producer. 为前缀。为了方便起见，如果有多个输出绑定并且它们都需要一个公共值，则可以使用前缀 spring.cloud.stream.kafka.streams.default.producer. 进行配置。

streamPartitionerBeanName: 要在消费者处使用的自定义出站分区程序 bean 名称。应用程序可以提供自定义 StreamPartitioner 作为 Spring bean，并且可以将此 bean 的名称提供给生产者以代替默认分区程序。

+ 默认值：请参阅上面关于出站分区支持的讨论。

以下属性适用于 Kafka Streams 消费者，必须以 spring.cloud.stream.kafka.streams.bindings.<绑定名称>.consumer. 为前缀。为了方便起见，如果有多个输入绑定并且它们都需要一个公共值，则可以使用前缀 spring.cloud.stream.kafka.streams.default.consumer. 进行配置。

注意：在消费者上使用 resetOffsets 对 Kafka Streams 绑定程序没有影响。与基于消息通道的绑定程序不同，Kafka Streams 绑定程序不会按需寻求开始或结束。

在 Kafka Streams 中，您可以使用 num.stream.threads 属性控制处理器可以创建的线程数。您可以使用上面在绑定程序、函数、生产者或消费者级别描述的各种 configuration 选项来执行此操作。您还可以使用核心 Spring Cloud Stream 为此目的提供的 concurrency 属性。使用此属性时，您需要在消费者上使用它。当您在函数或 StreamListener 中有多个输入绑定时，请在第一个输入绑定上设置此属性。例如，当设置 spring.cloud.stream.bindings.process-in-0.consumer.concurrency 时，它将被绑定程序转换为 num.stream.threads。如果您有多个处理器，并且一个处理器定义了绑定级别并发性，而其他处理器没有，那么那些没有绑定级别并发性的处理器将默认回退到通过 spring.cloud.stream.kafka.streams.binder.configuration.num.stream.threads 指定的绑定程序范围属性。如果此绑定程序配置不可用，则应用程序将使用 Kafka Streams 设置的默认值。

我们建议在使用 eclipse 时使用 m2eclipe eclipse 插件。如果您还没有安装 m2eclipse，它可以在“eclipse 市场”中找到。

不幸的是，m2e 还不支持 Maven 3.3，因此在将项目导入 Eclipse 后，您还需要告诉 m2eclipse 使用项目的 .settings.xml 文件。如果您不这样做，您可能会看到与项目中的 POM 相关的许多不同的错误。打开您的 Eclipse 首选项，展开 Maven 首选项，然后选择用户设置。在用户设置字段中，单击浏览并导航到您导入的 Spring Cloud 项目，选择该项目中的 .settings.xml 文件。单击应用，然后单击确定以保存首选项更改。

如果您不想使用 m2eclipse，您可以使用以下命令生成 eclipse 项目元数据

生成的 Eclipse 项目可以通过从“文件”菜单中选择“导入现有项目”来导入。

[[contributing] == 贡献

Spring Cloud 在非限制性的 Apache 2.0 许可下发布，并遵循非常标准的 Github 开发流程，使用 Github 跟踪器来处理问题并将拉取请求合并到主分支。如果您想贡献即使是微不足道的东西，请不要犹豫，但请遵循以下指南。