以下属性仅适用于 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的函数模型中不提供任何绑定接口。

以下是一个具有两个输入的 Sink 的示例。

以下是我们上面看到的基于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 时，这种情况也是如此，这些 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 绑定器特定的代理属性（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 进行序列化。如果它无法推断键的类型，则需要使用配置进行指定。

值 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映射，否则可能会失败。

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

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.<binding name>.producer. 为前缀。为方便起见，如果有多个输出绑定并且它们都需要一个公共值，则可以通过使用前缀 spring.cloud.stream.kafka.streams.default.producer. 进行配置。

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

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

以下属性适用于 Kafka Streams 消费者，并且必须以 spring.cloud.stream.kafka.streams.bindings.<binding-name>.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 跟踪器来处理问题并将拉取请求合并到 master 分支。如果您想贡献任何东西，即使是很小的东西，也请不要犹豫，但请遵循以下准则。