参考指南

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

从版本 3.0 开始，当设置为 时，轮询 Kafka 收到的所有记录都将作为 to 侦听器方法显示。 否则，将一次使用一条记录调用该方法。 批处理的大小由 Kafka 使用者属性 ， ， ;有关更多信息，请参阅 Kafka 文档。spring.cloud.stream.binding.<name>.consumer.batch-modetrueConsumerList<?>max.poll.recordsmin.fetch.bytesfetch.max.wait.msspring-doc.cn

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

ResultMetadata meta = sendResultMsg.getHeaders().get(KafkaHeaders.RECORD_METADATA, RecordMetadata.class)spring-doc.cn

失败的发送将转到 producer 错误通道（如果已配置）;请参见错误通道。 默认值：nullspring-doc.cn

+spring-doc.cn

在本节中，我们将展示上述属性在特定情况下的用法。spring-doc.cn

默认情况下，使用与原始记录相同的分区将记录发布到 Dead-Letter 主题。 这意味着 Dead-Letter 主题必须至少具有与原始记录一样多的分区。spring-doc.cn

要更改此行为，请将 implementation 作为 添加到 application context。 只能存在一个这样的 bean。 该函数随 consumer group、failed 和 exception 一起提供。 例如，如果您始终希望路由到分区 0，则可以使用：DlqPartitionFunction@BeanConsumerRecordspring-doc.cn

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

这些示例假定原始目标是 ，使用者组是 。so8400outso8400spring-doc.cn

有几种策略需要考虑：spring-doc.cn

以下代码清单显示了示例应用程序：spring-doc.cn

从 Spring Cloud Stream 开始，Kafka Streams Binder 允许使用 Java 8 中提供的函数式编程样式来设计和开发应用程序。 这意味着应用程序可以简洁地表示为类型为 或 的 lambda 表达式。3.0.0java.util.function.Functionjava.util.function.Consumerspring-doc.cn

让我们举一个非常基本的例子。spring-doc.cn

尽管简单，但这是一个完整的独立 Spring Boot 应用程序，它利用 Kafka Streams 进行流处理。 这是一个没有出站绑定，只有一个入站绑定的使用者应用程序。 应用程序使用数据，它只是在标准输出上记录 key 和 value 中的信息。 该应用程序包含批注和标记为 . Bean 方法的类型是使用 进行参数化的。 然后，在实现中，我们返回一个 Consumer 对象，它本质上是一个 lambda 表达式。 在 lambda 表达式中，提供了用于处理数据的代码。KStreamSpringBootApplicationBeanjava.util.function.ConsumerKStreamspring-doc.cn

在此应用程序中，有一个类型为 的输入绑定。 Binders 使用 name 为应用程序创建此绑定，即函数 Bean 名称的名称后跟一个短划线字符 （） 和文字，后跟另一个短划线，然后是参数的序号位置。 您可以使用此绑定名称来设置其他属性，例如 destination。 例如。KStreamprocess-in-0-inspring.cloud.stream.bindings.process-in-0.destination=my-topicspring-doc.cn

一旦构建为 uber-jar（例如 ），您就可以运行上面的示例，如下所示。kstream-consumer-app.jarspring-doc.cn

这是另一个示例，其中它是一个同时具有输入和输出绑定的完整处理器。 这是一个经典的字数统计示例，其中应用程序从主题接收数据，然后在滚动时间窗口中计算每个单词的出现次数。spring-doc.cn

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

一旦构建为 uber-jar（例如 ），您就可以运行上面的示例，如下所示。wordcount-processor.jarspring-doc.cn

此应用程序将使用来自 Kafka 主题的消息，并将计算结果发布到输出 主题。wordscountsspring-doc.cn

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

我们上面看到的两个示例具有单个 input binding。在这两种情况下，绑定都接收来自单个主题的记录。 如果要将多个主题多路复用到单个绑定中，可以在下面提供逗号分隔的 Kafka 主题作为目标。KStreamKStreamspring-doc.cn

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

此外，如果要将主题与常规 exression 匹配，还可以提供主题模式作为目标。spring-doc.cn

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

尽管上面概述的函数式编程模型是首选方法，但如果您愿意，您仍然可以使用基于经典的方法。StreamListenerspring-doc.cn

以下是一些示例。spring-doc.cn

以下是使用 的字数统计示例的等效示例。StreamListenerspring-doc.cn

如你所见，这有点冗长，因为你需要提供和其他额外的注释，比如 and 才能使其成为一个完整的应用程序。 是指定包含绑定的 Binding 接口的位置。 在本例中，我们使用的是具有以下 Contract 的 stock binding 接口。EnableBindingStreamListenerSendToEnableBindingKafkaStreamsProcessorspring-doc.cn

Binder 将为输入和输出创建绑定，因为你正在使用包含这些声明的绑定接口。KStreamKStreamspring-doc.cn

除了函数式样式中提供的编程模型的明显差异之外，这里需要提到的一点是，绑定名称是您在绑定接口中指定的名称。 例如，在上面的应用程序中，由于我们使用的是 ，因此绑定名称是 和 。 Binding 属性需要使用这些名称。例如 ，等等。 请记住，这与函数式样式有根本的不同，因为 Binders 会为应用程序生成绑定名称。 这是因为应用程序没有在功能模型中使用 .KafkaStreamsProcessorinputoutputspring.cloud.stream.bindings.input.destinationspring.cloud.stream.bindings.output.destinationEnableBindingspring-doc.cn

这是另一个 sink 示例，其中有两个 inputs。spring-doc.cn

以下是我们上面看到的相同基于处理器的等效项。StreamListenerBiFunctionspring-doc.cn

最后，这里相当于具有三个 inputs 和 curried 函数的应用程序。StreamListenerspring-doc.cn

你可能会注意到，上面两个例子更加冗长，因为除了 provide 之外，你还需要编写自己的自定义绑定接口。 使用功能模型，您可以避免所有这些仪式细节。EnableBindingspring-doc.cn

在我们继续查看 Kafka Streams Binder 提供的通用编程模型之前，这里是多个输出绑定的版本。StreamListenerspring-doc.cn

回顾一下，我们已经回顾了使用 Kafka Streams Binder 时的各种编程模型选择。spring-doc.cn

Binder 为 和 input 提供绑定功能。 和 bindings 仅在 input 上可用。 Binder 支持 .KStreamKTableGlobalKTableKTableGlobalKTableKStreamspring-doc.cn

Kafka Streams Binder 编程模型的结果是，Binder 为您提供了使用全功能编程模型或使用基于命令式方法的灵活性。StreamListenerspring-doc.cn

Binder 允许在单个 Spring Cloud Stream 应用程序中拥有多个 Kafka Streams 处理器。 您可以按以下方式进行申请。spring-doc.cn

在这种情况下，Binder 将创建 3 个具有不同应用程序 ID 的单独 Kafka Streams 对象（更多内容见下文）。 但是，如果应用程序中有多个处理器，则必须告诉 Spring Cloud Stream，需要激活哪些功能。 以下是激活功能的方法。spring-doc.cn

spring.cloud.stream.function.definition: process;anotherProcess;yetAnotherProcessspring-doc.cn

如果您希望某些功能不立即激活，您可以将其从此列表中删除。spring-doc.cn

当您在同一应用程序中拥有单个 Kafka Streams 处理器和其他类型的 bean 时，这些 bean 通过不同的 Binders（例如，基于常规 Kafka Message Channel Binders 的函数 Bean）进行处理时，也是如此Functionspring-doc.cn

应用程序 ID 是您需要为 Kafka Streams 应用程序提供的必需属性。 Spring Cloud Stream Kafka Streams Binder 允许您以多种方式配置此应用程序 ID。spring-doc.cn

如果您只有一个处理器或在应用程序中，则可以使用以下属性在 Binder 级别进行设置：StreamListenerspring-doc.cn

spring.cloud.stream.kafka.streams.binder.applicationId.spring-doc.cn

为方便起见，如果您只有一个处理器，您还可以用作 the property 来委托应用程序 ID。spring.application.namespring-doc.cn

如果应用程序中有多个 Kafka Streams 处理器，则需要为每个处理器设置应用程序 ID。 对于函数模型，您可以将其作为属性附加到每个函数。spring-doc.cn

例如，假设您有以下功能。spring-doc.cn

和spring-doc.cn

然后，您可以使用以下 Binder 级别属性为每个应用程序设置应用程序 ID。spring-doc.cn

spring.cloud.stream.kafka.streams.binder.functions.process.applicationIdspring-doc.cn

和spring-doc.cn

spring.cloud.stream.kafka.streams.binder.functions.anotherProcess.applicationIdspring-doc.cn

对于 ，您需要在处理器上的第一个 input binding 上设置此项。StreamListenerspring-doc.cn

例如，假设您有以下两个基于处理器的处理器。StreamListenerspring-doc.cn

然后，您必须使用以下 binding 属性为此设置应用程序 ID。spring-doc.cn

spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationIdspring-doc.cn

和spring-doc.cn

spring.cloud.stream.kafka.streams.bindings.anotherInput.consumer.applicationIdspring-doc.cn

对于基于函数的模型，这种在绑定级别设置应用程序 ID 的方法也适用。 但是，如果您使用的是函数模型，则在上面看到的 binder 级别设置每个函数要容易得多。spring-doc.cn

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

如果应用程序未提供应用程序 ID，则在这种情况下，Binder 将为您自动生成静态应用程序 ID。 这在开发方案中很方便，因为它避免了显式提供应用程序 ID 的需要。 以这种方式生成的应用程序 ID 在应用程序重启时将是静态的。 在函数模型的情况下，生成的应用程序 ID 将是函数 bean 名称，后跟文字 ，例如，如果是函数 bean 名称。 在 的情况下，生成的应用程序 ID 将使用包含类名称，后跟方法名称，后跟文字 ，而不是使用函数 bean 名称。applicationIDprocess-applicationIDprocessStreamListenerapplicationIdspring-doc.cn

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

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

例如，假设您有这个函数。spring-doc.cn

Binder 将生成名称为 、 和 的绑定。 现在，如果你想把它们完全改成别的，也许是更多特定于域的绑定名称，那么你可以按如下方式进行。process-in-0process-in-1process-out-0spring-doc.cn

springc.cloud.stream.function.bindings.process-in-0=usersspring-doc.cn

springc.cloud.stream.function.bindings.process-in-0=regionsspring-doc.cn

和spring-doc.cn

spring.cloud.stream.function.bindings.process-out-0=clicksspring-doc.cn

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

请记住，对于上述函数式编程模型，在大多数情况下，遵守默认绑定名称是有意义的。 您可能仍然希望进行此覆盖的唯一原因是，当您拥有大量配置属性并且希望将绑定映射到对域更友好的内容时。spring-doc.cn

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

当涉及到 Binder 级别属性时，是否使用通过常规 Kafka Binder - 提供的 broker 属性并不重要。 Kafka Streams Binder 将首先检查是否设置了 Kafka Streams Binder 特定的代理属性 （），如果未找到，则查找 .spring.cloud.stream.kafka.binder.brokersspring.cloud.stream.kafka.streams.binder.brokersspring.cloud.stream.kafka.binder.brokersspring-doc.cn

键总是使用本机 Serdes 反序列化。spring-doc.cn

对于值，默认情况下，入站上的反序列化由 Kafka 本机执行。 请注意，这是对以前版本的 Kafka Streams Binder 默认行为的重大更改，其中反序列化是由框架完成的。spring-doc.cn

Kafka Streams Binder 将尝试通过查看 或 的类型签名来推断匹配的类型。 这是它与 Serdes 匹配的顺序。Serdejava.util.function.Function|ConsumerStreamListenerspring-doc.cn

如果上述策略均无效，则应用程序必须通过配置提供 'Serde'。 这可以通过两种方式进行配置 - binding 或 default。spring-doc.cn

首先，Binder 将查看是否在绑定级别提供了 a。 例如，如果您有以下处理器，Serdespring-doc.cn

然后，您可以使用以下内容提供绑定级别：Serdespring-doc.cn

如果您希望将默认键/值 Serdes 用于入站反序列化，则可以在 Binder 级别执行此操作。spring-doc.cn

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

例如，如果你有与上述相同的 BiFunction 处理器，那么你需要单独禁用所有输入的原生解码。否则，本机解码仍将应用于您未禁用的那些。spring.cloud.stream.bindings.process-in-0.consumer.nativeDecoding: falsespring-doc.cn

默认情况下，Spring Cloud Stream 将用作内容类型并使用适当的 json 消息转换器。 您可以通过使用以下属性和适当的 Bean 来使用自定义消息转换器。application/jsonMessageConverterspring-doc.cn

出站序列化几乎遵循与上述入站反序列化相同的规则。 与入站反序列化一样，与以前版本的 Spring Cloud Stream 相比，一个主要变化是出站上的序列化由 Kafka 本地处理。 在 Binder 的 3.0 版本之前，这是由框架本身完成的。spring-doc.cn

出站上的键始终由 Kafka 使用由 Binder 推断的匹配项进行序列化。 如果它无法推断出键的类型，则需要使用 configuration 指定。Serdespring-doc.cn

值 serdes 是使用用于入站反序列化的相同规则推断的。 首先，它进行匹配以查看出站类型是否来自应用程序中提供的 bean。 如果不匹配，它会检查它是否与 Kafka 公开的匹配项匹配，例如 - 、 和 。 如果这不起作用，那么它就会回退到 Spring Kafka 项目提供的，但首先查看默认配置以查看是否有匹配项。 请记住，所有这些都对应用程序是透明的。 如果这些都不起作用，则用户必须提供 to use by configuration。SerdeIntegerLongShortDoubleFloatbyte[]UUIDStringJsonSerdeSerdeSerdespring-doc.cn

假设您使用的是与上述相同的处理器。然后，您可以按如下方式配置出站键/值 Serdes。BiFunctionspring-doc.cn

如果 Serde 推理失败，并且未提供绑定级别 Serdes，则 Binders 将回退到 ，但请查看默认 Serdes 进行匹配。JsonSerdespring-doc.cn

默认 serdes 的配置方式与上面相同，在 deserialization 下描述。spring-doc.cn

spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde spring.cloud.stream.kafka.streams.binder.configuration.default.value.serdespring-doc.cn

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

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

当 Spring Cloud Stream 完成转换时，默认情况下，它将用作内容类型并使用适当的 json 消息转换器。 您可以通过使用以下属性和相应的 Bean 来使用自定义消息转换器。application/jsonMessageConverterspring-doc.cn

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

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

Kafka Streams Binder 允许使用以下属性指定上述反序列化异常处理程序。spring-doc.cn

或spring-doc.cn

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

当设置了上述属性时，所有 disserialization error 中的记录都会自动发送到 DLQ 主题。spring-doc.cn

您可以设置发布 DLQ 消息的主题名称，如下所示。spring-doc.cn

如果设置了此选项，则错误记录将发送到主题 。 如果未设置，则它将创建一个名为 . 例如，如果绑定的目标主题是 ，应用程序 ID 是 ，则默认 DLQ 主题是 。 如果您打算启用 DLQ，则始终建议为每个输入绑定显式创建一个 DLQ 主题。custom-dlqerror.<input-topic-name>.<application-id>inputTopicprocess-applicationIderror.inputTopic.process-applicationIdspring-doc.cn

该属性适用于整个应用程序。 这意味着，如果同一应用程序中有多个函数或方法，则此属性将应用于所有函数或方法。 但是，如果单个处理器中有多个处理器或多个输入绑定，则可以使用 Binders 为每个输入使用者绑定提供的更精细的 DLQ 控件。spring.cloud.stream.kafka.streams.binder.deserializationExceptionHandlerStreamListenerspring-doc.cn

如果您有以下处理器，spring-doc.cn

并且您只想在第一个输入绑定上启用 DLQ，在第二个绑定上启用 logAndSkip，那么您可以在使用者上执行此操作，如下所示。spring-doc.cn

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

以这种方式设置反序列化异常处理程序的优先级高于在 Binder 级别设置。spring-doc.cn

默认情况下，使用与原始记录相同的分区将记录发布到 Dead-Letter 主题。 这意味着 Dead-Letter 主题必须至少具有与原始记录一样多的分区。spring-doc.cn

要更改此行为，请将 implementation 作为 添加到 application context。 只能存在一个这样的 bean。 该函数提供 Consumer Group（大多数情况下与应用程序 ID 相同）、failed 和 exception。 例如，如果您始终希望路由到分区 0，则可以使用：DlqPartitionFunction@BeanConsumerRecordspring-doc.cn

在 Kafka Streams Binder 中使用异常处理功能时，需要记住几点。spring-doc.cn

与上述对反序列化异常处理程序的支持不同，Binders 不提供此类处理 生产异常的一类机制。 但是，您仍然可以使用定制器配置生产异常处理程序，您可以在下面的后续部分中找到更多详细信息。StreamsBuilderFactoryBeanspring-doc.cn

如上所述，Binders 没有提供将全局状态存储注册为功能的第一类方法。 为此，您需要使用定制器。 这是如何做到这一点的。spring-doc.cn

同样，如果您有多个处理器，则需要使用上述应用程序 ID 过滤掉其他对象，从而将全局状态存储附加到右侧。StreamsBuilderStreamsBuilderFactoryBeanspring-doc.cn

在错误处理部分中，我们指出了 Binder 没有提供处理生产异常的第一类方法。 尽管如此，您仍然可以使用定制器来注册生产异常处理程序。见下文。StreamsBuilderFacotryBeanspring-doc.cn

同样，如果您有多个处理器，则可能需要针对正确的 进行适当设置。 您还可以使用 configuration 属性添加此类 production 异常处理程序（有关更多信息，请参见下文），但如果您选择使用编程方法，则这是一个选项。StreamsBuilderFactoryBeanspring-doc.cn

以下属性在 Binder 级别可用，并且必须以spring.cloud.stream.kafka.streams.binder.spring-doc.cn

有关可能进入 streams 配置的所有属性的更多信息，请参阅 Apache Kafka Streams 文档中的 JavaDocs。 您可以设置的所有配置都可以通过它进行设置。 使用此属性时，它适用于整个应用程序，因为这是 Binder 级别属性。 如果应用程序中有多个处理器，则所有处理器都将获取这些属性。 对于像这样的属性，这将成为问题，因此您必须仔细检查如何使用此 Binder 级别属性映射 中的属性。StreamsConfigStreamsConfigapplication.idStreamsConfigconfigurationspring-doc.cn

以下属性仅适用于 Kafka Streams 创建者，并且必须以 为前缀为方便，如果有多个输出绑定，并且它们都需要一个通用值，则可以使用 prefix 进行配置。spring.cloud.stream.kafka.streams.bindings.<binding name>.producer.spring.cloud.stream.kafka.streams.default.producer.spring-doc.cn

streamPartitionerBeanName 中： 要在使用者处使用的自定义出站分区器 bean 名称。 应用程序可以提供 custom 作为 Spring bean，并且可以将此 bean 的名称提供给生产者使用，而不是默认名称。StreamPartitionerspring-doc.cn

+ Default：请参阅上面关于出站分区支持的讨论。spring-doc.cn

以下属性可供 Kafka Streams 使用者使用，并且必须以 为前缀为方便，如果有多个输入绑定，并且它们都需要一个通用值，则可以使用 prefix 进行配置 。spring.cloud.stream.kafka.streams.bindings.<binding-name>.consumer.spring.cloud.stream.kafka.streams.default.consumer.spring-doc.cn

注意：对使用者使用不会对 Kafka Streams Binder 产生任何影响。 与基于消息通道的 Binder 不同，Kafka Streams Binder 不会按需开始或结束。resetOffsetsspring-doc.cn

在 Kafka Streams 中，您可以使用该属性控制处理器可以创建的线程数。 为此，您可以使用上面在 binder、functions、producer 或 consumer 级别下描述的各种选项来完成。 您还可以使用核心 Spring Cloud Stream 提供的属性来实现此目的。 使用时，您需要在消费者上使用它。 当您在 function 或 中有多个 input bindings 时，请在第一个 input 绑定上设置此项。 例如，当设置 时，它将被 Binder 翻译为 。num.stream.threadsconfigurationconcurrencyStreamListenerspring.cloud.stream.bindings.process-in-0.consumer.concurrencynum.stream.threadsspring-doc.cn