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

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

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

以下是可能使用此属性的两个用例。

有关主题偏移量的更多控制，请参见使用 KafkaRebalanceListener；当提供了 listener 时，resetOffsets 不应设置为 true，否则会导致错误。 >>>>>>> 7bc90c10…​ GH-1084: 添加 txCommitRecovered 属性

从 3.0 版本开始，当 spring.cloud.stream.binding.<name>.consumer.batch-mode 设置为 true 时，通过轮询 Kafka Consumer 收到的所有记录将以 List<?> 的形式呈现给 listener 方法。否则，方法将一次处理一条记录。批次的大小由 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 创建到应用程序上下文中。当 binder 检测到此类 bean 时，它会优先使用，否则将使用 dlqName 属性。如果两者都找不到，它将默认为 error.<destination>.<group>。这里是一个将 DlqDestinationResolver 作为 @Bean 的示例。

提供 DlqDestinationResolver 的实现时，要记住一个重要的事情是，binder 中的 provisioner 不会自动为应用程序创建主题。这是因为 binder 无法推断出实现可能发送到的所有 DLQ 主题的名称。因此，如果你使用此策略提供 DLQ 名称，确保这些主题事先创建是应用程序的责任。

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

示例假设原始目的地是 so8400out，消费者组是 so8400。

有几种策略可以考虑

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

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

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

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

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

一旦构建为 uber-jar（例如 kstream-consumer-app.jar），就可以按如下方式运行上述示例。

这是另一个示例，它是一个具有输入和输出绑定的完整 processor。这是经典的词频统计示例，应用程序从主题接收数据，然后在 tumbling time-window 中计算每个单词的出现次数。

这里同样，基本主题与之前的示例相同，但这里我们有两个输入。Java 的 BiFunction 支持用于将输入绑定到期望的目的地。binder 为输入生成的默认绑定名称分别是 process-in-0 和 process-in-1。默认输出绑定是 process-out-0。在此示例中，BiFunction 的第一个参数被绑定为第一个输入的 KStream，第二个参数被绑定为第二个输入的 KTable。

一旦构建为 uber-jar（例如 wordcount-processor.jar），就可以按如下方式运行上述示例。

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

Spring Cloud Stream 将确保来自入站和出站主题的消息自动绑定为 KStream 对象。作为开发人员，你可以专注于代码的业务方面，即编写 processor 中所需的逻辑。由 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 绑定接口，它具有以下契约。

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

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

这是另一个有两个输入的 sink 示例。

以下是上面看到的基于 BiFunction 的 processor 的 StreamListener 等价版本。

最后，这里是具有三个输入和 curried 函数的应用程序的 StreamListener 等价版本。

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

在我们继续探讨 Kafka Streams binder 提供的通用编程模型之前，这里是多个输出绑定的 StreamListener 版本。

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

binder 为输入提供 KStream、KTable 和 GlobalKTable 的绑定能力。KTable 和 GlobalKTable 绑定仅在输入端可用。Binder 支持 KStream 的输入和输出绑定。

Kafka Streams binder 编程模型的要点在于，binder 提供了灵活性，你可以选择完全函数式编程模型，或者使用基于 StreamListener 的命令式方法。

Binder 允许在单个 Spring Cloud Stream 应用程序中包含多个 Kafka Streams processor。你可以拥有如下所示的应用程序。

在这种情况下，binder 将创建 3 个具有不同 application ID 的独立 Kafka Streams 对象（更多内容见下文）。但是，如果应用程序中有一个以上的 processor，你必须告诉 Spring Cloud Stream 哪些函数需要被激活。以下是激活函数的方法。

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

如果你希望某些函数不立即激活，可以将其从列表中移除。

当你在同一个应用程序中有一个单一的 Kafka Streams processor 和其他类型通过不同 binder 处理的 Function bean 时（例如，基于常规 Kafka Message Channel binder 的函数 bean），这也适用。

Application ID 是 Kafka Streams 应用程序必须提供的强制属性。Spring Cloud Stream Kafka Streams binder 允许你通过多种方式配置此 application ID。

如果应用程序中只有一个 processor 或 StreamListener，则可以使用以下属性在 binder 级别进行设置

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

为了方便起见，如果你只有一个 processor，还可以使用 spring.application.name 作为属性来委托设置 application ID。

如果应用程序中有多个 Kafka Streams processor，则需要为每个 processor 设置 application ID。对于函数式模型，可以将其作为属性附加到每个函数上。

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

和

然后，你可以使用以下 binder 级别属性为每个函数设置 application ID。

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

和

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

对于 StreamListener，你需要将其设置在 processor 的第一个输入绑定上。

例如，假设你有以下两个基于 StreamListener 的 processor。

那么你必须使用以下绑定属性来设置其 application ID。

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

和

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

对于基于函数的模型，这种在绑定级别设置 application ID 的方法也适用。然而，如果你使用函数式模型，像我们上面看到的那样在 binder 级别按函数设置要容易得多。

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

如果应用程序没有提供 application ID，那么在这种情况下，binder 会自动为你生成一个静态 application ID。这在开发场景中很方便，因为它避免了显式提供 application ID 的需要。通过这种方式生成的 application ID 在应用程序重启后将保持静态。对于函数式模型，生成的 application ID 将是函数 bean 名称后跟文字 applicationID，例如如果函数 bean 名称是 process，则生成的 ID 是 process-applicationID。对于 StreamListener，生成的 application ID 不使用函数 bean 名称，而是使用包含类名后跟方法名，再后跟文字 applicationId。

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

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

例如，假设你有这个函数。

Binder 将生成名为 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 broker 服务器信息。如果你没有提供此信息，binder 会默认期望 broker 运行在 localhost:9092。如果不是这种情况，则需要覆盖此设置。有几种方法可以做到这一点。

对于 binder 级别属性，无论你是否使用通过常规 Kafka binder 提供的 broker 属性 - spring.cloud.stream.kafka.binder.brokers 都没有关系。Kafka Streams binder 首先会检查是否设置了 Kafka Streams binder 特定的 broker 属性（spring.cloud.stream.kafka.streams.binder.brokers），如果找不到，则会查找 spring.cloud.stream.kafka.binder.brokers。

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

对于值，默认情况下，入站的反序列化由 Kafka 原生执行。请注意，这是与之前版本 Kafka Streams binder 的默认行为的重大改变，之前版本的反序列化是由框架完成的。

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

如果上述策略都不奏效，那么应用程序必须通过配置提供 `Serde`。这可以通过两种方式进行配置 - 绑定级别或默认级别。

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

那么，您可以使用以下方式提供绑定级别的 Serde：

如果您希望将默认的 key/value 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 版本之前，这是由框架本身完成的。

出站的 key 总是由 Kafka 使用绑定器推断出的匹配 Serde 进行序列化。如果无法推断 key 的类型，则需要通过配置指定。

值 Serdes 的推断使用与入站反序列化相同的规则。首先检查出站类型是否与应用程序中提供的 Bean 匹配。如果不匹配，则检查是否与 Kafka 暴露的 Serde（例如 - Integer, Long, Short, Double, Float, byte[], UUID 和 String）匹配。如果仍然不匹配，则回退到 Spring Kafka 项目提供的 JsonSerde，但在此之前会先查看默认的 Serde 配置是否存在匹配。请记住，所有这些对应用程序都是透明的。如果这些都不奏效，用户必须通过配置提供要使用的 Serde。

假设您正在使用上面相同的 BiFunction 处理器。那么您可以按如下方式配置出站 key/value 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，否则可能会失败。

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

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

或

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

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

您可以通过以下方式设置 DLQ 消息发布的 topic 名称。

您可以提供一个 DlqDestinationResolver 的实现，这是一个函数式接口。DlqDestinationResolver 将 ConsumerRecord 和异常作为输入，然后允许指定一个 topic 名称作为输出。通过访问 Kafka ConsumerRecord，可以在 BiFunction 的实现中检查 header 记录。

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

提供 DlqDestinationResolver 的实现时，要记住一个重要的事情是，binder 中的 provisioner 不会自动为应用程序创建主题。这是因为 binder 无法推断出实现可能发送到的所有 DLQ 主题的名称。因此，如果你使用此策略提供 DLQ 名称，确保这些主题事先创建是应用程序的责任。

如果应用程序中存在 DlqDestinationResolver bean，则该 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 方法检索与给定存储和 key 组合关联的 KeyQueryMetadata 对象。

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

如上所述，绑定器没有提供一种一流的方式来注册全局状态存储作为一项功能。为此，您需要使用定制器。以下是如何做到这一点。

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

在错误处理部分，我们提到绑定器没有提供一种一流的方式来处理生产异常。尽管如此，您仍然可以使用 StreamsBuilderFacotryBean 定制器来注册生产异常处理器。见下文。

再一次，如果您有多个处理器，您可能需要适当设置正确的 StreamsBuilderFactoryBean。您也可以使用配置属性添加此类生产异常处理器（更多内容见下文），但如果您选择采用编程式方法，这是一种选择。

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

有关所有可用于 streams 配置的属性的更多信息，请参阅 Apache Kafka Streams 文档中的 StreamsConfig JavaDocs。您可以在 StreamsConfig 中设置的所有配置都可以通过此属性设置。使用此属性时，它适用于整个应用程序，因为这是绑定器级别的属性。如果应用程序中有多个处理器，所有处理器都将获取这些属性。对于像 application.id 这样的属性，这会带来问题，因此您必须仔细检查如何使用此绑定器级别的 configuration 属性映射 StreamsConfig 中的属性。

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

streamPartitionerBeanName：要在消费者端使用的自定义出站分区器 bean 名称。应用可以将自定义的 StreamPartitioner 作为 Spring bean 提供，并将该 bean 的名称提供给生产者，以代替默认分区器。

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

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

注意：在消费者端使用 resetOffsets 对 Kafka Streams binder 没有影响。与基于消息通道的 binder 不同，Kafka Streams binder 不会按需定位到开始或结束。

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

我们建议在使用 Eclipse 时使用 m2eclipse Eclipse 插件。如果您尚未安装 m2eclipse，可以从“Eclipse Marketplace”获取。

遗憾的是，m2e 尚不支持 Maven 3.3，因此将项目导入 Eclipse 后，您还需要告诉 m2eclipse 使用项目中的 .settings.xml 文件。如果您不这样做，可能会看到与项目中 POM 相关的许多不同错误。打开 Eclipse 偏好设置，展开 Maven 偏好设置，然后选择 User Settings。在 User Settings 字段中，单击 Browse 并导航到您导入的 Spring Cloud 项目，选择该项目中的 .settings.xml 文件。点击 Apply 然后点击 OK 保存偏好设置更改。

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

生成的 Eclipse 项目可以通过从 file 菜单中选择 import existing projects 来导入。

[[contributing] == 贡献

Spring Cloud 以非限制性的 Apache 2.0 许可发布，并遵循非常标准的 Github 开发流程，使用 Github 跟踪器处理问题并将 pull requests 合并到 master 分支。如果您想贡献哪怕是微不足道的东西，请不要犹豫，但请遵循以下准则。