如何使用 reactor-netty TcpClient 链接多个发送和接收操作答案

【问题标题】：How to chain multiple send and receive operations with reactor-netty TcpClient如何使用 reactor-netty TcpClient 链接多个发送和接收操作
【发布时间】：2020-04-13 00:50:12
【问题描述】：

我需要在由顺序发送 -> 接收 -> 发送 -> 接收组成的 TCP 连接中执行自定义握手并返回最后一个接收值，但我在链接执行时遇到问题。

这是我所拥有的：

    DisposableServer server = TcpServer.create()
            .host("localhost")
            .port(4059)
            .wiretap(true)
            .handle((nettyInbound, nettyOutbound) ->
                    nettyInbound.receive().asByteArray().flatMap(bytes -> {
                        log.info("Server inbound: {}", bytes);
                        if (Arrays.equals(bytes, new byte[]{1, 2, 3})) {
                            nettyOutbound.sendByteArray(Mono.just(new byte[]{7, 6, 5})).then().subscribe();
                        } else if (Arrays.equals(bytes, new byte[]{5, 6, 7})) {
                            nettyOutbound.sendByteArray(Mono.just(new byte[]{9, 8, 7})).then().subscribe();
                        }
                        return Mono.empty();
                    }))
            .bindNow();

    TcpClient.create()
            .host("localhost")
            .port(4059)
            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
            .wiretap(true)
            .connect()
            .flatMap(connection ->
                    connection.outbound().sendByteArray(Mono.just(new byte[]{1, 2, 3}))
                            .then(connection.inbound().receive().asByteArray().next().flatMap(bytes -> {
                                log.info("bytes {}", bytes);
                                return Mono.empty();
                            })).sendByteArray(Mono.just(new byte[]{5, 6, 7}))
                            .then(connection.inbound().receive().asByteArray().next().flatMap(bytes -> {
                                log.info("bytes {}", bytes);
                                return Mono.empty();
                            }))
                            .then()
            )
            .subscribe();

    server.onDispose().block();

最大的问题是TcpClient 的第二个receive 的flatMap 没有被执行，并且在日志中wiretap 正确显示数据已被读取但未发送到flatMap。

日志还显示客户端发送了多条相同的消息：

14:25:04.394 [reactor-tcp-nio-2] DEBUG reactor.netty.tcp.TcpClient - [id: 0x3d2ab8be, L:/127.0.0.1:52044 - R:localhost/127.0.0.1:4059] WRITE: 3B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03                                        |...             |
+--------+-------------------------------------------------+----------------+
14:25:04.480 [reactor-tcp-nio-2] DEBUG reactor.netty.tcp.TcpClient - [id: 0x3d2ab8be, L:/127.0.0.1:52044 - R:localhost/127.0.0.1:4059] FLUSH
14:25:04.509 [reactor-tcp-nio-2] DEBUG reactor.netty.channel.FluxReceive - [id: 0x3d2ab8be, L:/127.0.0.1:52044 - R:localhost/127.0.0.1:4059] Subscribing inbound receiver [pending: 0, cancelled:false, inboundDone: false]
14:25:04.525 [reactor-tcp-nio-3] DEBUG reactor.netty.tcp.TcpServer - [id: 0xf6dd2f6d, L:/127.0.0.1:4059 - R:/127.0.0.1:52044] READ: 3B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03                                        |...             |
+--------+-------------------------------------------------+----------------+
14:25:04.527 [reactor-tcp-nio-3] INFO com.example.TcpTest - Server inbound: [1, 2, 3]
14:25:04.528 [reactor-tcp-nio-3] DEBUG reactor.netty.tcp.TcpServer - [id: 0xf6dd2f6d, L:/127.0.0.1:4059 - R:/127.0.0.1:52044] WRITE: 3B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 07 06 05                                        |...             |
+--------+-------------------------------------------------+----------------+
14:25:04.528 [reactor-tcp-nio-3] DEBUG reactor.netty.tcp.TcpServer - [id: 0xf6dd2f6d, L:/127.0.0.1:4059 - R:/127.0.0.1:52044] FLUSH
14:25:04.529 [reactor-tcp-nio-3] DEBUG reactor.netty.tcp.TcpServer - [id: 0xf6dd2f6d, L:/127.0.0.1:4059 - R:/127.0.0.1:52044] READ COMPLETE
14:25:04.529 [reactor-tcp-nio-2] DEBUG reactor.netty.tcp.TcpClient - [id: 0x3d2ab8be, L:/127.0.0.1:52044 - R:localhost/127.0.0.1:4059] READ: 3B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 07 06 05                                        |...             |
+--------+-------------------------------------------------+----------------+
14:25:04.530 [reactor-tcp-nio-2] INFO com.example.TcpTest - bytes [7, 6, 5]
14:25:04.531 [reactor-tcp-nio-2] DEBUG reactor.netty.tcp.TcpClient - [id: 0x3d2ab8be, L:/127.0.0.1:52044 - R:localhost/127.0.0.1:4059] WRITE: 3B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03                                        |...             |
+--------+-------------------------------------------------+----------------+
14:25:04.536 [reactor-tcp-nio-2] DEBUG reactor.netty.tcp.TcpClient - [id: 0x3d2ab8be, L:/127.0.0.1:52044 - R:localhost/127.0.0.1:4059] FLUSH
14:25:04.536 [reactor-tcp-nio-2] DEBUG reactor.netty.tcp.TcpClient - [id: 0x3d2ab8be, L:/127.0.0.1:52044 - R:localhost/127.0.0.1:4059] WRITE: 3B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 05 06 07                                        |...             |
+--------+-------------------------------------------------+----------------+
14:25:04.536 [reactor-tcp-nio-3] DEBUG reactor.netty.tcp.TcpServer - [id: 0xf6dd2f6d, L:/127.0.0.1:4059 - R:/127.0.0.1:52044] READ: 3B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03                                        |...             |
+--------+-------------------------------------------------+----------------+
14:25:04.536 [reactor-tcp-nio-2] DEBUG reactor.netty.tcp.TcpClient - [id: 0x3d2ab8be, L:/127.0.0.1:52044 - R:localhost/127.0.0.1:4059] FLUSH
14:25:04.536 [reactor-tcp-nio-3] INFO com.example.TcpTest - Server inbound: [1, 2, 3]
14:25:04.536 [reactor-tcp-nio-3] DEBUG reactor.netty.tcp.TcpServer - [id: 0xf6dd2f6d, L:/127.0.0.1:4059 - R:/127.0.0.1:52044] WRITE: 3B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 07 06 05                                        |...             |
+--------+-------------------------------------------------+----------------+
14:25:04.537 [reactor-tcp-nio-2] DEBUG reactor.netty.channel.FluxReceive - [id: 0x3d2ab8be, L:/127.0.0.1:52044 - R:localhost/127.0.0.1:4059] Subscribing inbound receiver [pending: 0, cancelled:true, inboundDone: false]
14:25:04.537 [reactor-tcp-nio-2] DEBUG reactor.netty.tcp.TcpClient - [id: 0x3d2ab8be, L:/127.0.0.1:52044 - R:localhost/127.0.0.1:4059] READ COMPLETE
14:25:04.537 [reactor-tcp-nio-3] DEBUG reactor.netty.tcp.TcpServer - [id: 0xf6dd2f6d, L:/127.0.0.1:4059 - R:/127.0.0.1:52044] FLUSH
14:25:04.537 [reactor-tcp-nio-3] DEBUG reactor.netty.tcp.TcpServer - [id: 0xf6dd2f6d, L:/127.0.0.1:4059 - R:/127.0.0.1:52044] READ COMPLETE
14:25:04.537 [reactor-tcp-nio-2] DEBUG reactor.netty.tcp.TcpClient - [id: 0x3d2ab8be, L:/127.0.0.1:52044 - R:localhost/127.0.0.1:4059] READ: 3B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 07 06 05                                        |...             |
+--------+-------------------------------------------------+----------------+
14:25:04.537 [reactor-tcp-nio-3] DEBUG reactor.netty.tcp.TcpServer - [id: 0xf6dd2f6d, L:/127.0.0.1:4059 - R:/127.0.0.1:52044] READ: 3B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 05 06 07                                        |...             |
+--------+-------------------------------------------------+----------------+
14:25:04.538 [reactor-tcp-nio-3] INFO com.example.TcpTest - Server inbound: [5, 6, 7]
14:25:04.538 [reactor-tcp-nio-3] DEBUG reactor.netty.tcp.TcpServer - [id: 0xf6dd2f6d, L:/127.0.0.1:4059 - R:/127.0.0.1:52044] WRITE: 3B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 09 08 07                                        |...             |
+--------+-------------------------------------------------+----------------+
14:25:04.538 [reactor-tcp-nio-3] DEBUG reactor.netty.tcp.TcpServer - [id: 0xf6dd2f6d, L:/127.0.0.1:4059 - R:/127.0.0.1:52044] FLUSH
14:25:04.538 [reactor-tcp-nio-2] DEBUG reactor.netty.channel.FluxReceive - [id: 0x3d2ab8be, L:/127.0.0.1:52044 - R:localhost/127.0.0.1:4059] Dropping frame PooledUnsafeDirectByteBuf(ridx: 0, widx: 3, cap: 1024), 0 in buffer
14:25:04.539 [reactor-tcp-nio-2] DEBUG reactor.netty.tcp.TcpClient - [id: 0x3d2ab8be, L:/127.0.0.1:52044 - R:localhost/127.0.0.1:4059] READ COMPLETE
14:25:04.539 [reactor-tcp-nio-3] DEBUG reactor.netty.tcp.TcpServer - [id: 0xf6dd2f6d, L:/127.0.0.1:4059 - R:/127.0.0.1:52044] READ COMPLETE
14:25:04.539 [reactor-tcp-nio-2] DEBUG reactor.netty.tcp.TcpClient - [id: 0x3d2ab8be, L:/127.0.0.1:52044 - R:localhost/127.0.0.1:4059] READ: 3B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 09 08 07                                        |...             |
+--------+-------------------------------------------------+----------------+
14:25:04.539 [reactor-tcp-nio-2] DEBUG reactor.netty.channel.FluxReceive - [id: 0x3d2ab8be, L:/127.0.0.1:52044 - R:localhost/127.0.0.1:4059] Dropping frame PooledUnsafeDirectByteBuf(ridx: 0, widx: 3, cap: 512), 0 in buffer
14:25:04.539 [reactor-tcp-nio-2] DEBUG reactor.netty.tcp.TcpClient - [id: 0x3d2ab8be, L:/127.0.0.1:52044 - R:localhost/127.0.0.1:4059] READ COMPLETE

有人可以指出正确的方向如何正确链接TcpClient send -> receive -> send -> receive 并返回最后收到的值吗？

【问题讨论】：

标签： java reactive-programming project-reactor reactor-netty

【解决方案1】：

想通了。问题中的解决方案不起作用的原因是因为您不能两次调用connection.inbound().receive()，而只能调用一次，并且必须在该调用中处理整个通信流程。

我写了一篇正确解决方法的文章：How to Implement a Custom Handshaking Protocol via TCP Using Reactor Netty

【讨论】：