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

Question

我需要在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并返回最后收到的值？

Answer 1

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

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