前言

最近在开发api-gateway的时候遇到了一个问题，网上能够找到的解决方案也很少，之后由公司的大佬解决了这个问题。写下这篇文章记录一下解决方案。希望可以帮助到更多的人。

环境

java版本：8
框架：spring-cloud 2.0.0.RC1

介绍

api-gateway主要接收前端请求，然后对请求的数据进行验证，验证之后请求反向代理到服务器。当请求 method 为 GET 时，可以顺利通过api-gateway。当请求 method 为 POST 时，api-gateway则会报如下错误：

2018-07-18 11:49:04.073 ERROR 3025 --- [ctor-http-nio-4] .a.w.r.e.DefaultErrorWebExceptionHandler : Failed to handle request [POST http://localhost:9000/api/hello]

java.lang.IllegalStateException: Only one connection receive subscriber allowed.

    at reactor.ipc.netty.channel.FluxReceive.startReceiver(FluxReceive.java:276) ~[reactor-netty-0.7.5.RELEASE.jar:0.7.5.RELEASE]

    at reactor.ipc.netty.channel.FluxReceive.lambda$subscribe$2(FluxReceive.java:127) ~[reactor-netty-0.7.5.RELEASE.jar:0.7.5.RELEASE]

    at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:163) ~[netty-common-4.1.22.Final.jar:4.1.22.Final]

    at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:404) ~[netty-common-4.1.22.Final.jar:4.1.22.Final]

    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:463) ~[netty-transport-4.1.22.Final.jar:4.1.22.Final]

    at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:886) ~[netty-common-4.1.22.Final.jar:4.1.22.Final]

    at java.lang.Thread.run(Thread.java:748) ~[na:1.8.0_171]

错误分析

实际上spring-cloud-gateway反向代理的原理是，首先读取原请求的数据，然后构造一个新的请求，将原请求的数据封装到新的请求中，然后再转发出去。然而我们在他封装之前读取了一次request body，而request body只能读取一次。因此就出现了上面的错误。

解决方案

对于上面的错误我们给出的解决方案是：

读取request body的时候，我们再封装一次request，转发出去

下面是我们的代码：

@Component

public class PostFilter extends AbstractNameValueGatewayFilterFactory {

    private static final String X_APP_ID_HEADER = "X-app-id";

    public static final String X_FORWARDED_FOR = "X-Forwarded-For";

    @Override

    public GatewayFilter apply(NameValueConfig nameValueConfig) {

        return (exchange, chain) -> {

            URI uri = exchange.getRequest().getURI();

            URI ex = UriComponentsBuilder.fromUri(uri).build(true).toUri();

            ServerHttpRequest request = exchange.getRequest().mutate().uri(ex).build();

            if("POST".equalsIgnoreCase(request.getMethodValue())){//判断是否为POST请求

                Flux<DataBuffer> body = request.getBody();

                AtomicReference<String> bodyRef = new AtomicReference<>();//缓存读取的request body信息

                body.subscribe(dataBuffer -> {

                    CharBuffer charBuffer = StandardCharsets.UTF_8.decode(dataBuffer.asByteBuffer());

                    DataBufferUtils.release(dataBuffer);

                    bodyRef.set(charBuffer.toString());

                });//读取request body到缓存

                String bodyStr = bodyRef.get();//获取request body

                System.out.println(bodyStr);//这里是我们需要做的操作

                DataBuffer bodyDataBuffer = stringBuffer(bodyStr);

                Flux<DataBuffer> bodyFlux = Flux.just(bodyDataBuffer);

                request = new ServerHttpRequestDecorator(request){

                    @Override

                    public Flux<DataBuffer> getBody() {

                        return bodyFlux;

                    }

                };//封装我们的request

            }

            return chain.filter(exchange.mutate().request(request).build());

        };

    }

    protected DataBuffer stringBuffer(String value) {

        byte[] bytes = value.getBytes(StandardCharsets.UTF_8);

        NettyDataBufferFactory nettyDataBufferFactory = new NettyDataBufferFactory(ByteBufAllocator.DEFAULT);

        DataBuffer buffer = nettyDataBufferFactory.allocateBuffer(bytes.length);

        buffer.write(bytes);

        return buffer;

    }

}

至此，该问题得到解决。

巴特西

[spring cloud] [error] java.lang.IllegalStateException: Only one connection receive subscriber allowed.

前言

环境

介绍

错误分析

解决方案

最新文章

热门文章