如何读取反应器-netty服务器的请求体?

Question

如何读取反应器-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中不支持

调用程序

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();
}

参考文档

中有更多示例