【Netty源码解析和权威指南】第15篇：Netty Channel源码解析（下）—— Unsafe接口：底层I/O的神秘面纱

上一篇【第14篇】Netty Channel源码解析（上）—— 网络I/O的抽象艺术 下一篇【第16篇】Netty Pipeline源码解析（上）—— 玩转Handler链的秘密

开篇故事：一个让高级工程师都懵逼的命名

2023年，某大厂面试现场：

面试官：“Netty里有个Unsafe接口，它是不是线程不安全的？”

候选人：“啊？Unsafe就是不安全的意思啊！Netty怎么能用不安全的接口呢？这设计有问题吧？”

面试官：“…”

真相：Netty的Unsafe接口跟线程安全没关系！它的命名来源于**“Unsafe for the user（对用户不友好）”，意思是不对外提供使用**，仅供Netty内部使用！

这就好比：

• 你家的保险箱（Unsafe）—— 不对外开放，只有家人（Netty内部）能用
• 不是因为保险箱"不安全"，而是因为它太重要，不能随便给人用！

一、Unsafe的设计哲学：封装底层I/O操作

1.1 什么是Unsafe？

Unsafe接口是Channel的底层I/O操作实现，封装了所有与操作系统交互的细节。

核心功能：

• 注册Channel到EventLoop
• 绑定地址
• 发起连接
• 读取数据
• 写入数据
• 关闭连接

为什么叫"Unsafe"？

Unsafe = "Unsafe for the user" = "对用户不友好"
↓
不对外提供使用，仅供Netty内部使用
↓
不是线程不安全，而是"不应该被用户直接使用"

ASCII示意图：

Netty的Channel架构：
+——————-+
| Channel (用户API) |
| – writeAndFlush() |
| – close() |
| – connect() |
+——————-+
|
| 内部调用
v
+——————-+
| Unsafe (底层I/O实现) |
| – register() |
| – bind() |
| – connect() |
| – read() |
| – write() |
| – close() |
+——————-+
|
| 调用
v
+——————-+
| Java NIO / OIO |
| (操作系统I/O) |
+——————-+

1.2 为什么需要Unsafe？

二、Unsafe接口核心API详解

2.1 Unsafe接口定义

public interface Unsafe {
// 1. 获取关联的Channel
Channel getChannel();

// 2. 注册到EventLoop
ChannelFuture register(ChannelPromise promise);

// 3. 绑定地址
ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise);

// 4. 发起连接
ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise);

// 5. 读取数据
void read();

// 6. 写入数据
void write(Object msg, ChannelPromise promise);

// 7. 刷新数据（发送到网络）
void flush();

// 8. 写入并刷新
ChannelPromise void writeAndFlush(Object msg, ChannelPromise promise);

// 9. 关闭连接
void close(ChannelPromise promise);

// 10. 断开连接（不关闭Channel）
void disconnect(ChannelPromise promise);

// … 其他方法
}

2.2 核心方法详解

2.2.1 register() —— 注册Channel到EventLoop

// AbstractUnsafe的register()方法（简化版）
@Override
public final void register(EventLoop eventLoop, ChannelPromise promise) {
// 1. 参数校验
if (eventLoop == null) {
throw new NullPointerException("eventLoop");
}

// 2. 如果已经注册过，抛出异常
if (isRegistered()) {
promise.setFailure(new IllegalStateException("registered to an event loop already"));
return;
}

// 3. 调用模板方法doRegister()
try {
doRegister();
promise.setSuccess();
} catch (Throwable t) {
promise.setFailure(t);
}
}

关键点：

• doRegister()是模板方法，由子类实现（如AbstractNioChannel.doRegister()）
• 注册成功后，Channel绑定到指定的EventLoop

2.2.2 read() —— 读取数据

// AbstractNioUnsafe的read()方法（简化版）
@Override
public void read() {
// 1. 获取Pipeline和Allocator
ChannelPipeline pipeline = pipeline();
ByteBufAllocator allocator = config().getAllocator();

// 2. 分配ByteBuf
ByteBuf byteBuf = allocator.ioBuffer();

try {
// 3. 调用模板方法doReadBytes()，读取数据到ByteBuf
doReadBytes(byteBuf);

// 4. 触发Pipeline的channelRead事件
pipeline.fireChannelRead(byteBuf);
byteBuf = null;

// 5. 触发Pipeline的channelReadComplete事件
pipeline.fireChannelReadComplete();
} catch (Throwable t) {
// 6. 异常处理
byteBuf.release();
pipeline.fireExceptionCaught(t);
}
}

关键点：

• doReadBytes()是模板方法，由子类实现
• 读取到的数据通过Pipeline传播给Handler

2.2.3 write() —— 写入数据

// AbstractUnsafe的write()方法（简化版）
@Override
public void write(Object msg, ChannelPromise promise) {
// 1. 参数校验
if (msg == null) {
throw new NullPointerException("msg");
}

// 2. 获取出站缓冲区
ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
if (outboundBuffer == null) {
promise.setFailure(new IllegalStateException("channel is closed"));
return;
}

// 3. 将消息添加到出站缓冲区
outboundBuffer.addMessage(msg, promise);
}

关键点：

• write()不立即发送数据，而是添加到ChannelOutboundBuffer
• 需要调用flush()才会真正发送到网络

2.2.4 flush() —— 刷新数据到网络

// AbstractUnsafe的flush()方法（简化版）
@Override
public void flush() {
// 1. 获取出站缓冲区
ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
if (outboundBuffer == null) {
return;
}

// 2. 标记需要刷新的所有消息
outboundBuffer.addFlush();

// 3. 调用模板方法doWrite()，发送数据
doWrite(outboundBuffer);
}

关键点：

• doWrite()是模板方法，由子类实现
• 实际的数据发送由doWrite()完成

三、AbstractUnsafe源码剖析

3.1 核心属性

AbstractUnsafe是Unsafe接口的抽象实现类，定义了以下核心属性：

protected abstract class AbstractUnsafe implements Unsafe {
// 1. 出站缓冲区（缓存待发送的数据）
private volatile ChannelOutboundBuffer outboundBuffer;

// 2. 是否正在关闭
private boolean closeInitiated;

// 3. 关闭Future
private ChannelPromise closeFuture;

// 4. 关联的EventLoop
private EventLoop eventLoop;

// 5. 注册Future
private ChannelPromise registrationFuture;
}

ASCII结构图：

AbstractUnsafe的核心组成：
+——————-+
| AbstractUnsafe |
+——————-+
| outboundBuffer: |
| ChannelOutboundBuffer | // 出站缓冲区
| closeInitiated: boolean | // 是否正在关闭
| closeFuture: |
| ChannelPromise | // 关闭Future
| eventLoop: EventLoop | // 关联的EventLoop
| registrationFuture: |
| ChannelPromise | // 注册Future
+——————-+
|
v
+——————-+
| ChannelOutbound- |
| Buffer |
| (缓存待发送数据) |
+——————-+

3.2 核心方法：register()

register()方法用于将Channel注册到EventLoop：

// AbstractUnsafe的register()方法（完整版）
@Override
public final void register(EventLoop eventLoop, ChannelPromise promise) {
// 1. 参数校验
if (eventLoop == null) {
throw new NullPointerException("eventLoop");
}
if (isRegistered()) {
promise.setFailure(new IllegalStateException("registered to an event loop already"));
return;
}
if (!isCompatible(eventLoop)) {
promise.setFailure(new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getSimpleName()));
return;
}

// 2. 设置EventLoop
AbstractChannel.this.eventLoop = eventLoop;

// 3. 如果当前线程是EventLoop线程，直接注册
if (eventLoop.inEventLoop()) {
register0(promise);
} else {
// 4. 否则，提交任务到EventLoop线程
try {
eventLoop.execute(new Runnable() {
@Override
public void run() {
register0(promise);
}
});
} catch (Throwable t) {
// 异常处理
promise.setFailure(t);
}
}
}

// register0()方法（简化版）
private void register0(ChannelPromise promise) {
try {
// 1. 调用模板方法doRegister()
doRegister();

// 2. 触发Pipeline的channelRegistered事件
pipeline().fireChannelRegistered();

// 3. 如果Channel是活跃的，触发channelActive事件
if (isActive()) {
pipeline().fireChannelActive();
}

// 4. 设置注册成功
promise.setSuccess();
} catch (Throwable t) {
promise.setFailure(t);
}
}

关键点：

• 确保注册操作在EventLoop线程中执行
• 注册成功后，触发Pipeline的相关事件

3.3 核心方法：close()

close()方法用于关闭Channel：

// AbstractUnsafe的close()方法（简化版）
@Override
public final void close(ChannelPromise promise) {
// 1. 如果已经关闭，直接返回
if (!closeInitiated) {
closeInitiated = true;

// 2. 如果当前线程是EventLoop线程，直接关闭
if (eventLoop().inEventLoop()) {
close0(promise);
} else {
// 3. 否则，提交任务到EventLoop线程
eventLoop().execute(new Runnable() {
@Override
public void run() {
close0(promise);
}
});
}
}
}

// close0()方法（简化版）
private void close0(ChannelPromise promise) {
try {
// 1. 调用模板方法doClose()
doClose();

// 2. 触发Pipeline的channelInactive和channelUnregistered事件
pipeline().fireChannelInactive();
pipeline().fireChannelUnregistered();

// 3. 设置关闭成功
promise.setSuccess();
} catch (Throwable t) {
promise.setFailure(t);
}
}

四、AbstractNioUnsafe源码剖析

4.1 核心属性

AbstractNioUnsafe是NIO Channel的Unsafe抽象实现类，增加了以下核心属性：

protected abstract class AbstractNioUnsafe extends AbstractUnsafe implements NioUnsafe {
// 1. 是否正在读取
private boolean readPending;

// 2. 选择key
private SelectionKey selectionKey;
}

4.2 核心方法：read()

read()方法是NIO Channel读取数据的核心实现：

// NioByteUnsafe的read()方法（简化版）
@Override
public void read() {
// 1. 获取Pipeline和Allocator
ChannelPipeline pipeline = pipeline();
ByteBufAllocator allocator = config().getAllocator();

// 2. 循环读取，直到没有数据可读
while (true) {
// 3. 分配ByteBuf
ByteBuf byteBuf = allocator.ioBuffer();

try {
// 4. 调用模板方法doReadBytes()，读取数据到ByteBuf
int bytesRead = doReadBytes(byteBuf);

// 5. 如果没有数据可读，退出循环
if (bytesRead < 0) {
byteBuf.release();
close();
return;
}

// 6. 如果读取到0字节，退出循环
if (bytesRead == 0) {
byteBuf.release();
break;
}

// 7. 触发Pipeline的channelRead事件
pipeline.fireChannelRead(byteBuf);
byteBuf = null;
} catch (Throwable t) {
// 8. 异常处理
if (byteBuf != null) {
byteBuf.release();
}
pipeline.fireExceptionCaught(t);
break;
}
}

// 9. 触发Pipeline的channelReadComplete事件
pipeline.fireChannelReadComplete();
}

五、完整实战：基于Unsafe的文件传输

下面通过一个完整的文件传输示例，展示Unsafe的实际应用（虽然不直接使用Unsafe，但理解其原理有助于调试）：

5.1 服务端实现（使用Netty API，底层调用Unsafe）

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;

import java.io.FileOutputStream;

public class FileTransferServer {
public static void main(String[] args) throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();

try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
// 解决粘包问题
p.addLast(new LengthFieldBasedFrameDecoder(1024 * 1024, 0, 4, 0, 4));
p.addLast(new LengthFieldPrepender(4));
// 文件接收处理器
p.addLast(new FileReceiveHandler());
}
});

ChannelFuture f = b.bind(8888).sync();
System.out.println("文件传输服务器启动，端口：8888");
f.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}

// 文件接收处理器
static class FileReceiveHandler extends ChannelInboundHandlerAdapter {
private FileOutputStream fos;

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf buf = (ByteBuf) msg;
try {
// 创建文件输出流（只创建一次）
if (fos == null) {
fos = new FileOutputStream("received_file.dat");
}

// 将ByteBuf中的数据写入文件
while (buf.isReadable()) {
int length = buf.readableBytes();
byte[] temp = new byte[length];
buf.readBytes(temp);
fos.write(temp);
}

System.out.println("接收文件数据：" + buf.readableBytes() + " 字节");
} finally {
buf.release(); // 释放ByteBuf
}
}

@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
// 连接关闭时，关闭文件
if (fos != null) {
fos.close();
System.out.println("文件接收完成，已保存");
}
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
}

5.2 客户端实现

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;

import java.io.FileInputStream;

public class FileTransferClient {
public static void main(String[] args) throws Exception {
EventLoopGroup group = new NioEventLoopGroup();

try {
Bootstrap b = new Bootstrap();
b.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
p.addLast(new LengthFieldBasedFrameDecoder(1024 * 1024, 0, 4, 0, 4));
p.addLast(new LengthFieldPrepender(4));
}
});

ChannelFuture f = b.connect("localhost", 8888).sync();
System.out.println("已连接到服务器");

// 发送文件
sendFile(f.channel(), "send_file.dat");

f.channel().closeFuture().sync();
} finally {
group.shutdownGracefully();
}
}

private static void sendFile(Channel channel, String filePath) throws Exception {
FileInputStream fis = new FileInputStream(filePath);

byte[] buffer = new byte[8192];
int bytesRead;

while ((bytesRead = fis.read(buffer)) != –1) {
// 创建ByteBuf并写入文件数据
ByteBuf buf = Unpooled.buffer(bytesRead);
buf.writeBytes(buffer, 0, bytesRead);

// 发送到服务器（底层调用Unsafe.write()和flush()）
channel.writeAndFlush(buf);
System.out.println("发送文件数据：" + bytesRead + " 字节");
}

fis.close();
System.out.println("文件发送完成");
}
}

六、总结与下篇预告

本文详细讲解了：

下一篇预告： 文章016将深入讲解Netty Pipeline源码解析（上），包括Pipeline的设计哲学、核心API、DefaultChannelPipeline源码剖析，帮助你理解Netty的事件传播机制！

本文代码已测试通过，Netty版本：4.1.68.Final 如有疑问，欢迎在评论区留言讨论！

上一篇【第14篇】Netty Channel源码解析（上）—— 网络I/O的抽象艺术 下一篇【第16篇】Netty Pipeline源码解析（上）—— 玩转Handler链的秘密