Netty是Java生态最流行的网络框架，理解其线程模型和高性能原理是开发高性能应用的基础。
本文深入剖析Netty的线程模型、核心组件和高性能原理，帮助你掌握网络编程的精髓。
| 模型 | 特点 | 优点 | 缺点 |
|---|
| 传统BIO | 一连接一线程 | 简单 | 资源浪费 |
| NIO | 多路复用 | 高效 | 编程复杂 |
| AIO | 异步非阻塞 | 理论最优 | 实现复杂 |
Reactor模式：
单Reactor单线程：
┌─────────────────────────────────────────────────────────────┐
│ Reactor线程 │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────┐ │
│ │ 事件循环 │ │ Accept │ │ Read/Write │ │
│ │ (select) │→ │ 连接 │→ │ 处理 │ │
│ └─────────────┘ └─────────────┘ └─────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
主从Reactor多线程：
┌─────────────────────────────────────────────────────────────┐
│ 主Reactor │ 从Reactor线程池 │
│ ┌─────────────┐│ ┌─────────────┐ ┌─────────────────────┐│
│ │ 事件循环 ││ │ 从Reactor │ │ Worker线程池 ││
│ │ (select) │→ │ 事件循环 │→ │ 处理业务逻辑 ││
│ └─────────────┘│ └─────────────┘ └─────────────────────┘│
│ Accept连接 │ 处理IO读写 │
└─────────────────────────────────────────────────────────────┘
Netty线程模型：
┌─────────────────────────────────────────────────────────────┐
│ Boss Group │
│ ┌─────────────┐ ┌─────────────┐ │
│ │ NioEventLoop│ │ NioEventLoop│ (处理Accept事件) │
│ └─────────────┘ └─────────────┘ │
├─────────────────────────────────────────────────────────────┤
│ Worker Group │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────┐ │
│ │ NioEventLoop│ │ NioEventLoop│ │ NioEventLoop │ │
│ └─────────────┘ └─────────────┘ └─────────────────────┘ │
│ (处理IO读写事件) │
├─────────────────────────────────────────────────────────────┤
│ TaskQueue │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────┐ │
│ │ 普通任务 │ │ 定时任务 │ │ IO事件任务 │ │
│ └─────────────┘ └─────────────┘ └─────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
// NioEventLoop核心逻辑
public final class NioEventLoop extends SingleThreadEventLoop {
@Override
protected void run() {
for (;;) {
try {
// 1. 处理IO事件
selectKeys = selector.select();
// 2. 处理selectedKeys
processSelectedKeys(selectKeys);
// 3. 处理任务队列
runAllTasks();
} catch (Throwable t) {
// 处理异常
}
}
}
private void processSelectedKeys(SelectionKey[] selectedKeys) {
for (SelectionKey k : selectedKeys) {
if (k.isAcceptable()) {
// 处理Accept事件
accept(k);
}
if (k.isReadable()) {
// 处理Read事件
read(k);
}
if (k.isWritable()) {
// 处理Write事件
write(k);
}
}
}
}
Netty组件关系：
┌─────────────────────────────────────────────────────────────┐
│ ServerBootstrap │
│ ├─ group(bossGroup, workerGroup) │
│ ├─ channel(NioServerSocketChannel.class) │
│ ├─ childHandler(new ChannelInitializer<SocketChannel>()) │
│ └─ bind(8080) │
├─────────────────────────────────────────────────────────────┤
│ ChannelPipeline │
│ ├─ addLast(new LengthFieldBasedFrameDecoder()) │
│ ├─ addLast(new StringDecoder()) │
│ ├─ addLast(new ServerHandler()) │
│ └─ ... │
├─────────────────────────────────────────────────────────────┤
│ ChannelHandler │
│ ├─ channelActive() → 连接建立 │
│ ├─ channelRead() → 接收数据 │
│ ├─ channelInactive() → 连接断开 │
│ └─ exceptionCaught() → 异常处理 │
└─────────────────────────────────────────────────────────────┘
| 组件 | 作用 |
|---|
| EventLoopGroup | 事件循环组 |
| Channel | 网络连接 |
| ChannelPipeline | 处理器链 |
| ChannelHandler | 处理器 |
| ByteBuf | 字节缓冲 |
public class NettyServer {
public static void main(String[] args) throws Exception {
// Boss Group：处理Accept事件
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
// Worker Group：处理IO读写事件
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) {
ch.pipeline().addLast(new ServerHandler());
}
});
// 绑定端口
ChannelFuture f = b.bind(8080).sync();
// 等待服务端关闭
f.channel().closeFuture().sync();
} finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
}
}
public class NettyClient {
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
public void initChannel(SocketChannel ch) {
ch.pipeline().addLast(new ClientHandler());
}
});
// 连接服务端
ChannelFuture f = b.connect("localhost", 8080).sync();
// 发送消息
f.channel().writeAndFlush("Hello Netty");
// 等待连接关闭
f.channel().closeFuture().sync();
} finally {
group.shutdownGracefully();
}
}
}
// Netty零拷贝实现
// 1. 文件传输
FileRegion region = new DefaultFileRegion(fileChannel, 0, fileChannel.size());
channel.writeAndFlush(region);
// 2. 组合ByteBuf
CompositeByteBuf composite = compositeBuffer();
composite.addComponent(true, buf1);
composite.addComponent(true, buf2);
// 3. ByteBuf切片
ByteBuf sliced = buf.slice(0, 100);
// Netty内存池
// 1. 池化ByteBuf
ByteBuf buf = PooledByteBufAllocator.DEFAULT.buffer(1024);
// 2. 非池化ByteBuf
ByteBuf buf = UnpooledByteBufAllocator.DEFAULT.buffer(1024);
// 3. 内存池优势
// - 减少GC压力
// - 提高性能
// - 内存复用
// Netty无锁设计
// 1. 单线程处理
// 每个EventLoop单线程，避免锁竞争
// 2. 任务队列
// 使用MPSC队列(多生产者单消费者)
// 3. 事件驱动
// 基于事件的异步处理
| 编解码器 | 作用 | 使用场景 |
|---|
| StringDecoder | 字符串解码 | 文本协议 |
| LengthFieldBasedFrameDecoder | 长度字段解码 | 自定义协议 |
| HttpRequestDecoder | HTTP请求解码 | HTTP服务 |
| ProtobufDecoder | Protobuf解码 | 高性能协议 |
// 自定义协议编解码器
public class MyMessageDecoder extends ByteToMessageDecoder {
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
// 读取魔数
int magic = in.readShort();
// 读取消息类型
byte type = in.readByte();
// 读取数据长度
int length = in.readInt();
// 读取数据
byte[] data = new byte[length];
in.readBytes(data);
// 构建消息对象
MyMessage msg = new MyMessage(type, data);
out.add(msg);
}
}
public class MyMessageEncoder extends MessageToByteEncoder<MyMessage> {
@Override
protected void encode(ChannelHandlerContext ctx, MyMessage msg, ByteBuf out) {
// 写入魔数
out.writeShort(0x1234);
// 写入消息类型
out.writeByte(msg.getType());
// 写入数据长度
out.writeInt(msg.getData().length);
// 写入数据
out.writeBytes(msg.getData());
}
}
public class ChatServerHandler extends SimpleChannelInboundHandler<String> {
private static final ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
@Override
public void channelActive(ChannelHandlerContext ctx) {
channels.add(ctx.channel());
broadcast("用户 " + ctx.channel().remoteAddress() + " 加入聊天室");
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) {
broadcast("[" + ctx.channel().remoteAddress() + "]: " + msg);
}
@Override
public void channelInactive(ChannelHandlerContext ctx) {
channels.remove(ctx.channel());
broadcast("用户 " + ctx.channel().remoteAddress() + " 离开聊天室");
}
private void broadcast(String message) {
for (Channel channel : channels) {
channel.writeAndFlush(message);
}
}
}
public class HttpServerHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) {
// 处理请求
String uri = request.uri();
String method = request.method().name();
// 构建响应
FullHttpResponse response = new DefaultFullHttpResponse(
HttpVersion.HTTP_1_1,
HttpResponseStatus.OK,
Unpooled.copiedBuffer("Hello Netty", CharsetUtil.UTF_8)
);
response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain");
response.headers().set(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());
ctx.writeAndFlush(response);
}
}
| 优化点 | 说明 |
|---|
| 合理配置线程数 | Boss=1, Worker=CPU核数*2 |
| 使用内存池 | 减少GC压力 |
| 零拷贝 | 文件传输使用FileRegion |
| 合理使用ByteBuf | 避免频繁创建销毁 |
| 问题 | 原因 | 解决方案 |
|---|
| 内存泄漏 | 未释放ByteBuf | 使用try-finally |
| 线程阻塞 | 在EventLoop中执行耗时操作 | 使用业务线程池 |
| 连接泄漏 | 未关闭Channel | 在finally中关闭 |
