我需要在由顺序发送 -> 接收 -> 发送 -> 接收并返回最后一个接收值组成的 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 没有被执行并且在日志中 窃听 正确显示数据已读取但未发送到该 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 并返回最后接收到的值？
最佳答案
想通了。问题中的解决方案不起作用的原因是因为您不能调用 connection.inbound().receive() 两次，而只能调用一次，并且必须在该调用中处理整个通信流。
我写了一篇带有正确解决方案的文章:How to Implement a Custom Handshaking Protocol via TCP Using Reactor Netty
关于java - 如何使用 reactor-netty TcpClient 链接多个发送和接收操作，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/59426393/