如何在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（）正在阻塞，这在线程reactor-http-nio-2中不受支持

调用者

卷曲http://127.0.0.1:45441/test1/test-d“12312321312”-i-H“内容类型：应用程序/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();
}

参考文档中有更多示例