如何读取 reactor-netty 服务器中的请求体？

Question

如何读取reactor-netty服务器中的请求体？ 我想获取请求主体以确定响应的内容，但我在示例代码中找不到相关示例。

我的代码：

public static void main(String[] args) throws IOException {
        Consumer<ByteBuf> onSuccess = (ByteBuf request) -> {
            System.out.println("onSuccess: Request received!");
        };
        Consumer<Throwable> onError = (Throwable ex) -> {
            ex.getMessage();
            System.out.println(ex.getMessage());
        };
        Runnable onCompletion = () -> {
            System.out.println("Message Completed");

        };
        CountDownLatch latch = new CountDownLatch(1);
        DisposableServer server =
                HttpServer.create().handle(new BiFunction<HttpServerRequest, HttpServerResponse, Publisher<Void>>() {
                    @Override
                    public Publisher<Void> apply(HttpServerRequest httpServerRequest, HttpServerResponse httpServerResponse) {
                        Mono<byte[]> mono = httpServerRequest.receive()
                                .aggregate()
                                .asByteArray()
                                .doOnNext(new Consumer<byte[]>() {
                                    @Override
                                    public void accept(byte[] bytes) {
                                        System.out.println(1);
                                    }
                                })
                                .doOnError(onError)
                                .doOnTerminate(onCompletion)
                                .flatMap(bytes -> {
                                    return Mono.just(bytes);
                                });
                        mono.block();
                        // I want to get http body;

                        return httpServerResponse.sendString(Mono.just("Hello world"));
                    }
                }).host("localhost")
                        .port(45441)
                        .bindNow();
        System.in.read();
    }

异常

block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor-http-nio-2

调用者

curl http://127.0.0.1:45441/test1/test -d "12312321312" -i -H 'Content-Type:application/json' -vvv

pom

   <dependencies>
        <dependency>
            <groupId>io.projectreactor.netty</groupId>
            <artifactId>reactor-netty-core</artifactId>
        </dependency>
        <dependency>
            <groupId>io.projectreactor.netty</groupId>
            <artifactId>reactor-netty-http</artifactId>
        </dependency>
        <!-- https://mvnrepository.com/artifact/io.netty/netty-transport-native-kqueue -->
        <!--    <dependency>-->
        <!--        <groupId>io.netty</groupId>-->
        <!--        <artifactId>netty-transport-native-kqueue</artifactId>-->
        <!--        <version>4.1.66.Final</version>-->
        <!--        <classifier>osx-x86_64</classifier>-->
        <!--    </dependency>-->

        <!--        <dependency>-->
        <!--            <groupId>io.netty</groupId>-->
        <!--            <artifactId>netty-all</artifactId>-->
        <!--            <version>4.1.66.Final</version>-->
        <!--        </dependency>-->

        <!-- https://mvnrepository.com/artifact/io.netty/netty-transport-native-kqueue -->
        <!-- https://mvnrepository.com/artifact/io.netty/netty-transport -->
    </dependencies>
    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>io.projectreactor</groupId>
                <artifactId>reactor-bom</artifactId>
                <version>2020.0.10</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

谢谢您的回答！

Answer 1

使用 Reactor Netty，您必须执行相反的逻辑：在收到这些字节时发送这些字节。你永远不应该阻塞在事件循环中。上面的例子可以像下面这样重写：

public static void main(String[] args) throws IOException {
    Consumer<Throwable> onError = (Throwable ex) -> {
        System.out.println(ex.getMessage());
    };
    Runnable onCompletion = () -> {
        System.out.println("Message Completed");
    };
    DisposableServer server =
            HttpServer.create()
                    .handle((req, res) ->
                            res.sendByteArray(req.receive()
                                    .aggregate()
                                    .asByteArray()
                                    .doOnNext(bytes -> System.out.println(1))
                                    .doOnError(onError)
                                    .doOnTerminate(onCompletion)
                                    .flatMap(Mono::just)))
                    .host("localhost")
                    .port(45441)
                    .bindNow();

    server.onDispose()
            .block();
}

reference documentation

中还有更多示例