java - 如何从 Flux 主体获取数据

Question

我创建了一个 spring-cloud gateway 项目，但是我无法从 serverHttpRequest.body 获取数据，似乎 body.subscribe 不起作用。我应该如何从 body 获取数据并解析它？和 AtomicReference<String> bodyRef = new AtomicReference<>()还遇到一个问题，bodyRef 总是等于 null

spring-cloud-gateway 2.1.0

@Component
public class ApiGlobalFilter implements GlobalFilter, Ordered {

    @Override
    public int getOrder() {
        return 0;
    }

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {

        if (HttpMethod.POST.toString().equals(exchange.getRequest().getMethodValue())){

       /*//     exchange.getRequest().get

            String bodyStr = resolveBodyFromRequest(serverHttpRequest);*/
            ServerHttpRequest serverHttpRequest = exchange.getRequest();

            Flux<DataBuffer> body = serverHttpRequest.getBody();
          //  System.out.println(body.toString());
/*            AtomicReference<String> bodyRef = new AtomicReference<>();
            serverHttpRequest.getBody().subscribe(dataBuffer -> {
                CharBuffer charBuffer = StandardCharsets.UTF_8.decode(dataBuffer.asByteBuffer());
                DataBufferUtils.release(dataBuffer);
                bodyRef.set(charBuffer.toString());
            });
            String bodyStr = bodyRef.get();*/


            AtomicReference<String> bodyRef = new AtomicReference<>();

            body.subscribe(buffer -> {
                System.out.println("start111111111");
                byte[] bytes = new byte[buffer.readableByteCount()];
                buffer.read(bytes);
                DataBufferUtils.release(buffer);
                try {
                    String bodyString = new String(bytes, "utf-8");

                    bodyRef.set(bodyString);
                    System.out.println(bodyString);
                    System.out.println("finish11111111");
                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                }
            });


            String bodyStr = bodyRef.get();
            System.out.println(bodyStr);

            URI uri = serverHttpRequest.getURI();
            ServerHttpRequest request = serverHttpRequest.mutate().uri(uri).build();
            DataBuffer bodyDataBuffer = stringBuffer(bodyStr);
            Flux<DataBuffer> bodyFlux = Flux.just(bodyDataBuffer);

            request = new ServerHttpRequestDecorator(request) {
                @Override
                public Flux<DataBuffer> getBody() {
                    return bodyFlux;
                }
            };

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


          //  return chain.filter(exchange);
        }

        String token = exchange.getRequest().getQueryParams().getFirst("token");
        if (StringUtils.isBlank(token)) {
            ServerHttpResponse response = exchange.getResponse();
            Map<String,Object> message = new HashMap<>();
            message.put("status", -1);
            message.put("data", "fail");
            byte[] bits = message.toString().getBytes(StandardCharsets.UTF_8);
            DataBuffer buffer = response.bufferFactory().wrap(bits);
            response.setStatusCode(HttpStatus.UNAUTHORIZED);
            response.getHeaders().add("Content-Type", "text/json;charset=UTF-8");
            return response.writeWith(Mono.just(buffer));
        }
        return chain.filter(exchange);
    }



    private String resolveBodyFromRequest(ServerHttpRequest serverHttpRequest) {
        Flux<DataBuffer> body = serverHttpRequest.getBody();

        AtomicReference<String> bodyRef = new AtomicReference<>();
        body.subscribe(buffer -> {
            CharBuffer charBuffer = StandardCharsets.UTF_8.decode(buffer.asByteBuffer());
            DataBufferUtils.release(buffer);
            bodyRef.set(charBuffer.toString());
        });

        return bodyRef.get();
    }


    private 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;
    }

    @Bean
    public HiddenHttpMethodFilter hiddenHttpMethodFilter() {
        return new HiddenHttpMethodFilter() {
            @Override
            public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
                return chain.filter(exchange);
            }
        };
    }

}

从 body 获取数据

Answer 1

此问题可能是由较高版本的spring-cloud引起的。我降低版本然后它就可以工作了。 https://blog.51cto.com/thinklili/2329184这篇博客可能会解释这个问题