生产与消费消息
你可以通过简单地编写函数并将它们作为 @Bean 暴露来编写 Spring Cloud Stream 应用程序。你也可以使用基于 Spring Integration 注解的配置或基于 Spring Cloud Stream 注解的配置，尽管从 spring-cloud-stream 3.x 开始，我们推荐使用函数式实现。
Spring Cloud Function 支持
概述
自 Spring Cloud Stream v2.1 起，定义流处理器和源的另一种方法是使用对 Spring Cloud Function 的内置支持，它们可以表示为 java.util.function.[Supplier/Function/Consumer] 类型的 bean。
要指定哪个功能 Bean 绑定到由绑定公开的外部目标，你必须提供 spring.cloud.function.definition 属性。
如果你只有一个类型为 java.util.function.[Supplier/Function/Consumer] 的 bean，你可以跳过 spring.cloud.function.definition 属性，因为这样的功能性 bean 会被自动发现。然而，使用该属性被认为是避免混淆的最佳实践。有时，这种自动发现可能会带来问题，因为类型为 java.util.function.[Supplier/Function/Consumer] 的单个 bean 可能用于处理消息以外的其他目的，但由于它是唯一的，它会被自动发现并自动绑定。对于这些罕见的情况，你可以通过设置 spring.cloud.stream.function.autodetect 属性为 false 来禁用自动发现。
这是一个应用程序示例，它将消息处理器暴露为 java.util.function.Function，通过充当数据的消费者和生产者，有效地支持 直通 语义。
@SpringBootApplication
public class MyFunctionBootApp {
public static void main(String[] args) {
SpringApplication.run(MyFunctionBootApp.class);
}
@Bean
public Function<String, String> toUpperCase() {
return s -> s.toUpperCase();
}
}
在前面的示例中，我们定义了一个类型为 java.util.function.Function 的 bean，名为 toUpperCase，它充当消息处理器，其 'input' 和 'output' 必须绑定到由提供的目标绑定器暴露的外部目标。默认情况下，'input' 和 'output' 的绑定名称将分别为 toUpperCase-in-0 和 toUpperCase-out-0。有关用于建立绑定名称的命名约定的详细信息，请参阅 Functional Binding Names 部分。
以下是支持其他语义的简单功能应用示例：
这是一个将源语义暴露为 java.util.function.Supplier 的示例
@SpringBootApplication
public static class SourceFromSupplier {
@Bean
public Supplier<Date> date() {
return () -> new Date(12345L);
}
}
这是一个将 sink 语义 暴露为 java.util.function.Consumer 的示例
@SpringBootApplication
public static class SinkFromConsumer {
@Bean
public Consumer<String> sink() {
return System.out::println;
}
}
供应商（来源）
Function 和 Consumer 在触发调用的方式上非常直观。它们是基于发送到它们所绑定目标的数据（事件）来触发的。换句话说，它们是经典的事件驱动组件。
然而，Supplier 在触发方面自成一类。根据定义，它是数据源（起源），不订阅任何入站目的地，因此必须通过其他机制来触发。此外，还存在 Supplier 实现的问题，它可能是命令式的或响应式的，这直接关系到此类供应商的触发方式。
考虑以下示例：
@SpringBootApplication
public static class SupplierConfiguration {
@Bean
public Supplier<String> stringSupplier() {
return () -> "Hello from Supplier";
}
}
前面的 Supplier bean 在其 get() 方法被调用时会产生一个字符串。然而，谁调用这个方法以及调用的频率如何？框架提供了一个默认的轮询机制（回答了“谁？”的问题），该机制将触发 supplier 的调用，默认情况下每秒调用一次（回答了“调用的频率如何？”的问题）。换句话说，上述配置每秒生成一条消息，并且每条消息都会被发送到由 binder 暴露的 output 目标。要了解如何自定义轮询机制，请参阅 轮询配置属性 部分。
考虑一个不同的例子：
@SpringBootApplication
public static class SupplierConfiguration {
@Bean
public Supplier<Flux<String>> stringSupplier() {
return () -> Flux.fromStream(Stream.generate(new Supplier<String>() {
@Override
public String get() {
try {
Thread.sleep(1000);
return "Hello from Supplier";
} catch (Exception e) {
// ignore
}
}
})).subscribeOn(Schedulers.elastic()).share();
}
}
前面的 Supplier bean 采用了响应式编程风格。通常情况下，与命令式的 supplier 不同，它应该只被触发一次，因为其 get() 方法的调用会产生（供应）连续的消息流，而不是单个消息。
该框架能够识别编程风格的差异，并确保这样的供应商（supplier）只会被触发一次。
然而，想象一下这样的使用场景：你需要轮询某个数据源，并返回一个有限的数据流来表示结果集。响应式编程风格正是这种 Supplier 的完美机制。然而，考虑到生成的数据流的有限性，这种 Supplier 仍然需要定期调用。
考虑以下示例，它通过生成有限的数据流来模拟这种用例：
@SpringBootApplication
public static class SupplierConfiguration {
@PollableBean
public Supplier<Flux<String>> stringSupplier() {
return () -> Flux.just("hello", "bye");
}
}
bean 本身被 PollableBean 注解（@Bean 的子集）标注，从而向框架表明，尽管此类供应商的实现是响应式的，但它仍然需要被轮询。
PollableBean 中定义了一个 splittable 属性，它向该注解的后处理器发出信号，表示由被注解组件产生的结果需要被拆分，并且默认设置为 true。这意味着框架会将返回的结果拆分，并将每个项作为单独的消息发送出去。如果这不是期望的行为，你可以将其设置为 false，此时这样的供应者将直接返回生成的 Flux 而不进行拆分。
供应商与线程
正如你现在所了解的那样，与由事件触发的 Function 和 Consumer（它们有输入数据）不同，Supplier 没有任何输入，因此由不同的机制触发——poller，它可能具有不可预测的线程机制。虽然大多数情况下线程机制的细节与函数的下游执行无关，但在某些情况下，特别是在集成框架可能对线程亲和性有特定期望的情况下，这可能会带来问题。例如，Spring Cloud Sleuth 依赖于存储在线程本地中的跟踪数据。对于这些情况，我们提供了另一种通过 StreamBridge 的机制，用户可以更好地控制线程机制。你可以在 向输出发送任意数据（例如外部事件驱动源） 部分获取更多详细信息。
消费者（响应式）
Reactive Consumer 有点特殊，因为它有一个 void 返回类型，这使得框架无法引用订阅。大多数情况下，你不需要编写 Consumer<Flux<?>>，而是将其编写为 Function<Flux<?>, Mono<Void>>，并在流的最后一个操作符上调用 then 操作符。
例如：
public Function<Flux<?>, Mono<Void>> consumer() {
return flux -> flux.map(..).filter(..).then();
}
但如果你确实需要编写一个显式的 Consumer<Flux<?>>，请记得订阅传入的 Flux。
同样需要注意的是，在混合使用响应式和命令式函数时，函数组合也适用相同的规则。Spring Cloud Function 确实支持将响应式函数与命令式函数组合，但你必须注意一些限制。例如，假设你将响应式函数与命令式消费者组合在一起。这种组合的结果是一个响应式的 Consumer。然而，正如本节前面所讨论的那样，没有办法订阅这样的消费者，因此这个限制只能通过使你的消费者变为响应式并手动订阅（如前所述），或者将你的函数改为命令式来解决。
轮询配置属性
以下属性由 Spring Cloud Stream 公开，并以 spring.integration.poller. 为前缀：
fixedDelay
固定延迟
默认轮询器的固定延迟时间（以毫秒为单位）。
默认值：1000L。
maxMessagesPerPoll
默认轮询器每次轮询事件的最大消息数。
默认值：1L。
cron
cron 是一个用于在 Unix 和类 Unix 操作系统上执行计划任务的工具。它允许用户按照预定的时间间隔自动运行命令或脚本。cron 使用一个称为 crontab 的文件来存储和管理这些计划任务。
Cron 表达式的值用于 Cron 触发器。
默认值：无。
initialDelay
周期性触发器的初始延迟。
默认值：0。
timeUnit
应用于延迟值的 TimeUnit。
默认值：毫秒。
例如，--spring.integration.poller.fixed-delay=2000 将轮询间隔设置为每两秒轮询一次。
每个绑定的轮询配置
上一节展示了如何配置一个单一的默认轮询器，该轮询器将应用于所有绑定。虽然这非常适合 Spring Cloud Stream 为微服务设计的模型，其中每个微服务代表单个组件（例如，Supplier），因此默认的轮询器配置已经足够，但在某些边缘情况下，您可能会有多个组件需要不同的轮询配置。
对于这种情况，请使用按绑定方式配置轮询器。例如，假设你有一个输出绑定 supply-out-0。在这种情况下，你可以使用 spring.cloud.stream.bindings.supply-out-0.producer.poller.. 前缀为该绑定配置轮询器（例如，spring.cloud.stream.bindings.supply-out-0.producer.poller.fixed-delay=2000）。
向输出发送任意数据（例如外部事件驱动源）
在某些情况下，数据的实际来源可能来自外部的（非本地的）系统，而不是绑定器。例如，数据来源可能是一个经典的 REST 端点。我们如何将这样的数据源与 spring-cloud-stream 使用的功能机制进行桥接呢？
Spring Cloud Stream 提供了两种机制，让我们更详细地了解一下它们
在这里，对于这两个示例，我们将使用一个标准的 MVC 端点方法 delegateToSupplier，该方法绑定到根 web 上下文，通过 StreamBridge 机制将传入的请求委托给流处理。
@SpringBootApplication
@Controller
public class WebSourceApplication {
public static void main(String[] args) {
SpringApplication.run(WebSourceApplication.class, "--spring.cloud.stream.output-bindings=toStream");
}
@Autowired
private StreamBridge streamBridge;
@RequestMapping
@ResponseStatus(HttpStatus.ACCEPTED)
public void delegateToSupplier(@RequestBody String body) {
System.out.println("Sending " + body);
streamBridge.send("toStream", body);
}
}
在这里，我们自动装配了一个 StreamBridge bean，它允许我们有效地将数据发送到输出绑定，从而将非流式应用程序与 Spring Cloud Stream 桥接起来。请注意，前面的示例没有定义任何源函数（例如，Supplier bean），这使得框架没有触发器来提前创建源绑定，这在配置中包含函数 bean 的情况下是典型的。这是没有问题的，因为 StreamBridge 将在第一次调用其 send(..) 操作时启动输出绑定的创建（以及必要时目标的自动配置），并将其缓存以供后续重用（有关更多详细信息，请参见 StreamBridge 和动态目标）。
然而，如果你想在初始化（启动）时预先创建一个输出绑定，你可以利用 spring.cloud.stream.output-bindings 属性，在该属性中你可以声明你的源名称。提供的名称将用作创建源绑定的触发器。你可以使用 ; 来表示多个源（多个输出绑定）（例如，--spring.cloud.stream.output-bindings=foo;bar）。
另外，请注意 streamBridge.send(..) 方法接收一个 Object 类型的数据。这意味着你可以发送 POJO 或 Message 给它，并且它会按照与任何 Function 或 Supplier 相同的流程处理输出，从而提供与函数相同的一致性水平。这意味着输出的类型转换、分区等操作都会像来自函数产生的输出一样被处理。
使用异步发送的 StreamBridge
StreamBridge 使用了 Spring Integration 框架提供的发送机制，该框架是 Spring Cloud Stream 的核心。默认情况下，这种机制使用发送者的线程。换句话说，发送是阻塞的。虽然这在许多情况下是可以接受的，但在某些情况下，你可能希望这种发送是异步的。要实现这一点，可以在调用其中一个发送方法之前，使用 StreamBridge 的 setAsync(true) 方法。
使用异步发送进行可观测性上下文传播
当使用框架提供的可观测性支持以及支持的 Spring 框架时，打破线程边界会影响可观测性上下文的一致性，从而影响你的追踪历史。为了避免这种情况，你只需添加 Micrometer 的 context-propagation 依赖即可（见下文）。
<dependency>
<groupId>io.micrometer</groupId>
<artifactId>context-propagation</artifactId>
<version>1.1.0</version>
</dependency>
StreamBridge 和动态目的地
StreamBridge 也可以用于输出目的地事先未知的情况，类似于 Routing FROM consumer 部分中描述的用例。
让我们来看一个示例
@SpringBootApplication
@Controller
public class WebSourceApplication {
public static void main(String[] args) {
SpringApplication.run(WebSourceApplication.class, args);
}
@Autowired
private StreamBridge streamBridge;
@RequestMapping
@ResponseStatus(HttpStatus.ACCEPTED)
public void delegateToSupplier(@RequestBody String body) {
System.out.println("Sending " + body);
streamBridge.send("myDestination", body);
}
}
如你所见，前面的示例与之前的示例非常相似，除了通过 spring.cloud.stream.output-bindings 属性提供的显式绑定指令（这里没有提供）。在这里，我们将数据发送到名为 myDestination 的目标，该目标并不作为绑定存在。因此，该名称将被视为动态目标，如从消费者路由部分所述。
在前面的示例中，我们使用 ApplicationRunner 作为外部源来为流提供数据。
一个更实际的例子，其中外部数据源是 REST 端点。
@SpringBootApplication
@Controller
public class WebSourceApplication {
public static void main(String[] args) {
SpringApplication.run(WebSourceApplication.class);
}
@Autowired
private StreamBridge streamBridge;
@RequestMapping
@ResponseStatus(HttpStatus.ACCEPTED)
public void delegateToSupplier(@RequestBody String body) {
streamBridge.send("myBinding", body);
}
}
正如你在 delegateToSupplier 方法中所看到的，我们使用 StreamBridge 将数据发送到 myBinding 绑定。在这里，你还受益于 StreamBridge 的动态特性——如果 myBinding 不存在，它将自动创建并缓存，否则将使用现有的绑定。
缓存动态目的地（绑定）可能会导致内存泄漏，尤其是在存在大量动态目的地的情况下。为了提供一定程度的控制，我们为输出绑定提供了一个自回收的缓存机制，默认缓存大小为 10。这意味着如果你的动态目的地数量超过这个数值，现有的绑定可能会被回收，从而需要重新创建，这可能会导致轻微的性能下降。你可以通过 spring.cloud.stream.dynamic-destination-cache-size 属性来增加缓存大小，将其设置为所需的值。
curl -H "Content-Type: text/plain" -X POST -d "hello from the other side" http://localhost:8080/
通过展示两个例子，我们想强调该方法适用于任何类型的外部数据源。
如果你正在使用 Solace PubSub+ 绑定器，Spring Cloud Stream 已经保留了 scst_targetDestination 头信息（可通过 BinderHeaders.TARGET_DESTINATION 获取），该头信息允许将消息从其绑定的配置目的地重定向到此头信息指定的目标目的地。这使得绑定器能够管理发布到动态目的地所需的资源，从而减轻框架的负担，并避免前面 Note 中提到的缓存问题。更多信息请参见这里。
使用 StreamBridge 输出内容类型
你也可以在必要时通过以下方法签名提供特定的内容类型：public boolean send(String bindingName, Object data, MimeType outputContentType)。或者，如果你以 Message 的形式发送数据，其内容类型将被保留。
使用特定绑定器类型与 StreamBridge
Spring Cloud Stream 支持多种绑定器场景。例如，您可能从 Kafka 接收数据并将其发送到 RabbitMQ。
有关多个绑定器场景的更多信息，请参阅 绑定器 部分，特别是 类路径上的多个绑定器。
如果您计划使用 StreamBridge 并且在应用程序中配置了多个 binder，您还必须告诉 StreamBridge 使用哪个 binder。为此，send 方法还有两个变体：
public boolean send(String bindingName, @Nullable String binderType, Object data)
public boolean send(String bindingName, @Nullable String binderType, Object data, MimeType outputContentType)
如你所见，你可以提供一个额外的参数 - binderType，它告诉 BindingService 在创建动态绑定时使用哪个绑定器。
对于使用了 spring.cloud.stream.output-bindings 属性的情况，或者绑定已经在不同的 binder 下创建的情况，binderType 参数将不会产生任何效果。
使用 StreamBridge 的通道拦截器
由于 StreamBridge 使用 MessageChannel 来建立输出绑定，因此在通过 StreamBridge 发送数据时，你可以激活通道拦截器。应用程序可以决定在 StreamBridge 上应用哪些通道拦截器。Spring Cloud Stream 不会将所有检测到的通道拦截器注入到 StreamBridge 中，除非它们被标记为 @GlobalChannelInterceptor(patterns = "*")。
让我们假设你在应用程序中有以下两个不同的 StreamBridge 绑定。
streamBridge.send("foo-out-0", message);
和
streamBridge.send("bar-out-0", message);
streamBridge.send("bar-out-0", 消息);
现在，如果你想在 StreamBridge 绑定上应用一个通道拦截器，那么你可以声明以下 GlobalChannelInterceptor bean。
@Bean
@GlobalChannelInterceptor(patterns = "*")
public ChannelInterceptor customInterceptor() {
return new ChannelInterceptor() {
@Override
public Message<?> preSend(Message<?> message, MessageChannel channel) {
...
}
};
}
然而，如果您不喜欢上述的全局方法，并且希望为每个绑定设置一个专用的拦截器，那么您可以按照以下步骤操作。
@Bean
@GlobalChannelInterceptor(patterns = "foo-*")
public ChannelInterceptor fooInterceptor() {
return new ChannelInterceptor() {
@Override
public Message<?> preSend(Message<?> message, MessageChannel channel) {
...
}
};
}
和
@Bean
@GlobalChannelInterceptor(patterns = "bar-*")
public ChannelInterceptor barInterceptor() {
return new ChannelInterceptor() {
@Override
public Message<?> preSend(Message<?> message, MessageChannel channel) {
...
}
};
}
您可以根据业务需求灵活地使模式更加严格或定制化。
通过这种方法，应用程序能够决定将哪些拦截器注入到 StreamBridge 中，而不是应用所有可用的拦截器。
StreamBridge 通过 StreamOperations 接口提供了一个契约，该接口包含了 StreamBridge 的所有 send 方法。因此，应用程序可以选择使用 StreamOperations 进行自动装配。这对于单元测试代码非常方便，因为可以通过为 StreamOperations 接口提供模拟或类似机制来测试使用 StreamBridge 的代码。
响应式函数支持
由于 Spring Cloud Function 构建在 Project Reactor 之上，因此在实现 Supplier、Function 或 Consumer 时，您无需做太多工作即可受益于响应式编程模型。
例如：
@SpringBootApplication
public static class SinkFromConsumer {
@Bean
public Function<Flux<String>, Flux<String>> reactiveUpperCase() {
return flux -> flux.map(val -> val.toUpperCase());
}
}
在选择响应式或命令式编程模型时，必须理解一些重要的事项。
完全响应式还是仅仅是API？
使用响应式API并不一定意味着你可以从该API的所有响应式特性中获益。换句话说，像背压（back-pressure）和其他高级特性只有在与兼容系统（如Reactive Kafka绑定器）一起工作时才会生效。如果你使用的是普通的Kafka、RabbitMQ或任何其他非响应式绑定器，你只能从响应式API本身的便利性中获益，而不能享受到其高级特性，因为实际的流来源或目标并不是响应式的。
错误处理和重试
在本手册中，你会多次看到关于基于框架的错误处理、重试和其他特性以及与之相关的配置属性的参考。重要的是要理解，这些特性仅影响命令式函数，而在响应式函数中你不应该抱有相同的期望。原因如下……响应式函数和命令式函数之间存在根本性的差异。命令式函数是一个消息处理器，框架在接收到每条消息时都会调用它。因此，对于N条消息，该函数将被调用N次，正因为如此，我们可以包装该函数并添加额外的功能，如错误处理、重试等。响应式函数是一个初始化函数。它只会被调用一次，以获取用户提供的Flux/Mono的引用，并将其与框架提供的引用连接起来。之后，我们（框架）对流的可见性和控制权完全丧失。因此，在响应式函数中，你必须依赖响应式API的丰富功能来处理错误和重试（例如doOnError()、.onError*()等）。
函数组合
使用函数式编程模型，你还可以受益于函数组合，即可以从一组简单的函数中动态组合出复杂的处理程序。举个例子，让我们将以下函数 bean 添加到上面定义的应用程序中
@Bean
public Function<String, String> wrapInQuotes() {
return s -> "\"" + s + "\"";
}
并修改 spring.cloud.function.definition 属性，以反映你打算从 toUpperCase 和 wrapInQuotes 组合一个新函数的意图。为此，Spring Cloud Function 依赖于 |（管道）符号。因此，为了完成我们的示例，我们的属性现在将如下所示：
--spring.cloud.function.definition=toUpperCase|wrapInQuotes
Spring Cloud Function 提供的函数式组合支持的一个巨大好处是，你可以组合响应式和命令式函数。
组合的结果是一个单一的函数，正如你所猜测的，它可能有一个非常长且相当晦涩的名称（例如，foo|bar|baz|xyz. . .），这在涉及其他配置属性时会带来很大的不便。这正是 Functional Binding Names 部分中描述的描述性绑定名称功能可以派上用场的地方。
例如，如果我们想给 toUpperCase|wrapInQuotes 一个更具描述性的名称，我们可以通过以下属性来实现：spring.cloud.stream.function.bindings.toUpperCase|wrapInQuotes-in-0=quotedUpperCaseInput，这样其他配置属性就可以引用该绑定名称（例如，spring.cloud.stream.bindings.quotedUpperCaseInput.destination=myDestination）。
函数组合与横切关注点
函数组合有效地允许您通过将复杂性分解为一组简单且易于管理/测试的组件来应对复杂性，这些组件在运行时仍然可以作为一个整体表示。但这并不是唯一的好处。
你也可以使用组合来解决某些横切非功能性关注点，例如内容增强。例如，假设你有一个传入的消息，可能缺少某些头信息，或者某些头信息的状态与你的业务函数所期望的不完全一致。现在，你可以实现一个单独的函数来处理这些问题，然后将其与主要的业务函数组合在一起。
让我们来看一个例子
@SpringBootApplication
public class DemoStreamApplication {
public static void main(String[] args) {
SpringApplication.run(DemoStreamApplication.class,
"--spring.cloud.function.definition=enrich|echo",
"--spring.cloud.stream.function.bindings.enrich|echo-in-0=input",
"--spring.cloud.stream.bindings.input.destination=myDestination",
"--spring.cloud.stream.bindings.input.group=myGroup");
}
@Bean
public Function<Message<String>, Message<String>> enrich() {
return message -> {
Assert.isTrue(!message.getHeaders().containsKey("foo"), "Should NOT contain 'foo' header");
return MessageBuilder.fromMessage(message).setHeader("foo", "bar").build();
};
}
@Bean
public Function<Message<String>, Message<String>> echo() {
return message -> {
Assert.isTrue(message.getHeaders().containsKey("foo"), "Should contain 'foo' header");
System.out.println("Incoming message " + message);
return message;
};
}
}
虽然这个示例很简单，但它展示了如何通过一个函数来丰富传入的 Message，添加额外的头信息（非功能性关注点），从而使另一个函数——echo——能够从中受益。echo 函数保持了简洁，仅专注于业务逻辑。你还可以看到 spring.cloud.stream.function.bindings 属性的使用，它简化了组合绑定的名称。
具有多个输入和输出参数的函数
从 3.0 版本开始，spring-cloud-stream 提供了对具有多个输入和/或多个输出（返回值）的函数的支持。这实际上意味着什么？它针对的是哪些类型的用例？
-
大数据：想象一下，你正在处理的数据源非常杂乱无章，包含各种类型的数据元素（例如，订单、交易等），你需要有效地对其进行整理。
-
数据聚合：另一个用例可能要求你将来自两个或多个输入 _streams_ 的数据元素进行合并。
上述内容描述了您可能需要使用单个函数来接受和/或生成多个数据流的一些用例。这正是我们在这里所针对的用例类型。
另外，请注意这里对流概念的不同强调。假设是，只有当这些函数能够访问实际的数据流（而不是单个元素）时，它们才有价值。因此，我们依赖于 Project Reactor 提供的抽象（即 Flux 和 Mono），这些抽象已经作为 spring-cloud-functions 引入的依赖项的一部分存在于类路径中。
另一个重要的方面是多个输入和输出的表示。虽然 Java 提供了多种不同的抽象来表示多个某物，但这些抽象1) 是无界的，2) 缺乏元数，3) 缺乏类型信息，而这些在本上下文中都是重要的。举个例子，让我们看看 Collection 或数组，它们只允许我们描述多个单一类型或将所有内容上转型为 Object，这影响了 spring-cloud-stream 的透明类型转换功能等等。
为了满足所有这些需求，初始支持依赖于使用由 Project Reactor 提供的另一种抽象——元组（Tuples）的签名。然而，我们正在努力允许更灵活的签名。
请参考 绑定与绑定名称 部分，了解用于建立此类应用程序所使用的 绑定名称 的命名约定。
让我们看几个示例：
@SpringBootApplication
public class SampleApplication {
@Bean
public Function<Tuple2<Flux<String>, Flux<Integer>>, Flux<String>> gather() {
return tuple -> {
Flux<String> stringStream = tuple.getT1();
Flux<String> intStream = tuple.getT2().map(i -> String.valueOf(i));
return Flux.merge(stringStream, intStream);
};
}
}
上面的示例展示了一个函数，它接受两个输入（第一个为 String 类型，第二个为 Integer 类型），并生成一个 String 类型的输出。
因此，对于上面的例子，两个输入绑定将命名为 gather-in-0 和 gather-in-1，为了保持一致性，输出绑定也遵循相同的命名规则，命名为 gather-out-0。
了解这一点将允许您设置特定于绑定的属性。例如，以下内容将覆盖 gather-in-0 绑定的 content-type：
--spring.cloud.stream.bindings.gather-in-0.content-type=text/plain
@SpringBootApplication
public class SampleApplication {
@Bean
public static Function<Flux<Integer>, Tuple2<Flux<String>, Flux<String>>> scatter() {
return flux -> {
Flux<Integer> connectedFlux = flux.publish().autoConnect(2);
UnicastProcessor even = UnicastProcessor.create();
UnicastProcessor odd = UnicastProcessor.create();
Flux<Integer> evenFlux = connectedFlux.filter(number -> number % 2 == 0).doOnNext(number -> even.onNext("EVEN: " + number));
Flux<Integer> oddFlux = connectedFlux.filter(number -> number % 2 != 0).doOnNext(number -> odd.onNext("ODD: " + number));
return Tuples.of(Flux.from(even).doOnSubscribe(x -> evenFlux.subscribe()), Flux.from(odd).doOnSubscribe(x -> oddFlux.subscribe()));
};
}
}
上面的示例与之前的示例有些相反，它演示了一个函数，该函数接受一个类型为 Integer 的单个输入，并产生两个输出（两者都是 String 类型）。
因此，对于上面的示例，输入绑定是 scatter-in-0，输出绑定是 scatter-out-0 和 scatter-out-1。
你可以使用以下代码进行测试：
@Test
public void testSingleInputMultiOutput() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
SampleApplication.class))
.run("--spring.cloud.function.definition=scatter")) {
InputDestination inputDestination = context.getBean(InputDestination.class);
OutputDestination outputDestination = context.getBean(OutputDestination.class);
for (int i = 0; i < 10; i++) {
inputDestination.send(MessageBuilder.withPayload(String.valueOf(i).getBytes()).build());
}
int counter = 0;
for (int i = 0; i < 5; i++) {
Message<byte[]> even = outputDestination.receive(0, 0);
assertThat(even.getPayload()).isEqualTo(("EVEN: " + String.valueOf(counter++)).getBytes());
Message<byte[]> odd = outputDestination.receive(0, 1);
assertThat(odd.getPayload()).isEqualTo(("ODD: " + String.valueOf(counter++)).getBytes());
}
}
}
单一应用中的多个功能
可能还需要在单个应用程序中对多个消息处理程序进行分组。您可以通过定义多个函数来实现这一点。
@SpringBootApplication
public class SampleApplication {
@Bean
public Function<String, String> uppercase() {
return value -> value.toUpperCase();
}
@Bean
public Function<String, String> reverse() {
return value -> new StringBuilder(value).reverse().toString();
}
}
在上面的示例中，我们有一个配置，定义了两个函数 uppercase 和 reverse。首先，如前所述，我们需要注意到这里存在冲突（多个函数），因此我们需要通过提供 spring.cloud.function.definition 属性来解决这个问题，该属性指向我们想要绑定的实际函数。不过在这里，我们将使用 ; 分隔符来指向这两个函数（参见下面的测试用例）。
与具有多个输入/输出的函数一样，请参考[[绑定和绑定名称]](#Binding and Binding names)部分，以了解用于建立此类应用程序使用的绑定名称的命名约定。
你可以使用以下代码进行测试：
@Test
public void testMultipleFunctions() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
ReactiveFunctionConfiguration.class))
.run("--spring.cloud.function.definition=uppercase;reverse")) {
InputDestination inputDestination = context.getBean(InputDestination.class);
OutputDestination outputDestination = context.getBean(OutputDestination.class);
Message<byte[]> inputMessage = MessageBuilder.withPayload("Hello".getBytes()).build();
inputDestination.send(inputMessage, "uppercase-in-0");
inputDestination.send(inputMessage, "reverse-in-0");
Message<byte[]> outputMessage = outputDestination.receive(0, "uppercase-out-0");
assertThat(outputMessage.getPayload()).isEqualTo("HELLO".getBytes());
outputMessage = outputDestination.receive(0, "reverse-out-0");
assertThat(outputMessage.getPayload()).isEqualTo("olleH".getBytes());
}
}
批量消费者
当使用支持批量监听器的 MessageChannelBinder 时，如果为消费者绑定启用了该功能，你可以将 spring.cloud.stream.bindings.<binding-name>.consumer.batch-mode 设置为 true，以便将整个批量的消息以 List 的形式传递给函数。
@Bean
public Function<List<Person>, Person> findFirstPerson() {
return persons -> persons.get(0);
}
批量类型转换
与单个消息消费者的类型转换类似，批处理要求将批处理中的每个消息都转换为请求的类型。例如，在前面的示例中，这个类型是 Person。
同样重要的是要理解，批处理中每条消息的头部信息是单独提供给代表整个批处理的 Message 的 MessageHeaders 的。这些消息及其对应的批处理头部信息是由各自的绑定器创建的，它们的结构可能有所不同。因此，你应该参考绑定器文档以了解批处理头部信息的结构。对于 Kafka 和 Rabbit，你可以分别搜索 amqp_batchedHeaders 和 kafka_batchConvertedHeaders。
简而言之，如果你有一个表示包含 5 个有效载荷的批次的消息，那么该消息将包含一组标头，其中每个标头对应具有相同索引的有效载荷。
然而，如果某个特定的负载无法转换会发生什么？在单条消息的场景中，我们只需返回 null 并使用未转换的消息调用你的方法，这要么会导致异常，要么允许你处理原始消息，具体取决于你的函数签名。
在批处理的情况下，情况会稍微复杂一些。对于未转换的有效载荷返回 null 会有效地减少批处理的大小。例如，如果原始批处理包含 5 条消息，其中有 2 条未能转换，那么转换后的批处理将只包含 3 条消息。这可能是可以接受的，但对应的批处理头信息呢？仍然会有 5 个头信息，因为它们在绑定器形成初始批处理时就已经创建了。这种不一致性使得很难将头信息与它们对应的有效载荷关联起来。
为了解决这个问题，我们提供了 MessageConverterHelper 接口。
public interface MessageConverterHelper {
/**
* This method will be called by the framework in cases when a message failed to convert.
* It allows you to signal to the framework if such failure should be considered fatal or not.
*
* @param message failed message
* @return true if conversion failure must be considered fatal.
*/
default boolean shouldFailIfCantConvert(Message<?> message) {
return false;
}
/**
* This method will be called by the framework in cases when a single message within batch of messages failed to convert.
* It provides a place for providing post-processing logic before message converter returns.
*
* @param message failed message.
* @param index index of failed message within the batch
*/
default void postProcessBatchMessageOnFailure(Message<?> message, int index) {
}
}
如果实现了这个接口，当特定负载无法转换时，框架的消息转换器逻辑将调用它来对批量消息进行后处理。
Kafka 和 RabbitMQ 的默认实现会自动移除相应的批量头信息，以维护批量负载与其头信息之间的关联性。但是，如果你需要为此类情况添加自定义行为，可以提供自己的实现并将其注册为一个 bean。
此外，该接口还提供了一种方法，允许对转换失败进行更确定性的处理。默认情况下，该方法返回 false，但如果你希望在发生转换错误时使整个流程失败，可以自定义其实现。
批处理生产者
你也可以在生产者端使用批处理的概念，通过返回一个消息集合来实现，这实际上提供了一种相反的效果，即集合中的每条消息将由绑定器单独发送。
考虑以下函数：
@Bean
public Function<String, List<Message<String>>> batch() {
return p -> {
List<Message<String>> list = new ArrayList<>();
list.add(MessageBuilder.withPayload(p + ":1").build());
list.add(MessageBuilder.withPayload(p + ":2").build());
list.add(MessageBuilder.withPayload(p + ":3").build());
list.add(MessageBuilder.withPayload(p + ":4").build());
return list;
};
}
返回列表中的每条消息将单独发送，最终会有四条消息发送到输出目的地。
Spring Integration 流作为函数
当你实现一个函数时，可能会遇到符合企业集成模式（EIP）类别的复杂需求。这些需求最好通过使用如Spring Integration（SI）这样的框架来处理，它是 EIP 的参考实现。
幸运的是，Spring Integration（SI）已经提供了通过集成流作为网关将集成流暴露为函数的支持。考虑以下示例：
@SpringBootApplication
public class FunctionSampleSpringIntegrationApplication {
public static void main(String[] args) {
SpringApplication.run(FunctionSampleSpringIntegrationApplication.class, args);
}
@Bean
public IntegrationFlow uppercaseFlow() {
return IntegrationFlow.from(MessageFunction.class, spec -> spec.beanName("uppercase"))
.<String, String>transform(String::toUpperCase)
.log(LoggingHandler.Level.WARN)
.bridge()
.get();
}
public interface MessageFunction extends Function<Message<String>, Message<String>> {
}
}
对于那些熟悉 Spring Integration（SI）的人来说，可以看到我们定义了一个类型为 IntegrationFlow 的 bean，其中我们声明了一个希望作为 Function<String, String> 暴露的集成流（使用 SI DSL），并将其命名为 uppercase。MessageFunction 接口允许我们显式声明输入和输出的类型，以便进行适当的类型转换。有关类型转换的更多信息，请参见 [[内容类型协商]](#内容类型协商) 部分。
要接收原始输入，你可以使用 from(Function.class, …)。
生成的函数被绑定到目标绑定器暴露的输入和输出目的地。
请参考 [[绑定与绑定名称]](#Binding and Binding names) 部分，以了解用于建立此类应用程序所使用的绑定名称的命名约定。
有关 Spring Integration 和 Spring Cloud Stream 在函数式编程模型方面的互操作性的更多详细信息，你可能会发现这篇文章非常有趣，因为它深入探讨了通过结合 Spring Integration 和 Spring Cloud Stream/Functions 的最佳实践可以应用的各种模式。
使用轮询消费者
概述
在使用轮询消费者时，你可以按需轮询 PollableMessageSource。为了定义轮询消费者的绑定，你需要提供 spring.cloud.stream.pollable-source 属性。
考虑以下轮询消费者绑定的示例：
--spring.cloud.stream.pollable-source=myDestination
在前面示例中的可轮询源名称 myDestination 将导致绑定名称为 myDestination-in-0，以保持与函数式编程模型的一致性。
在前面的示例中，你可能会这样使用轮询的消费者：
@Bean
public ApplicationRunner poller(PollableMessageSource destIn, MessageChannel destOut) {
return args -> {
while (someCondition()) {
try {
if (!destIn.poll(m -> {
String newPayload = ((String) m.getPayload()).toUpperCase();
destOut.send(new GenericMessage<>(newPayload));
})) {
Thread.sleep(1000);
}
}
catch (Exception e) {
// handle failure
}
}
};
}
一种更少手动操作且更符合 Spring 风格的方式是配置一个定时任务 bean。例如，
@Scheduled(fixedDelay = 5_000)
public void poll() {
System.out.println("Polling...");
this.source.poll(m -> {
System.out.println(m.getPayload());
}, new ParameterizedTypeReference<Foo>() { });
}
PollableMessageSource.poll() 方法接受一个 MessageHandler 参数（通常是一个 lambda 表达式，如这里所示）。如果消息被接收并成功处理，它将返回 true。
与消息驱动消费者一样，如果 MessageHandler 抛出异常，消息将被发布到错误通道，如 [错误处理](overview-error-handling.md) 中所述。
通常情况下，poll() 方法会在 MessageHandler 退出时确认消息。如果该方法异常退出，消息会被拒绝（不会重新入队），但请参阅处理错误。你可以通过自行负责确认来覆盖该行为，如下例所示：
@Bean
public ApplicationRunner poller(PollableMessageSource dest1In, MessageChannel dest2Out) {
return args -> {
while (someCondition()) {
if (!dest1In.poll(m -> {
StaticMessageHeaderAccessor.getAcknowledgmentCallback(m).noAutoAck();
// e.g. hand off to another thread which can perform the ack
// or acknowledge(Status.REQUEUE)
})) {
Thread.sleep(1000);
}
}
};
}
你必须在某个时刻对消息进行 ack（或 nack），以避免资源泄漏。
一些消息系统（例如 Apache Kafka）在日志中维护一个简单的偏移量。如果消息传递失败并使用 StaticMessageHeaderAccessor.getAcknowledgmentCallback(m).acknowledge(Status.REQUEUE); 重新排队，那么任何后续成功确认的消息都会被重新传递。
还有一个重载的 poll 方法，其定义如下：
poll(MessageHandler handler, ParameterizedTypeReference<?> type)
type 是一个转换提示，允许对传入的消息负载进行转换，如下例所示：
boolean result = pollableSource.poll(received -> {
Map<String, Foo> payload = (Map<String, Foo>) received.getPayload();
...
}, new ParameterizedTypeReference<Map<String, Foo>>() {});
处理错误
默认情况下，为可轮询的源配置了一个错误通道；如果回调抛出异常，会将一个 ErrorMessage 发送到错误通道（<destination>.<group>.errors）；该错误通道也会桥接到全局的 Spring Integration errorChannel。
你可以使用 @ServiceActivator 订阅任意一个错误通道来处理错误；如果没有订阅，错误将仅被记录，并且消息将被确认成功。如果错误通道的服务激活器抛出异常，消息将被拒绝（默认情况下），并且不会被重新投递。如果服务激活器抛出 RequeueCurrentMessageException 异常，消息将在代理处重新排队，并在后续轮询时再次被检索。
如果监听器直接抛出 RequeueCurrentMessageException，如上所述，消息将被重新排队，并且不会发送到错误通道。