对于最新的稳定版本，请使用 Spring Integration 6.5.1！spring-doc.cadn.net.cn

WebFlux 支持

WebFlux Spring 集成模块（spring-integration-webflux）允许以响应方式执行 HTTP 请求和处理入站 HTTP 请求。spring-doc.cadn.net.cn

您需要将此依赖项包含在您的项目中：spring-doc.cadn.net.cn

Mavenspring-doc.cadn.net.cn
Gradlespring-doc.cadn.net.cn

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-webflux</artifactId>
    <version>6.2.11</version>
</dependency>

compile "org.springframework.integration:spring-integration-webflux:6.2.11"

这io.projectreactor.netty:reactor-netty如果不是基于 Servlet 的服务器配置，则必须包含依赖项。spring-doc.cadn.net.cn

WebFlux 支持包括以下网关实现：WebFluxInboundEndpoint和WebFluxRequestExecutingMessageHandler. 该支持完全基于 Spring WebFlux 和 Project Reactor 基础。有关更多信息，请参阅 HTTP 支持，因为许多选项在响应式 HTTP 组件和常规 HTTP 组件之间共享。spring-doc.cadn.net.cn

WebFlux 命名空间支持

Spring Integration 提供了一个webflux命名空间和相应的架构定义。要将其包含在配置中，请在应用程序上下文配置文件中添加以下命名空间声明：spring-doc.cadn.net.cn

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xmlns:int="http://www.springframework.org/schema/integration"
  xmlns:int-webflux="http://www.springframework.org/schema/integration/webflux"
  xsi:schemaLocation="
    http://www.springframework.org/schema/beans
    https://www.springframework.org/schema/beans/spring-beans.xsd
    http://www.springframework.org/schema/integration
    https://www.springframework.org/schema/integration/spring-integration.xsd
    http://www.springframework.org/schema/integration/webflux
    https://www.springframework.org/schema/integration/webflux/spring-integration-webflux.xsd">
    ...
</beans>

WebFlux 入站组件

从 5.0 版开始，WebFluxInboundEndpoint实现WebHandler被提供。此组件类似于基于 MVC 的HttpRequestHandlingEndpointSupport，它通过新提取的BaseHttpInboundEndpoint. 它用于 Spring WebFlux 响应式环境（而不是 MVC）。以下示例显示了 WebFlux 端点的简单实现：spring-doc.cadn.net.cn

@Bean
public IntegrationFlow inboundChannelAdapterFlow() {
    return IntegrationFlow
        .from(WebFlux.inboundChannelAdapter("/reactivePost")
            .requestMapping(m -> m.methods(HttpMethod.POST))
            .requestPayloadType(ResolvableType.forClassWithGenerics(Flux.class, String.class))
            .statusCodeFunction(m -> HttpStatus.ACCEPTED))
        .channel(c -> c.queue("storeChannel"))
        .get();
}

@Bean
fun inboundChannelAdapterFlow() =
    integrationFlow(
        WebFlux.inboundChannelAdapter("/reactivePost")
            .apply {
                requestMapping { m -> m.methods(HttpMethod.POST) }
                requestPayloadType(ResolvableType.forClassWithGenerics(Flux::class.java, String::class.java))
                statusCodeFunction { m -> HttpStatus.ACCEPTED }
            })
    {
        channel { queue("storeChannel") }
    }

@Configuration
@EnableWebFlux
@EnableIntegration
public class ReactiveHttpConfiguration {

    @Bean
    public WebFluxInboundEndpoint simpleInboundEndpoint() {
        WebFluxInboundEndpoint endpoint = new WebFluxInboundEndpoint();
        RequestMapping requestMapping = new RequestMapping();
        requestMapping.setPathPatterns("/test");
        endpoint.setRequestMapping(requestMapping);
        endpoint.setRequestChannelName("serviceChannel");
        return endpoint;
    }

    @ServiceActivator(inputChannel = "serviceChannel")
    String service() {
        return "It works!";
    }

}

<int-webflux:inbound-gateway request-channel="requests" path="/sse">
    <int-webflux:request-mapping produces="text/event-stream"/>
</int-webflux:inbound-gateway>

配置类似于HttpRequestHandlingEndpointSupport（在示例前面提到），除了我们使用@EnableWebFlux将 WebFlux 基础设施添加到我们的集成应用程序中。此外，WebFluxInboundEndpoint执行sendAndReceive使用响应式 HTTP 服务器实现提供的基于按需的背压功能对下游流进行作。spring-doc.cadn.net.cn

回复部分也是非阻塞的，并且基于内部FutureReplyChannel，它平面映射到回复Mono用于按需解决。

您可以配置WebFluxInboundEndpoint使用自定义ServerCodecConfigurer一个RequestedContentTypeResolver，甚至ReactiveAdapterRegistry. 后者提供了一种机制，可用于以任何响应式类型返回回复：ReactorFlux、RxJavaObservable,Flowable，等。这样，我们就可以使用 Spring Integration 组件实现服务器发送事件场景，如以下示例所示：spring-doc.cadn.net.cn

@Bean
public IntegrationFlow sseFlow() {
    return IntegrationFlow
            .from(WebFlux.inboundGateway("/sse")
                    .requestMapping(m -> m.produces(MediaType.TEXT_EVENT_STREAM_VALUE)))
            .handle((p, h) -> Flux.just("foo", "bar", "baz"))
            .get();
}

@Bean
fun sseFlow() =
     integrationFlow(
            WebFlux.inboundGateway("/sse")
                       .requestMapping(m -> m.produces(MediaType.TEXT_EVENT_STREAM_VALUE)))
            {
                 handle { (p, h) -> Flux.just("foo", "bar", "baz") }
            }

@Bean
public WebFluxInboundEndpoint webfluxInboundGateway() {
    WebFluxInboundEndpoint endpoint = new WebFluxInboundEndpoint();
    RequestMapping requestMapping = new RequestMapping();
    requestMapping.setPathPatterns("/sse");
    requestMapping.setProduces(MediaType.TEXT_EVENT_STREAM_VALUE);
    endpoint.setRequestMapping(requestMapping);
    endpoint.setRequestChannelName("requests");
    return endpoint;
}

<int-webflux:inbound-channel-adapter id="reactiveFullConfig" channel="requests"
                               path="test1"
                               auto-startup="false"
                               phase="101"
                               request-payload-type="byte[]"
                               error-channel="errorChannel"
                               payload-expression="payload"
                               supported-methods="PUT"
                               status-code-expression="'202'"
                               header-mapper="headerMapper"
                               codec-configurer="codecConfigurer"
                               reactive-adapter-registry="reactiveAdapterRegistry"
                               requested-content-type-resolver="requestedContentTypeResolver">
            <int-webflux:request-mapping headers="foo"/>
            <int-webflux:cross-origin origin="foo" method="PUT"/>
            <int-webflux:header name="foo" expression="'foo'"/>
</int-webflux:inbound-channel-adapter>

有关更多可能的配置选项，请参阅请求映射支持和跨域资源共享（CORS）支持。spring-doc.cadn.net.cn

当请求正文为空或payloadExpression返回null，请求参数（MultiValueMap<String, String>）用于payload要处理的目标消息。spring-doc.cadn.net.cn

有效负载验证

从 5.2 版开始，WebFluxInboundEndpoint可以配置为Validator. 与 HTTP 支持中的 MVC 验证不同，它用于验证Publisher请求已由HttpMessageReader，然后再执行回退和payloadExpression功能。框架无法假设Publisher对象可以在构建最终有效负载之后。如果要求限制最终有效负载（或其Publisher元素），验证应向下游而不是 WebFlux 端点。更多信息请参阅 Spring WebFlux 文档。无效的有效负载将被拒绝，并带有IntegrationWebExchangeBindException（一个WebExchangeBindException扩展），包含所有验证Errors. 请参阅 Spring Framework 参考手册中有关验证的更多信息。spring-doc.cadn.net.cn

WebFlux 出站组件

这WebFluxRequestExecutingMessageHandler（从 5.0 版开始）实现类似于HttpRequestExecutingMessageHandler. 它使用WebClient来自 Spring Framework WebFlux 模块。要配置它，请定义类似于以下内容的 bean：spring-doc.cadn.net.cn

@Bean
public IntegrationFlow outboundReactive() {
    return f -> f
        .handle(WebFlux.<MultiValueMap<String, String>>outboundGateway(m ->
                UriComponentsBuilder.fromUriString("http://localhost:8080/foo")
                        .queryParams(m.getPayload())
                        .build()
                        .toUri())
                .httpMethod(HttpMethod.GET)
                .expectedResponseType(String.class));
}

@Bean
fun outboundReactive() =
    integrationFlow {
        handle(
            WebFlux.outboundGateway<MultiValueMap<String, String>>({ m ->
                UriComponentsBuilder.fromUriString("http://localhost:8080/foo")
                    .queryParams(m.getPayload())
                    .build()
                    .toUri()
            })
                .httpMethod(HttpMethod.GET)
                .expectedResponseType(String::class.java)
        )
    }

@ServiceActivator(inputChannel = "reactiveHttpOutRequest")
@Bean
public WebFluxRequestExecutingMessageHandler reactiveOutbound(WebClient client) {
    WebFluxRequestExecutingMessageHandler handler =
        new WebFluxRequestExecutingMessageHandler("http://localhost:8080/foo", client);
    handler.setHttpMethod(HttpMethod.POST);
    handler.setExpectedResponseType(String.class);
    return handler;
}

<int-webflux:outbound-gateway id="reactiveExample1"
    request-channel="requests"
    url="http://localhost/test"
    http-method-expression="headers.httpMethod"
    extract-request-payload="false"
    expected-response-type-expression="payload"
    charset="UTF-8"
    reply-timeout="1234"
    reply-channel="replies"/>

<int-webflux:outbound-channel-adapter id="reactiveExample2"
    url="http://localhost/example"
    http-method="GET"
    channel="requests"
    charset="UTF-8"
    extract-payload="false"
    expected-response-type="java.lang.String"
    order="3"
    auto-startup="false"/>

这WebClient exchange()作返回一个Mono<ClientResponse>，该Mono.map()steps）到AbstractIntegrationMessageBuilder作为WebFluxRequestExecutingMessageHandler. 与ReactiveChannel作为outputChannel这Mono<ClientResponse>评估将推迟到进行下游订阅。否则，它被视为asyncmode 和Mono响应适应SettableListenableFuture对于来自WebFluxRequestExecutingMessageHandler. 输出消息的目标有效负载取决于WebFluxRequestExecutingMessageHandler配置。这setExpectedResponseType(Class<?>)或setExpectedResponseTypeExpression(Expression)标识响应正文元素转换的目标类型。如果replyPayloadToFlux设置为true，则响应正文将转换为Flux与提供的expectedResponseType对于每个元素，并且这个Flux作为有效负载发送到下游。之后，您可以使用拆分器来迭代此Flux以反应性的方式。spring-doc.cadn.net.cn

此外，一个BodyExtractor<?, ClientHttpResponse>可以注入WebFluxRequestExecutingMessageHandler而不是expectedResponseType和replyPayloadToFlux性能。它可用于对ClientHttpResponse以及对正文和 HTTP 标头转换的更多控制。 Spring Integration 提供ClientHttpResponseBodyExtractor作为恒等函数来产生（下游）整体ClientHttpResponse以及任何其他可能的自定义逻辑。spring-doc.cadn.net.cn

从 5.2 版开始，WebFluxRequestExecutingMessageHandler支持反应性Publisher,Resource和MultiValueMap类型作为请求消息有效负载。各自的BodyInserter在内部用于填充到WebClient.RequestBodySpec. 当有效负载是响应式的Publisher，配置的publisherElementType或publisherElementTypeExpression可用于确定发布者元素类型的类型。表达式必须解析为Class<?>,String解析为目标Class<?>或ParameterizedTypeReference.spring-doc.cadn.net.cn

从 5.5 版本开始，WebFluxRequestExecutingMessageHandler暴露了一个extractResponseBody标志（即true默认情况下）仅返回响应正文，或返回整个ResponseEntity作为回复消息有效负载，独立于提供的expectedResponseType或replyPayloadToFlux. 如果ResponseEntity，这个标志被忽略了，整个ResponseEntity被返回。spring-doc.cadn.net.cn

有关更多可能的配置选项，请参阅 HTTP 出站组件。spring-doc.cadn.net.cn

WebFlux 标头映射

由于 WebFlux 组件完全基于 HTTP 协议，因此 HTTP 标头映射没有区别。有关用于映射标头的更多可能选项和组件，请参阅 HTTP 标头映射。spring-doc.cadn.net.cn

WebFlux 请求属性

从 6.0 版开始，WebFluxRequestExecutingMessageHandler可以通过以下方式配置为评估请求属性setAttributeVariablesExpression(). 该 SpEL 表达式必须在Map. 然后，这样的映射被传播到WebClient.RequestBodySpec.attributes(Consumer<Map<String, Object>> attributesConsumer)HTTP 请求配置回调。如果需要从Messageto request 和下游过滤器将获得对这些属性的访问权限以进行进一步处理。spring-doc.cadn.net.cn