Spring Integration WebFlux 隔次调用失败，报400错误求助

问题描述 #

搭建了一个Spring Integration管道，暴露/reactivePost端点接收POST请求，调用另一REST服务的GET接口后返回结果。但出现异常：首次请求可正常响应，第二次请求返回400 Bad Request。

管道配置代码 #

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.handler.LoggingHandler;
import org.springframework.integration.webflux.dsl.WebFlux;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.web.util.UriComponentsBuilder;

@Configuration
@EnableIntegration
@Slf4j
public class SpringIntegrationFLowConfiguration {

    @Bean
    public IntegrationFlow inboundPost(
            @Qualifier("inputChannel") MessageChannel inputChannel) {

        return IntegrationFlow
                .from(WebFlux.inboundGateway("/reactivePost")
                        .requestMapping(requestMapping -> requestMapping.methods(HttpMethod.POST))
                        .requestPayloadType(String.class)
                        .statusCodeFunction(m -> HttpStatus.OK))
                .channel(inputChannel)
                .get();
    }

    @Bean("inputChannel")
    public PublishSubscribeChannel inputChannel() {
        PublishSubscribeChannel channel = new PublishSubscribeChannel();
        channel.setDatatypes(String.class);
        return channel;
    }

    @Bean("outputGet")
    public IntegrationFlow outboundChannelFlow(
            @Qualifier("inputChannel") MessageChannel inputChannel) {

        return IntegrationFlow
                .from(inputChannel)
                .handle(WebFlux.<String>outboundGateway(message ->
                                UriComponentsBuilder.fromUriString("http://localhost:8080/")
                                        .pathSegment(message.getPayload())
                                        .build()
                                        .toUri())
                        .httpMethod(HttpMethod.GET)
                        .expectedResponseType(String.class))
                .get();
    }
}

目标REST服务控制器代码 #

@RestController
public class HelloController {

    @GetMapping("/{name}")
    public Mono<String> hello(@PathVariable(name = "name") String name) {
        return Mono.just("Hello " + name + "!");
    }
}

错误日志 #

r.n.http.server.HttpServerOperations     : [33dbeb61, L:/127.0.0.1:8080 - R:/127.0.0.1:51916] Decoding failed: FULL_REQUEST(decodeResult: failure(java.lang.IllegalArgumentException: text is empty (possibly HTTP/0.9)), version: HTTP/1.0, content: UnpooledByteBufAllocator$InstrumentedUnpooledUnsafeHeapByteBuf(ridx: 0, widx: 0, cap: 0)) GET /bad-request HTTP/1.0

原因分析 #

核心问题在于错误使用了PublishSubscribeChannel，同时流配置缺失请求响应闭环：

• PublishSubscribeChannel是发布订阅模式，会将消息分发给所有订阅者。你的两个IntegrationFlow都订阅了inputChannel：inboundPost流将消息发送到通道的同时，自身也作为订阅者监听通道；outputGet流同样订阅通道处理消息。
• 第一次请求时，消息被outputGet正常消费并返回结果给网关。第二次请求时，通道会将消息分发给两个订阅者，inboundPost流没有后续处理逻辑，导致消息无效（payload为空），进而引发目标服务的400错误。
• 另外，inboundPost流的入站网关没有正确绑定响应返回逻辑，无法形成完整的请求-响应闭环，加剧了后续请求的异常。

解决方案 #

方案一：替换通道类型 #

将PublishSubscribeChannel改为DirectChannel（默认单订阅者通道类型），避免消息被重复分发：

@Bean("inputChannel")
public MessageChannel inputChannel() {
    return new DirectChannel();
}

方案二：整合流逻辑（推荐） #

将入站网关和出站网关整合到同一个IntegrationFlow中，形成完整的请求-响应链路，无需单独配置通道：

@Configuration
@EnableIntegration
@Slf4j
public class SpringIntegrationFLowConfiguration {

    @Bean
    public IntegrationFlow reactivePostFlow() {
        return IntegrationFlow
                .from(WebFlux.inboundGateway("/reactivePost")
                        .requestMapping(m -> m.methods(HttpMethod.POST))
                        .requestPayloadType(String.class)
                        .statusCodeFunction(m -> HttpStatus.OK))
                .handle(WebFlux.<String>outboundGateway(payload ->
                                UriComponentsBuilder.fromUriString("http://localhost:8080/")
                                        .pathSegment(payload)
                                        .build()
                                        .toUri())
                        .httpMethod(HttpMethod.GET)
                        .expectedResponseType(String.class))
                .get();
    }
}

验证 #

修改后重复调用/reactivePost端点，所有请求都应正常调用目标服务并返回结果，不再出现400错误。

内容的提问来源于stack exchange，提问作者Poorya Hosseini