Java NIO：TCP 服务端与客户端、半包处理及非阻塞写入
一、NIO TCP 服务端的工作原理
Java NIO 通过 Selector、Channel 和 Buffer 实现非阻塞网络通信。
三个核心组件：
NIO 服务端可以通过一个 Selector 管理多个客户端连接，不需要为每个客户端单独创建一个阻塞等待线程。
基本执行流程：
创建 Selector
|
v
创建 ServerSocketChannel
|
v
configureBlocking(false)
|
v
注册 OP_ACCEPT
|
v
selector.select() 等待事件
|
+---- OP_ACCEPT
| |
| v
| accept() 接收连接
| |
| v
| 注册 OP_READ
|
+---- OP_READ
| |
| v
| channel.read(buffer)
| |
| v
| 解析业务消息
|
+---- OP_WRITE
|
v
channel.write(buffer)
|
v
发送剩余数据其中：
ServerSocketChannel：负责监听和接收 TCP 连接。SocketChannel：负责客户端连接的数据读写。SelectionKey：保存 Channel 在 Selector 上的注册关系及关注的事件。
二、TCP 半包与粘包
TCP 是字节流协议，不保证一次 read() 就能读取一条完整的业务消息。
例如客户端发送两条消息：
消息A：Hello
消息B：World服务端可能接收到：
情况1：
read1：Hello
read2：World
情况2（半包）：
read1：Hel
read2：loWorld
情况3（粘包）：
read1：HelloWorld注意：半包、粘包是业务消息边界的问题，并不是 TCP 传输错误。
解决方案：长度字段协议
常见的解决方式是定义：
+------------------+-------------------+
| BodyLength | Body |
| 4字节 | N字节 |
+------------------+-------------------+例如发送 Hello：
消息头：4字节，表示正文长度5
消息体：5字节，Hello
总长度：9字节发送端：
byte[] body = "Hello".getBytes(StandardCharsets.UTF_8);
ByteBuffer buffer = ByteBuffer.allocate(4 + body.length);
// 消息头
buffer.putInt(body.length);
// 消息正文
buffer.put(body);
buffer.flip();注意：消息长度应使用实际的 UTF-8 字节长度，而不是 String.length()，因为中文字符通常占用多个 UTF-8 字节。
接收端如何判断消息是否完整？
首先判断有没有完整的 4 字节消息头：
if (buffer.remaining() >= 4) {
int bodyLength = buffer.getInt();
}然后判断消息正文是否完整：
if (buffer.remaining() >= bodyLength) {
// 正文已经接收完整
}如果正文不完整，就保留剩余数据，等待后续网络数据到达。
三、compact() 在 TCP 半包处理中的应用
这是 Buffer 和 NIO 网络编程结合的重要知识点。
假设定义的消息协议为：
4字节长度 + 5字节正文但是服务端第一次只收到了：
4字节消息头 + 2字节正文此时不能直接处理消息，因为还差 3 个字节。
需要执行：
// 1. 切换为读取模式
buffer.flip();
// 2. 记录消息头开始的位置
buffer.mark();
// 3. 读取消息长度
int bodyLength = buffer.getInt();
// 4. 判断正文是否完整
if (buffer.remaining() < bodyLength) {
// 恢复到消息头起始位置
buffer.reset();
}
// 5. 保留未处理的数据
buffer.compact();这里有两个关键问题。
1. 为什么需要 mark() 和 reset()？
因为 getInt() 会读取 4 个字节，并将 position 向后移动 4 位。
如果正文尚未完整接收，却直接调用 compact()，那么已读取的 4 字节消息头就不会被保留。
下一次收到后续数据时，可能无法正确识别消息长度。
所以需要通过：
buffer.mark();
buffer.getInt();
buffer.reset();在消息不完整时恢复到消息头的位置。
2. 为什么使用 compact()，而不是 clear()？
因为：
clear()：重置 Buffer 状态，不保留未处理数据的逻辑有效范围。compact()：将未处理的数据移动到开头，腾出后续空间。
对于半包，必须保留尚未处理完整的数据。
标准流程：
接收网络数据
|
v
buffer.flip()
|
v
判断消息是否完整
|
+---- 完整 ----> 解析消息
|
+---- 不完整 --> reset() 恢复位置
|
v
buffer.compact()
|
v
继续接收数据需要注意：一个 Buffer 可能包含多条完整消息，因此实际解析时通常通过 while 循环连续处理，最后再执行 compact()。
四、Java 8：NIO TCP 服务端完整示例
实现功能：
一个 Selector 管理多个客户端。
接收客户端 TCP 消息。
使用 4 字节长度字段解决半包和粘包。
每个客户端保存独立的接收 Buffer。
使用发送队列处理非阻塞部分写入。
使用 OP_WRITE 事件处理待发送数据。
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.StandardCharsets;
import java.util.ArrayDeque;
import java.util.Iterator;
import java.util.Queue;
public class NioEchoServer {
private static final int MAX_BODY = 4096;
private static final int MAX_PENDING = 100;
// 每个客户端独立的连接状态
static class ClientState {
// 接收缓冲区
ByteBuffer readBuffer =
ByteBuffer.allocate(4 + MAX_BODY);
// 待发送队列
Queue<ByteBuffer> writeQueue =
new ArrayDeque<>();
}
public static void main(String[] args)
throws IOException {
try (Selector selector = Selector.open();
ServerSocketChannel server =
ServerSocketChannel.open()) {
server.configureBlocking(false);
server.bind(new InetSocketAddress(8080));
server.register(
selector,
SelectionKey.OP_ACCEPT
);
System.out.println("NIO服务端启动");
while (true) {
selector.select();
Iterator<SelectionKey> iterator =
selector.selectedKeys().iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
iterator.remove();
if (!key.isValid()) {
continue;
}
try {
if (key.isAcceptable()) {
accept(selector, key);
} else {
if (key.isReadable()) {
read(key);
}
if (key.isValid()
&& key.isWritable()) {
write(key);
}
}
} catch (IOException e) {
key.cancel();
key.channel().close();
System.out.println(
"IO异常：" + e.getMessage()
);
}
}
}
}
}
// 接收客户端连接
private static void accept(
Selector selector,
SelectionKey key) throws IOException {
ServerSocketChannel server =
(ServerSocketChannel) key.channel();
SocketChannel client;
while ((client = server.accept()) != null) {
client.configureBlocking(false);
client.register(
selector,
SelectionKey.OP_READ,
new ClientState()
);
System.out.println("客户端已连接");
}
}
// 读取客户端数据
private static void read(SelectionKey key)
throws IOException {
SocketChannel client =
(SocketChannel) key.channel();
ClientState state =
(ClientState) key.attachment();
ByteBuffer buffer = state.readBuffer;
int length = client.read(buffer);
if (length == -1) {
key.cancel();
client.close();
return;
}
if (length == 0) {
return;
}
buffer.flip();
// 一个Buffer中可能包含多条完整消息
while (buffer.remaining() >= 4) {
// 标记消息头起点
buffer.mark();
int bodyLength = buffer.getInt();
if (bodyLength < 0
|| bodyLength > MAX_BODY) {
throw new IOException("非法消息长度");
}
// 半包：恢复到消息头起点
if (buffer.remaining() < bodyLength) {
buffer.reset();
break;
}
// 读取完整正文
byte[] body = new byte[bodyLength];
buffer.get(body);
String message = new String(
body,
StandardCharsets.UTF_8
);
System.out.println("收到：" + message);
// 构造响应
byte[] response = ("收到：" + message)
.getBytes(StandardCharsets.UTF_8);
if (response.length > MAX_BODY ||
state.writeQueue.size() >= MAX_PENDING) {
throw new IOException("响应超出限制");
}
ByteBuffer output = ByteBuffer.allocate(
4 + response.length
);
output.putInt(response.length);
output.put(response);
output.flip();
state.writeQueue.offer(output);
// 开启可写事件
key.interestOps(
key.interestOps()
| SelectionKey.OP_WRITE
);
}
// 保留未处理的数据
buffer.compact();
}
// 处理部分写入
private static void write(SelectionKey key)
throws IOException {
SocketChannel client =
(SocketChannel) key.channel();
ClientState state =
(ClientState) key.attachment();
while (!state.writeQueue.isEmpty()) {
ByteBuffer buffer =
state.writeQueue.peek();
client.write(buffer);
// 没有写完，等待下一次OP_WRITE
if (buffer.hasRemaining()) {
return;
}
// 当前消息已发送完毕
state.writeQueue.poll();
}
// 没有待发送数据，关闭OP_WRITE关注
key.interestOps(
key.interestOps()
& ~SelectionKey.OP_WRITE
);
}
}代码重点
1. 为什么使用 SelectionKey.attachment()？
client.register(
selector,
SelectionKey.OP_READ,
new ClientState()
);因为每个客户端的 Buffer、半包状态和待发送数据都不同。
attachment() 可以获取该客户端独立的业务状态：
ClientState state =
(ClientState) key.attachment();这样就不用将所有连接的数据放到一个共享 Buffer 中。
2. 为什么 OP_READ 中需要 while 循环解析？
因为一次 channel.read() 可能接收到多条完整业务消息。
while (buffer.remaining() >= 4) {
// 尝试解析一条完整消息
}循环可以连续解析完整消息，遇到不完整消息时停止，并使用 compact() 保留剩余内容。
3. 为什么 write() 还需要判断 hasRemaining()？
因为非阻塞 SocketChannel 的一次 write() 不保证写完全部数据。
client.write(buffer);
if (buffer.hasRemaining()) {
return;
}如果还有剩余数据，Buffer 会保存在队列中，等待下一次 OP_WRITE 事件继续发送。
五、Java 8：TCP 客户端示例
为了简化示例，客户端使用 SocketChannel 的默认阻塞模式，服务端使用非阻塞 Selector。
import java.io.EOFException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;
public class NioEchoClient {
public static void main(String[] args)
throws Exception {
try (SocketChannel channel = SocketChannel.open(
new InetSocketAddress("127.0.0.1", 8080))) {
send(channel, "Hello NIO");
send(channel, "Hello Selector");
receive(channel);
receive(channel);
}
}
private static void send(
SocketChannel channel,
String message) throws Exception {
byte[] data = message.getBytes(
StandardCharsets.UTF_8
);
ByteBuffer buffer =
ByteBuffer.allocate(4 + data.length);
buffer.putInt(data.length);
buffer.put(data);
buffer.flip();
while (buffer.hasRemaining()) {
channel.write(buffer);
}
}
private static void receive(
SocketChannel channel) throws Exception {
// 读取4字节长度字段
ByteBuffer header = ByteBuffer.allocate(4);
readFully(channel, header);
header.flip();
int bodyLength = header.getInt();
if (bodyLength < 0 || bodyLength > 4096) {
throw new IllegalStateException("非法消息长度");
}
// 读取正文
ByteBuffer body =
ByteBuffer.allocate(bodyLength);
readFully(channel, body);
body.flip();
byte[] bytes = new byte[body.remaining()];
body.get(bytes);
System.out.println(
new String(bytes, StandardCharsets.UTF_8)
);
}
private static void readFully(
SocketChannel channel,
ByteBuffer buffer) throws Exception {
while (buffer.hasRemaining()) {
int length = channel.read(buffer);
if (length == -1) {
throw new EOFException("连接结束");
}
}
}
}预期输出：
收到：Hello NIO
收到：Hello Selector这里使用 readFully() 是因为 TCP 一次读取不保证拿到全部数据。
客户端采用阻塞模式，因此可以循环等待指定长度的数据。非阻塞客户端则应配合 Selector，避免没有数据时持续空转。
六、非阻塞写入与 OP_WRITE 的注意事项
OP_WRITE 是 NIO 开发中非常容易踩坑的地方。
1. 为什么不能一直关注 OP_WRITE？
因为 SocketChannel 通常只要发送缓冲区有剩余空间，就会处于可写状态。
即使没有待发送数据，OP_WRITE 也可能持续就绪。
selector.select()
|
v
OP_WRITE就绪
|
v
处理写事件
|
v
再次select()
|
v
OP_WRITE再次就绪
|
v
CPU不断空转正确处理：
有剩余数据需要继续发送时：
key.interestOps(
key.interestOps() | SelectionKey.OP_WRITE
);待发送队列清空后：
key.interestOps(
key.interestOps() & ~SelectionKey.OP_WRITE
);重点：OP_WRITE 不是一有业务消息就必须注册，而是在非阻塞写入需要等待可写事件时使用。
如果数据能够立即全部写完，也可以不启用 OP_WRITE。
2. 为什么使用 writeQueue？
假设需要发送 1000 字节：
第一次write()：成功发送400字节
剩余：600字节此时不能直接丢弃 Buffer。
因为 position 已经移动到 400，limit 仍然是 1000。
需要保留这个 Buffer，下一次可写时继续发送剩余的 600 字节。
ByteBuffer buffer = state.writeQueue.peek();
client.write(buffer);
if (buffer.hasRemaining()) {
return;
}
state.writeQueue.poll();这样可以避免数据丢失，也不会重新发送已经写入的部分。
3. 为什么需要限制发送队列？
如果客户端接收速度很慢，而服务端持续产生消息，发送队列可能不断增长。
这会导致内存占用持续增加，甚至引起 OOM。
生产环境应设置发送队列大小或待发送字节数上限，并根据业务采用背压、暂停读取、丢弃或断开慢客户端等策略。
七、重点总结
NIO TCP 服务端通过 Selector 管理多个 SocketChannel。
OP_ACCEPT 负责处理客户端连接。
OP_READ 负责接收客户端数据。
OP_WRITE 负责处理需要继续发送的数据。
TCP 是字节流协议，不保证一次 read() 对应一条完整消息。
可以通过 4 字节长度字段 + 消息正文解决半包和粘包。
mark() 和 reset() 可用于在消息不完整时恢复消息头位置。
compact() 用于保留未处理的半包数据，等待后续接收。
SelectionKey.attachment() 可以保存每个客户端独立的连接状态。
非阻塞 write() 不保证一次写完所有数据。
writeQueue 用于保存未发送完的消息。
待发送队列为空时，应取消 OP_WRITE，避免 CPU 空转。
Selector 线程应避免执行耗时数据库查询、HTTP 请求等阻塞操作。
八、常见问题与答案
1. 为什么一个客户端需要一个独立的 ByteBuffer？
因为每个客户端的 TCP 数据接收进度不同，可能分别处于完整消息、半包或等待新数据等状态。
2. 为什么 TCP 半包处理中需要 compact()？
因为某条业务消息可能尚未接收完整，需要保留已经收到但尚未处理的数据，等待后续数据到达。
3. 为什么读取长度字段后需要 reset()？
如果正文不完整，reset() 可以恢复到消息头起点，使 compact() 同时保留消息头和已接收的正文数据。
4. 为什么一次 read() 后需要循环解析消息？
因为一次 TCP 读取可能包含多条完整业务消息，需要连续提取，直到剩余数据不足以组成完整消息。
5. 为什么非阻塞 write() 需要发送队列？
因为一次 write() 可能只能写入部分数据，发送队列用于保存尚未发送完的数据，以便后续继续写入。
6. 为什么不推荐长期注册 OP_WRITE？
因为 SocketChannel 通常很容易处于可写状态，持续监听会产生大量不必要的就绪事件，可能导致 CPU 空转。
7. Selector 能不能直接处理数据库业务？
技术上可以，但不推荐在 IO 事件循环中执行耗时阻塞操作，否则可能影响其他连接的响应速度。可以将耗时业务交给受控的业务线程池。
8. 这套代码能否直接用于生产环境？
不建议直接使用。这是用于理解 Selector、Channel、Buffer 和消息编解码的示例。生产环境还需要完善连接超时、心跳、背压、资源管理、异常恢复和性能监控等能力。