2

在 Reactor Netty 中，当通过 向 TCP 通道发送数据时out.send(publisher)，人们会期望任何发布者都能正常工作。但是，如果我们使用更复杂的带有延迟元素的而不是简单的立即Flux数，那么它就会停止正常工作。例如，如果我们使用这个 hello world TCP 回显服务器，它会按预期工作：

import reactor.core.publisher.Flux;
import reactor.netty.DisposableServer;
import reactor.netty.tcp.TcpServer;

import java.time.Duration;

public class Reactor1 {
    public static void main(String[] args) throws Exception {
        DisposableServer server = TcpServer.create()
            .port(3344)
            .handle((in, out) -> in
                .receive()
                .asString()
                .flatMap(s ->
                    out.sendString(Flux.just(s.toUpperCase()))
                ))
            .bind()
            .block();
        server.channel().closeFuture().sync();
    }
}

但是，如果我们更改out.sendString为

out.sendString(Flux.just(s.toUpperCase()).delayElements(Duration.ofSeconds(1)))

那么我们期望对于每个接收到的项目都会产生一个延迟一秒的输出。

但是，服务器的行为方式是，如果它在间隔期间接收到多个项目，它将只为第一个项目生成输出。例如，下面我们在第一秒输入aaand ，但只生成输出（一秒后）：bbAA

$ nc localhost 3344
aa
bb
AA <after one second>

然后，如果我们稍后键入额外的行，我们会得到输出（一秒钟后），但来自前一个输入：

cc
BB <after one second>

有什么想法可以send()延迟按预期工作Flux吗？

4

2 回答 2

1

我认为您不应该为out.sendString(...) This works 重新创建发布者：

DisposableServer server = TcpServer.create()
        .port(3344)
        .handle((in, out) -> out
                .options(NettyPipeline.SendOptions::flushOnEach)
                .sendString(in.receive()
                        .asString()
                        .map(String::toUpperCase)
                        .delayElements(Duration.ofSeconds(1))))
        .bind()
        .block();
server.channel().closeFuture().sync();
于 2019-03-06T12:27:29.987 回答
1

尝试使用concatMap。这有效：

DisposableServer server = TcpServer.create()
        .port(3344)
        .handle((in, out) -> in
                .receive()
                .asString()
                .concatMap(s ->
                        out.sendString(Flux.just(s.toUpperCase())
                                           .delayElements(Duration.ofSeconds(1)))
                ))
            .bind()
            .block();
server.channel().closeFuture().sync();

延迟传入流量

DisposableServer server = TcpServer.create()
        .port(3344)
        .handle((in, out) -> in
                .receive()
                .asString()
                .timestamp()
                .delayElements(Duration.ofSeconds(1))
                .concatMap(tuple2 ->
                        out.sendString(
                                Flux.just(tuple2.getT2().toUpperCase() +
                                        " " +
                                        (System.currentTimeMillis() - tuple2.getT1())
                                ))
                ))
        .bind()
        .block();
于 2019-03-06T15:54:18.543 回答