响应式流支持

Spring Integration 在框架的某些地方以及从不同方面为 响应式流 交互提供支持。我们将在此处讨论其中大部分内容，并在必要时提供指向目标章节的适当链接以获取详细信息。

前言

回顾一下，Spring Integration 扩展了 Spring 编程模型以支持众所周知的企业集成模式。Spring Integration 支持基于 Spring 的应用程序中的轻量级消息传递，并通过声明式适配器支持与外部系统的集成。Spring Integration 的主要目标是为构建企业集成解决方案提供一个简单的模型，同时保持对可维护、可测试代码至关重要的关注点分离。此目标在目标应用程序中通过 message、channel 和 endpoint 等一流公民实现，这些公民允许我们构建集成流（管道），其中（在大多数情况下）一个端点将消息生成到通道中，供另一个端点消费。通过这种方式，我们将集成交互模型与目标业务逻辑区分开来。这里的关键部分是中间的通道：流行为取决于其实现，而端点保持不变。

另一方面，响应式流是具有非阻塞背压的异步流处理标准。响应式流的主要目标是管理跨异步边界（例如将元素传递给另一个线程或线程池）的流数据交换，同时确保接收方不会被迫缓冲任意数量的数据。换句话说，背压是此模型不可或缺的一部分，以便允许在线程之间进行调解的队列是有界的。响应式流实现（例如 Project Reactor）的目的是在流应用程序的整个处理图中保留这些优点和特性。响应式流库的最终目标是以透明和流畅的方式为目标应用程序提供类型、运算符集和支持 API，就像使用可用的编程语言结构一样，但最终的解决方案不像正常的函数链调用那样是命令式的。它分为两个阶段：定义和执行，执行在稍后订阅最终的响应式发布者期间发生，并且数据需求从定义的底部推到顶部，根据需要应用背压——我们请求当前可以处理的尽可能多的事件。响应式应用程序看起来像一个 "stream"，或者正如我们在 Spring Integration 术语中习惯的那样——"flow"。事实上，自 Java 9 以来的 Reactive Streams SPI 在 java.util.concurrent.Flow 类中呈现。

从这里看，当我们在端点上应用一些响应式框架运算符时，Spring Integration 流似乎非常适合编写响应式流应用程序，但实际上问题要广泛得多，我们需要记住并非所有端点（例如 JdbcMessageHandler）都可以透明地在响应式流中处理。当然，Spring Integration 中响应式流支持的主要目标是允许整个过程完全响应式、按需启动并准备好背压。在通道适配器的目标协议和系统提供响应式流交互模型之前，这将是不可能的。在下面的部分中，我们将描述 Spring Integration 中为开发响应式应用程序同时保留集成流结构而提供的组件和方法。

Spring Integration 中的所有响应式流交互都使用 Project Reactor 类型（例如 Mono 和 Flux）实现。

消息网关

与响应式流交互的最简单点是 @MessagingGateway，我们只需将网关方法的返回类型设置为 Mono<?> ——当对返回的 Mono 实例进行订阅时，网关方法调用背后的整个集成流将执行。有关更多信息，请参阅 Reactor Mono。框架内部对完全基于响应式流兼容协议的入站网关也使用了类似的 Mono 回复方法（有关更多信息，请参阅下面的 响应式通道适配器）。发送和接收操作被包装在一个 Mono.defer() 中，并在 replyChannel 头部可用时链式评估回复。这样，特定响应式协议（例如 Netty）的入站组件将作为 Spring Integration 上执行的响应式流的订阅者和发起者。如果请求负载是响应式类型，最好在响应式流定义中处理它，将进程推迟到发起者订阅。为此，处理程序方法也必须返回响应式类型。有关更多信息，请参阅下一节。

响应式回复负载

当产生回复的 MessageHandler 返回一个响应式类型负载作为回复消息时，它会以异步方式处理，其中为 outputChannel 提供了常规的 MessageChannel 实现（async 必须设置为 true），并在输出通道是 ReactiveStreamsSubscribableChannel 实现（例如 FluxMessageChannel）时通过按需订阅进行扁平化。对于标准命令式 MessageChannel 用例，如果回复负载是**多值**发布者（有关更多信息，请参阅 ReactiveAdapter.isMultiValue()），它将被包装到 Mono.just() 中。结果是，Mono 必须由下游显式订阅或由 FluxMessageChannel 下游扁平化。对于 outputChannel 的 ReactiveStreamsSubscribableChannel，无需担心返回类型和订阅；一切都在框架内部顺利处理。

有关更多信息，请参阅 异步服务激活器。

另请参阅 Kotlin 协程 以获取更多信息。

FluxMessageChannel 和 ReactiveStreamsConsumer

FluxMessageChannel 是 MessageChannel 和 Publisher<Message<?>> 的组合实现。Flux 作为热源，在内部创建，用于从 send() 实现中汇集传入消息。Publisher.subscribe() 实现委托给该内部 Flux。此外，为了按需上游消费，FluxMessageChannel 提供了 ReactiveStreamsSubscribableChannel 契约的实现。为该通道提供的任何上游 Publisher（例如，参见下面的源轮询通道适配器和拆分器）在该通道准备好订阅时自动订阅。来自此委托发布者的事件汇集到上述内部 Flux 中。

FluxMessageChannel 的消费者必须是 org.reactivestreams.Subscriber 实例，以遵循响应式流契约。幸运的是，Spring Integration 中的所有 MessageHandler 实现也实现了 Project Reactor 的 CoreSubscriber。多亏了中间的 ReactiveStreamsConsumer 实现，整个集成流配置对目标开发人员保持透明。在这种情况下，流行为从命令式推送模型变为响应式拉取模型。ReactiveStreamsConsumer 还可以用于使用 IntegrationReactiveUtils 将任何 MessageChannel 转换为响应式源，从而使集成流部分响应式。

有关更多信息，请参阅 FluxMessageChannel。

从版本 5.5 开始，ConsumerEndpointSpec 引入了 reactive() 选项，使流中的端点成为 ReactiveStreamsConsumer，而与输入通道无关。可选的 Function<? super Flux<Message<?>>, ? extends Publisher<Message<?>>> 可以用于通过 Flux.transform() 操作自定义输入通道的源 Flux，例如使用 publishOn()、doOnNext()、retry() 等。此功能通过其 reactive() 属性在所有消息传递注解（@ServiceActivator、@Splitter 等）中表示为 @Reactive 子注解。

源轮询通道适配器

通常，SourcePollingChannelAdapter 依赖于由 TaskScheduler 启动的任务。轮询触发器是根据提供的选项构建的，用于定期调度任务以轮询目标数据源或事件。当 outputChannel 是 ReactiveStreamsSubscribableChannel 时，相同的 Trigger 用于确定下一次执行时间，但 SourcePollingChannelAdapter 不调度任务，而是根据 nextExecutionTime 值的 Flux.generate() 和从上一步的持续时间的 Mono.delay() 创建一个 Flux<Message<?>>。然后使用 Flux.flatMapMany() 轮询 maxMessagesPerPoll 并将其汇集到输出 Flux 中。此生成器 Flux 由提供的 ReactiveStreamsSubscribableChannel 订阅，遵循下游的背压。从版本 5.5 开始，当 maxMessagesPerPoll == 0 时，源根本不被调用，并且 flatMapMany() 通过 Mono.empty() 结果立即完成，直到 maxMessagesPerPoll 在稍后时间（例如通过控制总线）更改为非零值。这样，任何 MessageSource 实现都可以转换为响应式热源。

有关更多信息，请参阅 轮询消费者。

事件驱动通道适配器

MessageProducerSupport 是事件驱动通道适配器的基类，通常，其 sendMessage(Message<?>) 用作生产驱动 API 中的监听器回调。当消息生产者实现构建消息的 Flux 而不是基于监听器的功能时，此回调也可以轻松插入 doOnNext() Reactor 运算符。事实上，当消息生产者的 outputChannel 不是 ReactiveStreamsSubscribableChannel 时，框架中就是这样做的。然而，为了改善最终用户体验并允许更多准备好背压的功能，MessageProducerSupport 提供了一个 subscribeToPublisher(Publisher<? extends Message<?>>) API，供目标实现使用，当 Publisher<Message<?>>> 是目标系统的数据源时。通常，它在 doStart() 实现中用于调用目标驱动 API 以获取源数据的 Publisher。建议将响应式 MessageProducerSupport 实现与 FluxMessageChannel 作为 outputChannel 结合使用，以实现按需订阅和下游事件消费。当取消对 Publisher 的订阅时，通道适配器进入停止状态。对此通道适配器调用 stop() 会完成从源 Publisher 的生产。通道适配器可以重新启动，并自动订阅到新创建的源 Publisher。

消息源到响应式流

从版本 5.3 开始，提供了 ReactiveMessageSourceProducer。它结合了提供的 MessageSource 和事件驱动的生产到配置的 outputChannel 中。它内部将 MessageSource 包装到重复重新订阅的 Mono 中，生成一个 Flux<Message<?>>，用于订阅到上述的 subscribeToPublisher(Publisher<? extends Message<?>>)。此 Mono 的订阅使用 Schedulers.boundedElastic() 完成，以避免目标 MessageSource 中可能发生的阻塞。当消息源返回 null（没有数据可拉取）时，Mono 将进入 repeatWhenEmpty() 状态，并根据订阅者上下文中的 IntegrationReactiveUtils.DELAY_WHEN_EMPTY_KEY Duration 条目进行后续重新订阅的 delay。默认情况下，它是 1 秒。如果 MessageSource 在消息头中产生带有 IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK 信息的消息，则在原始 Mono 的 doOnSuccess() 中确认（如果需要），并在下游流抛出带有要拒绝的失败消息的 MessagingException 时在 doOnError() 中拒绝。此 ReactiveMessageSourceProducer 可用于任何需要将轮询通道适配器的功能转换为任何现有 MessageSource<?> 实现的响应式、按需解决方案的用例。

拆分器和聚合器

当 AbstractMessageSplitter 获得一个 Publisher 来执行其逻辑时，该过程会自然地遍历 Publisher 中的项目，将它们映射到消息，以便发送到 outputChannel。如果此通道是 ReactiveStreamsSubscribableChannel，则 Publisher 的 Flux 包装器将按需从该通道订阅，并且此拆分器行为更像一个 flatMap Reactor 运算符，我们将传入事件映射到多值输出 Publisher。当整个集成流在拆分器之前和之后都使用 FluxMessageChannel 构建时，这将最有意义，从而使 Spring Integration 配置与响应式流要求及其事件处理运算符保持一致。对于常规通道，Publisher 将转换为 Iterable，用于标准的迭代和生产拆分逻辑。

FluxAggregatorMessageHandler 是特定响应式流逻辑实现的另一个示例，可以将其视为 Project Reactor 中的 "reactive operator"。它基于 Flux.groupBy() 和 Flux.window()（或 buffer()）运算符。传入的消息汇入在创建 FluxAggregatorMessageHandler 时启动的 Flux.create() 中，使其成为热源。此 Flux 由 ReactiveStreamsSubscribableChannel 按需订阅，或者当 outputChannel 不是响应式时，直接在 FluxAggregatorMessageHandler.start() 中订阅。当整个集成流在此组件之前和之后都使用 FluxMessageChannel 构建时，此 MessageHandler 发挥其作用，使整个逻辑准备好背压。

有关更多信息，请参阅 流和 Flux 拆分 和 Flux 聚合器。

Java DSL

Java DSL 中的 IntegrationFlow 可以从任何 Publisher 实例开始（请参阅 IntegrationFlow.from(Publisher<Message<T>>)）。此外，通过 IntegrationFlowBuilder.toReactivePublisher() 运算符，IntegrationFlow 可以转换为响应式热源。在这两种情况下，内部都使用 FluxMessageChannel；它可以根据其 ReactiveStreamsSubscribableChannel 契约订阅入站 Publisher，并且它本身是下游订阅者的 Publisher<Message<?>>。通过动态 IntegrationFlow 注册，我们可以实现强大的逻辑，将响应式流与此集成流结合起来，与 Publisher 进行桥接。

从版本 5.5.6 开始，提供了 toReactivePublisher(boolean autoStartOnSubscribe) 运算符变体，用于控制返回的 Publisher<Message<?>> 背后的整个 IntegrationFlow 的生命周期。通常，响应式发布者的订阅和消费发生在稍后的运行时阶段，而不是在响应式流组合期间，甚至不是在 ApplicationContext 启动期间。为了避免在 Publisher<Message<?>> 订阅点上进行 IntegrationFlow 生命周期管理的样板代码，并为了更好的最终用户体验，引入了带有 autoStartOnSubscribe 标志的新运算符。它将（如果为 true）IntegrationFlow 及其组件标记为 autoStartup = false，因此 ApplicationContext 不会自动启动流中消息的生产和消费。相反，IntegrationFlow 的 start() 是从内部 Flux.doOnSubscribe() 启动的。无论 autoStartOnSubscribe 值如何，流都从 Flux.doOnCancel() 和 Flux.doOnTerminate() 停止——如果没有任何东西可以消费消息，那么生产消息就没有意义。

对于完全相反的用例，当 IntegrationFlow 应该调用响应式流并在完成后继续时，在 IntegrationFlowDefinition 中提供了 fluxTransform() 运算符。此时的流将转换为 FluxMessageChannel，该通道传播到提供的 fluxFunction，在 Flux.transform() 运算符中执行。函数的結果被包装到 Mono<Message<?>> 中，用于扁平化到输出 Flux 中，该 Flux 由另一个 FluxMessageChannel 订阅以进行下游流。

有关更多信息，请参阅 Java DSL 章​​节。

ReactiveMessageHandler

从版本 5.3 开始，框架原生支持 ReactiveMessageHandler。此类型的消息处理程序专为响应式客户端设计，这些客户端返回响应式类型以按需订阅低级操作执行，并且不提供任何回复数据以继续响应式流组合。当 ReactiveMessageHandler 在命令式集成流中使用时，handleMessage() 结果在返回后立即订阅，因为此类流中没有响应式流组合来遵守背压。在这种情况下，框架将此 ReactiveMessageHandler 包装到 ReactiveMessageHandlerAdapter 中——一个 MessageHandler 的普通实现。然而，当 ReactiveStreamsConsumer 参与流时（例如，当要消费的通道是 FluxMessageChannel 时），此 ReactiveMessageHandler 将与整个响应式流结合，使用 flatMap() Reactor 运算符，以在消费期间遵守背压。

一个开箱即用的 ReactiveMessageHandler 实现是用于出站通道适配器的 ReactiveMongoDbStoringMessageHandler。有关更多信息，请参阅 MongoDB 响应式通道适配器。

从版本 6.1 开始，IntegrationFlowDefinition 公开了一个方便的 handleReactive(ReactiveMessageHandler) 终端运算符。任何 ReactiveMessageHandler 实现（即使是使用 Mono API 的简单 lambda）都可以用于此运算符。框架会自动订阅返回的 Mono<Void>。以下是此运算符可能配置的简单示例

@Bean
public IntegrationFlow wireTapFlow1() {
    return IntegrationFlow.from("tappedChannel1")
            .wireTap("tapChannel", wt -> wt.selector(m -> m.getPayload().equals("foo")))
            .handleReactive((message) -> Mono.just(message).log().then());
}

此运算符的重载版本接受 Consumer<GenericEndpointSpec<ReactiveMessageHandlerAdapter>> 来定制围绕提供的 ReactiveMessageHandler 的消费者端点。

此外，还提供了基于 ReactiveMessageHandlerSpec 的变体。在大多数情况下，它们用于特定协议的通道适配器实现。请参阅下一节，其中包含指向具有相应响应式通道适配器的目标技术的链接。

响应式通道适配器

当集成目标协议提供响应式流解决方案时，在 Spring Integration 中实现通道适配器将变得非常简单。

入站、事件驱动的通道适配器实现是将请求（如有必要）包装到延迟的 Mono 或 Flux 中，并且仅当协议组件启动对监听器方法返回的 Mono 的订阅时才执行 send（并生成回复，如有）。通过这种方式，我们将响应式流解决方案精确地封装在此组件中。当然，订阅到输出通道的下游集成流应遵循响应式流规范，并以按需、背压就绪的方式执行。

这并非总是可用的，因为集成流中使用的 MessageHandler 处理器本身的性质（或当前实现）所限。当没有响应式实现时，可以使用线程池和队列或 FluxMessageChannel（参见上文）在集成端点之前和之后处理此限制。

响应式**事件驱动**入站通道适配器示例

public class CustomReactiveMessageProducer extends MessageProducerSupport {

    private final CustomReactiveSource customReactiveSource;

    public CustomReactiveMessageProducer(CustomReactiveSource customReactiveSource) {
        Assert.notNull(customReactiveSource, "'customReactiveSource' must not be null");
        this.customReactiveSource = customReactiveSource;
    }

    @Override
    protected void doStart() {
        Flux<Message<?>> messageFlux =
            this.customReactiveSource
                .map(event - >
                    MessageBuilder
                    .withPayload(event.getBody())
                    .setHeader(MyReactiveHeaders.SOURCE_NAME, event.getSourceName())
                    .build());

        subscribeToPublisher(messageFlux);
    }
}

用法如下

public class MainFlow {
  @Autowired
  private CustomReactiveMessageProducer customReactiveMessageProducer;

  @Bean
  public IntegrationFlow buildFlow() {
     return IntegrationFlow.from(customReactiveMessageProducer)
        .channel(outputChannel)
        .get();
  }
}

或者以声明的方式

public class MainFlow {
  @Bean
  public IntegrationFlow buildFlow() {
     return IntegrationFlow.from(new CustomReactiveMessageProducer(new CustomReactiveSource()))
        .handle(outputChannel)
        .get();
  }
}

或者甚至不使用通道适配器，我们始终可以通过以下方式使用 Java DSL

public class MainFlow {
  @Bean
  public IntegrationFlow buildFlow() {
    Flux<Message<?>> myFlux = this.customReactiveSource
                .map(event ->
                    MessageBuilder
                    .withPayload(event.getBody())
                    .setHeader(MyReactiveHeaders.SOURCE_NAME, event.getSourceName())
                    .build());
     return IntegrationFlow.from(myFlux)
        .handle(outputChannel)
        .get();
  }
}

响应式出站通道适配器实现是关于启动（或继续）响应式流，以根据目标协议提供的响应式 API 与外部系统交互。入站有效负载本身可以是响应式类型，也可以是整个集成流的事件，该事件是顶部响应式流的一部分。如果我们在单向、即发即弃场景中，则返回的响应式类型可以立即订阅，或者它向下游传播（请求-回复场景），以用于进一步的集成流或目标业务逻辑中的显式订阅，但仍保留响应式流语义。

响应式出站通道适配器示例

public class CustomReactiveMessageHandler extends AbstractReactiveMessageHandler {

    private final CustomEntityOperations customEntityOperations;

    public CustomReactiveMessageHandler(CustomEntityOperations customEntityOperations) {
        Assert.notNull(customEntityOperations, "'customEntityOperations' must not be null");
        this.customEntityOperations = customEntityOperations;
    }

    @Override
    protected Mono<Void> handleMessageInternal(Message<?> message) {
        return Mono.fromSupplier(() -> message.getHeaders().get("queryType", Type.class))
                .flatMap(mode -> {
                    switch (mode) {
                        case INSERT:
                            return handleInsert(message);
                        case UPDATE:
                            return handleUpdate(message);
                        default:
                            return Mono.error(new IllegalArgumentException());
                    }
                }).then();
    }

    private Mono<Void> handleInsert(Message<?> message) {
        return this.customEntityOperations.insert(message.getPayload())
                .then();
    }

    private Mono<Void> handleUpdate(Message<?> message) {
        return this.r2dbcEntityOperations.update(message.getPayload())
                .then();
    }

    public enum Type {
        INSERT,
        UPDATE,
    }
}

我们将能够使用这两个通道适配器

public class MainFlow {

  @Autowired
  private CustomReactiveMessageProducer customReactiveMessageProducer;

  @Autowired
  private CustomReactiveMessageHandler customReactiveMessageHandler;

  @Bean
  public IntegrationFlow buildFlow() {
     return IntegrationFlow.from(customReactiveMessageProducer)
        .transform(someOperation)
        .handle(customReactiveMessageHandler)
        .get();
  }
}

目前，Spring Integration 为 WebFlux、RSocket、MongoDb、R2DBC、ZeroMQ、GraphQL、Apache Cassandra 提供了通道适配器（或网关）实现。Redis Stream 通道适配器 也是响应式的，并使用 Spring Data 的 ReactiveStreamOperations。更多响应式通道适配器正在开发中，例如用于 Kafka 中的 Apache Kafka，基于 Spring for Apache Kafka 的 ReactiveKafkaProducerTemplate 和 ReactiveKafkaConsumerTemplate 等。对于许多其他非响应式通道适配器，建议使用线程池，以避免在响应式流处理期间阻塞。

响应式到命令式上下文传播

当 Context Propagation 库位于类路径上时，Project Reactor 可以获取 ThreadLocal 值（例如 Micrometer Observation 或 SecurityContextHolder）并将它们存储到 Subscriber 上下文中。反向操作也是可能的，当我们为了跟踪需要填充日志 MDC 或让从响应式流调用的服务从作用域恢复观察时。有关其用于上下文传播的特殊运算符的更多信息，请参阅 Project Reactor 文档。存储和恢复上下文工作顺利，如果我们的整个解决方案是单个响应式流组合，因为 Subscriber 上下文从下游一直到组合的开始（Flux 或 Mono）都是可见的。但是，如果应用程序在不同的 Flux 实例之间切换或切换到命令式处理再切换回来，则绑定到 Subscriber 的上下文可能不可用。对于这种情况，Spring Integration 提供了额外的功能（从版本 6.0.5 开始），将 Reactor ContextView 存储到从响应式流生成的 IntegrationMessageHeaderAccessor.REACTOR_CONTEXT 消息头中，例如当我们执行直接 send() 操作时。然后，此头在 FluxMessageChannel.subscribeTo() 中用于恢复此通道将发出的 Message 的 Reactor 上下文。目前，此头由 WebFluxInboundEndpoint 和 RSocketInboundGateway 组件填充，但可用于执行响应式到命令式集成的任何解决方案中。填充此头的逻辑如下

return requestMono
        .flatMap((message) ->
                Mono.deferContextual((context) ->
                        Mono.just(message)
                                .handle((messageToSend, sink) ->
                                        send(messageWithReactorContextIfAny(messageToSend, context)))));
...

private Message<?> messageWithReactorContextIfAny(Message<?> message, ContextView context) {
    if (!context.isEmpty()) {
        return getMessageBuilderFactory()
                .fromMessage(message)
                .setHeader(IntegrationMessageHeaderAccessor.REACTOR_CONTEXT, context)
                .build();
    }
    return message;
}

请注意，我们仍然需要使用 handle() 运算符来使 Reactor 从上下文中恢复 ThreadLocal 值。即使它作为头发送，框架也无法假设它是否会恢复到下游的 ThreadLocal 值。

为了从另一个 Flux 或 Mono 组合中的 Message 恢复上下文，可以执行此逻辑

Mono.just(message)
        .handle((messageToHandle, sink) -> ...)
        .contextWrite(StaticMessageHeaderAccessor.getReactorContext(message)));
© . This site is unofficial and not affiliated with VMware.