如何在Reactive Netty非阻塞场景下关闭连接并返回Mono<byte[]>

问题分析 #

你当前的核心问题是TCP连接未正确触发onComplete()，导致Mono<byte[]>一直处于未完成状态，最终HTTP请求超时。原代码和编辑后的代码存在两个关键问题：

• 原代码中single()要求Flux必须恰好发射一个元素并完成，但TCP inbound的Flux不会自动完成，除非连接被关闭
• 编辑后的代码中过早调用conn.dispose()，可能导致数据还没完全接收就断开连接，引发source was empty错误

正确的非阻塞实现方案 #

要实现类似Reactive Mongo的声明式非阻塞风格，关键是确保在数据接收完成后再释放连接，同时正确处理TCP流的完成信号。可以通过doFinally保证连接无论成功失败都会被释放，并用singleOrEmpty处理可能的空响应，或者根据TCP协议约定判断响应结束（比如固定长度、特定分隔符）。

方案一：基于连接自动释放的基础实现 #

TcpClient.create()
         .host(host)
         .port(port)
         .connect()
         .flatMap(conn -> {
             // 发送请求后接收响应，转换为Mono<byte[]>
             Mono<byte[]> responseMono = conn.outbound()
                 .sendByteArray(Mono.just(yourRequestBytes)) // 替换为实际请求字节数组
                 .then()
                 .thenMany(conn.inbound().receive().asByteArray())
                 .singleOrEmpty() // 处理空响应场景，避免source empty错误
                 .doFinally(signalType -> conn.dispose()); // 无论结果如何，最终释放连接

             return responseMono;
         });

方案二：基于TCP协议约定的响应结束判断（推荐） #

如果TCP服务器的响应有明确结束标识（比如固定长度、换行符），可以通过takeUntil主动终止Flux，触发onComplete()：

// 示例：假设响应以换行符\n作为结束标识
TcpClient.create()
         .host(host)
         .port(port)
         .connect()
         .flatMap(conn -> {
             Flux<byte[]> responseFlux = conn.outbound()
                 .sendByteArray(Mono.just(yourRequestBytes))
                 .then()
                 .thenMany(conn.inbound().receive().asByteArray())
                 .takeUntil(bytes -> new String(bytes).contains("\n")); // 根据协议判断响应结束

             // 将分段的响应字节数组合并为单个byte数组
             Mono<byte[]> responseMono = responseFlux
                 .reduce((acc, bytes) -> {
                     byte[] merged = new byte[acc.length + bytes.length];
                     System.arraycopy(acc, 0, merged, 0, acc.length);
                     System.arraycopy(bytes, 0, merged, acc.length, bytes.length);
                     return merged;
                 })
                 .doFinally(signalType -> conn.dispose());

             return responseMono;
         });

关键注意事项 #

• 禁止使用connectNow()：这是阻塞操作，违背Reactive非阻塞原则，必须使用connect()返回的Mono<Connection>
• 连接释放时机：必须在响应接收完成后再释放，doFinally是最安全的方式，能覆盖成功、失败、取消等所有场景
• 响应处理：如果服务器不会主动关闭连接，必须通过协议约定终止Flux，否则single()会一直等待导致超时

内容的提问来源于stack exchange，提问作者roroWorld