文章标签： #java #netty #nio #网络编程 #reactor #高性能 #面试
- 引言：网络编程框架的本质
- 理论基础：Unix IO模型与Reactor模式
- 演进史：从BIO到Netty
- 核心原理深度解析
- 实战案例：工业级应用
- 对比分析：Netty vs Mina vs Vert.x
- 性能分析：连接管理与内存优化
- 常见陷阱与最佳实践
- 面试题与参考答案
Netty不是简单的"NIO框架"，而是一个事件驱动的异步网络应用框架（Event-Driven Asynchronous Network Application Framework）。
核心认知：
网络编程的演进：
BIO（Blocking IO）：
Socket -> Thread
问题：1连接1线程，线程资源耗尽
NIO（Non-Blocking IO）：
Selector -> 多路复用 -> 单线程处理多连接
问题：编程复杂，需要处理各种边界情况
AIO（Asynchronous IO）：
CompletionHandler -> 操作系统回调
问题：Linux支持不完善，使用少
Netty的定位：
在NIO基础上，提供：
- 统一的异步编程模型（Future/Promise）
- 完善的内存管理（ByteBuf）
- 灵活的协议编解码（Codec）
- 高性能的线程模型（EventLoop）
- 丰富的开箱即用功能（SSL、HTTP、WebSocket等）
关键洞察：Netty的设计哲学是在保持NIO高性能的同时，提供与BIO一样简单的编程模型。
阻塞IO（Blocking IO）：
应用 -> recvfrom -> 内核等待数据 -> 数据拷贝到用户空间 -> 返回
特点：全程阻塞
非阻塞IO（Non-Blocking IO）：
应用 -> recvfrom -> 返回EAGAIN -> 轮询 -> 数据准备好 -> 拷贝 -> 返回
特点：轮询消耗CPU
IO多路复用（IO Multiplexing）：
应用 -> select/poll/epoll -> 等待多个Socket就绪 -> recvfrom -> 返回
特点：一个线程管理多个连接
信号驱动IO（Signal-Driven IO）：
应用 -> 注册信号处理器 -> 内核通知 -> recvfrom -> 返回
特点：Linux支持有限
异步IO（Asynchronous IO）：
应用 -> aio_read -> 立即返回 -> 内核完成拷贝后通知
特点：全程不阻塞，但Linux支持不完善
Netty的选择：IO多路复用（NIO）
原因：跨平台性好，性能高，编程模型成熟
Reactor模式核心组件：
┌─────────────────────────────────────────────┐
│ Reactor │
│ - 监听事件（Accept、Read、Write） │
│ - 分发给对应的Handler │
├─────────────────────────────────────────────┤
│ Acceptor │
│ - 处理连接建立事件 │
│ - 创建Handler并注册到Reactor │
├─────────────────────────────────────────────┤
│ Handler │
│ - 处理读写事件 │
│ - 执行业务逻辑 │
└─────────────────────────────────────────────┘
三种Reactor模型：
单线程Reactor：
Reactor（Accept + Read/Write + Business）
问题：单线程处理所有逻辑，无法利用多核
多线程Reactor：
Main Reactor（Accept）
-> Sub Reactor Pool（Read/Write + Business）
问题：业务逻辑可能阻塞IO线程
主从Reactor多线程（Netty采用）：
Main Reactor（Boss EventLoop - Accept）
-> Sub Reactor Pool（Worker EventLoop - Read/Write）
-> Business Thread Pool（执行业务逻辑）
优势：IO与业务分离，充分利用多核
Proactor模式（AIO采用）：
┌─────────────────────────────────────────────┐
│ Proactor │
│ - 发起异步操作 │
│ - 操作完成后通知CompletionHandler │
├─────────────────────────────────────────────┤
│ CompletionHandler │
│ - 处理异步操作完成事件 │
│ - 已完成数据拷贝，直接处理 │
└─────────────────────────────────────────────┘
与Reactor的区别：
- Reactor：监听就绪事件，应用自己读写数据
- Proactor：发起异步读写，操作系统完成数据拷贝后通知
Netty为什么不使用AIO/Proactor：
- Linux AIO不完善（仅支持文件IO）
- 性能不如NIO + Epoll
- 跨平台性差
Java BIO编程：
ServerSocket server = new ServerSocket(8080);
while (true) {
Socket socket = server.accept(); // 阻塞
new Thread(new Handler(socket)).start(); // 1连接1线程
}
问题：
- 线程资源消耗大
- 上下文切换开销
- C10K问题（1万连接就扛不住）
Java NIO三大核心组件：
- Channel：通道，替代Stream
- Buffer：缓冲区
- Selector：多路复用器
NIO编程：
Selector selector = Selector.open();
ServerSocketChannel serverChannel = ServerSocketChannel.open();
serverChannel.configureBlocking(false);
serverChannel.register(selector, SelectionKey.OP_ACCEPT);
while (true) {
selector.select(); // 阻塞等待事件
Iterator<SelectionKey> keys = selector.selectedKeys().iterator();
while (keys.hasNext()) {
SelectionKey key = keys.next();
if (key.isAcceptable()) {
// 处理连接
} else if (key.isReadable()) {
// 处理读
}
keys.remove();
}
}
问题：
- 编程复杂（需要处理半包、粘包）
- 空轮询Bug（JDK 6之前）
- 内存管理复杂（ByteBuffer限制多）
Apache Mina（2004）：
- 早期NIO框架
- 抽象了IO层
- 但性能不如后来的Netty
JBoss Netty（2004）：
- 2004年由Trustin Lee创建
- 设计目标：高性能、易用性
- 2011年成为Netty项目（独立）
Grizzly（Sun/Oracle）：
- GlassFish服务器的网络层
- 性能不错，但生态不如Netty
Netty 3.x（2011）：
- 基于Java NIO
- 引入ChannelPipeline
- 广泛的协议支持
Netty 4.x（2013）：
- 重大重构
- 引入ByteBuf（替代ByteBuffer）
- 内存池化
- 引用计数
- 性能大幅提升
Netty 5.x（废弃）：
- 尝试统一AIO和NIO
- 2015年废弃，合并回4.x
Netty 4.1.x（2015-至今）：
- HTTP/2支持
- 更好的内存管理
- 性能持续优化
- 成为事实标准（Dubbo、RocketMQ、Elasticsearch等）
Netty架构图：
┌─────────────────────────────────────────────────────────────┐
│ Netty Application │
├─────────────────────────────────────────────────────────────┤
│ Bootstrap / ServerBootstrap │
│ - 配置和启动Channel │
├─────────────────────────────────────────────────────────────┤
│ EventLoopGroup │
│ - BossGroup：接收连接（Accept） │
│ - WorkerGroup：处理IO（Read/Write） │
├─────────────────────────────────────────────────────────────┤
│ Channel │
│ - SocketChannel（客户端） │
│ - ServerSocketChannel（服务端） │
├─────────────────────────────────────────────────────────────┤
│ ChannelPipeline │
│ - ChannelHandler链 │
│ - InboundHandler（入站处理） │
│ - OutboundHandler（出站处理） │
├─────────────────────────────────────────────────────────────┤
│ ByteBuf │
│ - 网络数据缓冲区 │
│ - 引用计数管理 │
│ - 内存池化 │
├─────────────────────────────────────────────────────────────┤
│ Codec │
│ - Encoder（编码器） │
│ - Decoder（解码器） │
│ - 解决粘包/拆包 │
├─────────────────────────────────────────────────────────────┤
│ Transport │
│ - NIO（默认） │
│ - Epoll（Linux） │
│ - KQueue（Mac/BSD） │
└─────────────────────────────────────────────────────────────┘
主从Reactor多线程模型（Netty默认）：
┌─────────────────────────────────────────────────────────────┐
│ Boss EventLoopGroup │
│ （通常1个线程，负责Accept） │
│ EventLoop (NioEventLoop) │
│ │ │
│ ▼ │
│ ┌─────────────────────┐ │
│ │ Selector.select() │ │
│ │ 监听Accept事件 │ │
│ └─────────────────────┘ │
│ │ │
│ ▼ │
│ 创建SocketChannel，注册到WorkerGroup │
└─────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ Worker EventLoopGroup │
│ （通常CPU核心数*2个线程，负责Read/Write） │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ EventLoop│ │ EventLoop│ │ EventLoop│ │ EventLoop│ │
│ │Thread 1 │ │Thread 2 │ │Thread 3 │ │Thread 4 │ │
│ │Selector │ │Selector │ │Selector │ │Selector │ │
│ │Channel A│ │Channel B│ │Channel C│ │Channel D│ │
│ │Channel E│ │Channel F│ │Channel G│ │Channel H│ │
│ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │
│ │
│ 特点： │
│ - 一个EventLoop绑定一个线程 │
│ - 一个Channel的生命周期内只由一个EventLoop处理 │
│ - 一个EventLoop可以处理多个Channel │
│ - 无锁串行化设计，避免多线程竞争 │
└─────────────────────────────────────────────────────────────┘
/**
* EventLoop核心代码
*/
public class NioEventLoop extends SingleThreadEventLoop {
private Selector selector;
private Selector unwrappedSelector;
@Override
protected void run() {
for (;;) {
try {
// 1. 选择就绪的Channel
int selectCnt = selectStrategy.calculateStrategy(selectNowSupplier, hasTasks());
switch (selectCnt) {
case SelectStrategy.CONTINUE:
continue;
case SelectStrategy.BUSY_WAIT:
case SelectStrategy.SELECT:
select(wakenUp.getAndSet(false));
if (wakenUp.get()) {
selector.wakeup();
}
default:
}
// 2. 处理就绪事件
processSelectedKeys();
// 3. 执行定时任务和普通任务
runAllTasks();
} catch (Throwable t) {
handleLoopException(t);
}
}
}
private void processSelectedKeys() {
if (selectedKeys != null) {
processSelectedKeysOptimized(selectedKeys.flip());
} else {
processSelectedKeysPlain(selector.selectedKeys());
}
}
private void processSelectedKeysOptimized(SelectionKey[] selectedKeys) {
for (int i = 0;; i++) {
final SelectionKey k = selectedKeys[i];
if (k == null) {
break;
}
selectedKeys[i] = null;
final Object a = k.attachment();
if (a instanceof AbstractNioChannel) {
processSelectedKey(k, (AbstractNioChannel) a);
} else {
@SuppressWarnings("unchecked")
NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
processSelectedKey(k, task);
}
}
}
}ChannelPipeline结构：
ChannelPipeline:
HeadContext -> Decoder -> BusinessHandler -> Encoder -> TailContext
↑ ↓
Inbound（入站） Outbound（出站）
channelRead() write()
channelActive() bind()
exceptionCaught() connect()
入站事件传播：
Socket -> Head -> Decoder.decode() -> BusinessHandler.channelRead() -> Tail
出站事件传播：
BusinessHandler.write() -> Encoder.encode() -> Head -> Socket
/**
* ChannelHandler示例
*/
public class MyServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
ByteBuf buf = (ByteBuf) msg;
try {
System.out.println("收到消息：" + buf.toString(CharsetUtil.UTF_8));
// 写回客户端
ctx.writeAndFlush(Unpooled.copiedBuffer("Hello Client", CharsetUtil.UTF_8));
} finally {
ReferenceCountUtil.release(msg); // 释放ByteBuf
}
}
@Override
public void channelActive(ChannelHandlerContext ctx) {
System.out.println("客户端连接：" + ctx.channel().remoteAddress());
}
@Override
public void channelInactive(ChannelHandlerContext ctx) {
System.out.println("客户端断开：" + ctx.channel().remoteAddress());
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
}
/**
* ChannelInitializer
*/
public class MyServerInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel ch) {
ChannelPipeline pipeline = ch.pipeline();
// 解码器（解决粘包）
pipeline.addLast(new LengthFieldBasedFrameDecoder(65535, 0, 4, 0, 4));
// 字符串解码
pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
// 业务处理器
pipeline.addLast(new MyServerHandler());
// 字符串编码
pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
}
}ByteBuf vs Java ByteBuffer：
Java ByteBuffer：
position/limit/capacity
flip()切换读写模式
容易出错
Netty ByteBuf：
readerIndex / writerIndex / capacity
读写分离，不需要flip
自动扩容
引用计数
ByteBuf结构：
┌─────────────────────────────────────────────────────────────┐
│ 已读区域 │ 可读区域 │ 可写区域 │ 扩容区域（需要时） │
│ │ │ │ │
│ 0 │ readerIndex│ writerIndex│ capacity │
└─────────────────────────────────────────────────────────────┘
ByteBuf类型：
┌─────────────────────┬─────────────────────────────────────┐
│ 类型 │ 特点 │
├─────────────────────┼─────────────────────────────────────┤
│ UnpooledHeapByteBuf │ 堆内存，未池化，受GC影响 │
│ UnpooledDirectByteBuf│ 直接内存，未池化，零拷贝 │
│ PooledHeapByteBuf │ 堆内存，池化，减少GC │
│ PooledDirectByteBuf │ 直接内存，池化（默认），性能最好 │
└─────────────────────┴─────────────────────────────────────┘
/**
* ByteBuf使用示例
*/
public class ByteBufExample {
public void demonstrate() {
// 1. 创建ByteBuf
ByteBuf buf = Unpooled.buffer(1024); // 未池化
// ByteBuf buf = PooledByteBufAllocator.DEFAULT.buffer(1024); // 池化
// 2. 写数据（writerIndex移动）
buf.writeBytes("Hello".getBytes());
buf.writeInt(123);
// 3. 读数据（readerIndex移动）
byte[] bytes = new byte[5];
buf.readBytes(bytes);
int num = buf.readInt();
// 4. 自动扩容
ByteBuf smallBuf = Unpooled.buffer(10);
smallBuf.writeBytes("Hello World!!!!".getBytes()); // 自动扩容
// 5. 引用计数
ByteBuf refBuf = Unpooled.buffer(1024);
refBuf.retain(); // 引用计数+1
refBuf.release(); // 引用计数-1
// 引用计数为0时，内存回收（池化ByteBuf返回内存池）
// 6. CompositeByteBuf（零拷贝组合）
CompositeByteBuf compositeBuf = Unpooled.compositeBuffer();
compositeBuf.addComponents(true, buf1, buf2, buf3);
}
}内存池架构：
PooledByteBufAllocator
├── Tiny SubPage Pools（< 512B）
│ └── 32个PoolSubpage（16B, 32B, ..., 496B）
├── Small SubPage Pools（512B - 8KB）
│ └── 4个PoolSubpage（512B, 1024B, 2048B, 4096B）
├── Normal Pools（8KB - 16MB）
│ └── 3个PoolChunkList（按使用率分类）
│ ├── qInit（0-25%）
│ ├── q000（1-50%）
│ ├── q025（25-75%）
│ ├── q050（50-100%）
│ ├── q075（75-100%）
│ └── q100（100%）
└── Huge Pools（> 16MB）
└── 非池化分配
PoolChunk：
- 16MB的内存块
- 使用完全二叉树管理内存分配
- 每个节点代表一个内存块，记录是否已分配
引用计数：
- ReferenceCounted接口
- retain()增加引用
- release()减少引用，为0时回收
- 防止内存泄漏
TCP粘包/拆包原因：
1. TCP是流式协议，不保留消息边界
2. Nagle算法合并小数据包
3. MTU限制导致分包
解决方案：
1. 固定长度：
| 固定长度消息 | 固定长度消息 | 固定长度消息 |
- FixedLengthFrameDecoder
- 缺点：浪费带宽（消息不足填充空格）
2. 分隔符：
| 消息1\n | 消息2\n | 消息3\n |
- DelimiterBasedFrameDecoder
- 缺点：消息内容不能包含分隔符
3. 长度字段（推荐）：
| Length (4字节) | Content |
- LengthFieldBasedFrameDecoder
- 最灵活，推荐
4. 自定义协议：
| Magic (2B) | Version (1B) | Length (4B) | Content |
/**
* 自定义协议编解码器
*/
public class MyProtocol {
// 协议格式：| Magic (2字节) | Version (1字节) | Length (4字节) | Content |
public static final int MAGIC = 0x1234;
public static final int VERSION = 1;
public static final int HEADER_LENGTH = 7;
}
/**
* 编码器
*/
public class MyProtocolEncoder extends MessageToByteEncoder<MyMessage> {
@Override
protected void encode(ChannelHandlerContext ctx, MyMessage msg, ByteBuf out) {
out.writeShort(MyProtocol.MAGIC); // Magic
out.writeByte(MyProtocol.VERSION); // Version
byte[] content = msg.getContent().getBytes(StandardCharsets.UTF_8);
out.writeInt(content.length); // Length
out.writeBytes(content); // Content
}
}
/**
* 解码器
*/
public class MyProtocolDecoder extends ByteToMessageDecoder {
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
// 检查是否够头部
if (in.readableBytes() < MyProtocol.HEADER_LENGTH) {
return;
}
in.markReaderIndex();
short magic = in.readShort();
if (magic != MyProtocol.MAGIC) {
in.resetReaderIndex();
throw new IllegalArgumentException("Invalid magic: " + magic);
}
byte version = in.readByte();
int length = in.readInt();
// 检查是否够一个完整消息
if (in.readableBytes() < length) {
in.resetReaderIndex();
return;
}
byte[] content = new byte[length];
in.readBytes(content);
out.add(new MyMessage(new String(content, StandardCharsets.UTF_8)));
}
}Netty零拷贝技术：
1. CompositeByteBuf：
- 组合多个ByteBuf，无需拷贝
- 示例：HTTP协议头 + Body组合
2. FileRegion：
- 文件传输零拷贝
- FileChannel.transferTo()直接传输到Socket
3. Unpooled.wrappedBuffer()：
- 包装现有byte[]，无需拷贝
4. slice() / duplicate()：
- 共享底层内存，创建视图
- 无数据拷贝
传统数据传输 vs 零拷贝：
传统：
文件 -> 内核缓冲区 -> 用户缓冲区（ByteBuf）-> Socket缓冲区 -> 网卡
（4次拷贝，4次上下文切换）
零拷贝（FileRegion）：
文件 -> 内核缓冲区 -> 网卡
（2次拷贝，1次上下文切换）
/**
* Echo服务器
*/
public class EchoServer {
private int port;
public void start() throws InterruptedException {
// BossGroup：接收连接（1个线程）
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
// WorkerGroup：处理IO（CPU核心数*2个线程）
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) {
ChannelPipeline p = ch.pipeline();
p.addLast(new StringDecoder());
p.addLast(new StringEncoder());
p.addLast(new EchoServerHandler());
}
});
ChannelFuture f = b.bind(port).sync();
System.out.println("Echo Server started on port " + port);
f.channel().closeFuture().sync();
} finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
}
public static void main(String[] args) throws InterruptedException {
new EchoServer(8080).start();
}
}
/**
* Echo处理器
*/
public class EchoServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
ctx.write(msg); // 写回客户端
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) {
ctx.flush();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
}/**
* HTTP服务器
*/
public class HttpServer {
public void start(int port) throws InterruptedException {
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) {
ChannelPipeline p = ch.pipeline();
p.addLast(new HttpServerCodec()); // HTTP编解码
p.addLast(new HttpObjectAggregator(65536)); // 聚合HTTP消息
p.addLast(new HttpServerHandler()); // 业务处理器
}
});
ChannelFuture f = b.bind(port).sync();
f.channel().closeFuture().sync();
} finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
}
}
/**
* HTTP处理器
*/
public class HttpServerHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) {
String uri = request.uri();
HttpMethod method = request.method();
String responseContent = "Hello, " + uri;
FullHttpResponse response = new DefaultFullHttpResponse(
HttpVersion.HTTP_1_1,
HttpResponseStatus.OK,
Unpooled.copiedBuffer(responseContent, CharsetUtil.UTF_8)
);
response.headers()
.set(HttpHeaderNames.CONTENT_TYPE, "text/plain; charset=UTF-8")
.set(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());
ctx.writeAndFlush(response);
}
}/**
* WebSocket服务器
*/
public class WebSocketServer {
public void start(int port) throws InterruptedException {
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) {
ChannelPipeline p = ch.pipeline();
p.addLast(new HttpServerCodec());
p.addLast(new HttpObjectAggregator(65536));
p.addLast(new WebSocketServerProtocolHandler("/ws")); // WebSocket协议
p.addLast(new WebSocketFrameHandler());
}
});
ChannelFuture f = b.bind(port).sync();
f.channel().closeFuture().sync();
} finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
}
}
/**
* WebSocket帧处理器
*/
public class WebSocketFrameHandler extends SimpleChannelInboundHandler<WebSocketFrame> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame frame) {
if (frame instanceof TextWebSocketFrame) {
String text = ((TextWebSocketFrame) frame).text();
System.out.println("Received: " + text);
// 广播给所有客户端
ChannelGroup channels = WebSocketChannelPool.getInstance().getChannels();
channels.writeAndFlush(new TextWebSocketFrame("Echo: " + text));
}
}
}/**
* RPC请求
*/
public class RpcRequest {
private String requestId;
private String interfaceName;
private String methodName;
private Class<?>[] parameterTypes;
private Object[] parameters;
// getters and setters
}
/**
* RPC响应
*/
public class RpcResponse {
private String requestId;
private Object result;
private Throwable error;
// getters and setters
}
/**
* RPC编解码器
*/
public class RpcCodec {
// 编码：RpcRequest/RpcResponse -> ByteBuf
public static ByteBuf encode(Object msg) {
ByteBuf buf = Unpooled.buffer();
if (msg instanceof RpcRequest) {
RpcRequest request = (RpcRequest) msg;
buf.writeByte(0x01); // 请求类型
buf.writeLong(request.getRequestId().hashCode());
// 序列化请求体...
} else if (msg instanceof RpcResponse) {
RpcResponse response = (RpcResponse) msg;
buf.writeByte(0x02); // 响应类型
buf.writeLong(response.getRequestId().hashCode());
// 序列化响应体...
}
return buf;
}
// 解码：ByteBuf -> RpcRequest/RpcResponse
public static Object decode(ByteBuf buf) {
byte type = buf.readByte();
long requestId = buf.readLong();
if (type == 0x01) {
// 解码请求...
return new RpcRequest();
} else {
// 解码响应...
return new RpcResponse();
}
}
}
/**
* RPC服务器处理器
*/
public class RpcServerHandler extends SimpleChannelInboundHandler<RpcRequest> {
private Map<String, Object> serviceMap;
public RpcServerHandler(Map<String, Object> serviceMap) {
this.serviceMap = serviceMap;
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, RpcRequest request) {
RpcResponse response = new RpcResponse();
response.setRequestId(request.getRequestId());
try {
Object service = serviceMap.get(request.getInterfaceName());
Method method = service.getClass().getMethod(
request.getMethodName(), request.getParameterTypes());
Object result = method.invoke(service, request.getParameters());
response.setResult(result);
} catch (Exception e) {
response.setError(e);
}
ctx.writeAndFlush(response);
}
}/**
* 百万连接推送服务器
*/
public class PushServer {
public void start(int port) throws InterruptedException {
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 1024)
.childOption(ChannelOption.TCP_NODELAY, true)
.childOption(ChannelOption.SO_KEEPALIVE, true)
.childOption(ChannelOption.SO_RCVBUF, 65536)
.childOption(ChannelOption.SO_SNDBUF, 65536)
.childOption(ChannelOption.WRITE_BUFFER_WATER_MARK,
new WriteBufferWaterMark(32 * 1024, 64 * 1024))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) {
ChannelPipeline p = ch.pipeline();
p.addLast(new IdleStateHandler(60, 30, 0)); // 心跳检测
p.addLast(new PushServerHandler());
}
});
ChannelFuture f = b.bind(port).sync();
System.out.println("Push Server started on port " + port);
f.channel().closeFuture().sync();
} finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
}
}
/**
* 连接管理器
*/
public class ConnectionManager {
private static final ConnectionManager INSTANCE = new ConnectionManager();
private ConcurrentHashMap<String, Channel> connections = new ConcurrentHashMap<>();
public static ConnectionManager getInstance() {
return INSTANCE;
}
public void addConnection(String userId, Channel channel) {
connections.put(userId, channel);
}
public void removeConnection(String userId) {
connections.remove(userId);
}
public void pushToUser(String userId, Object message) {
Channel channel = connections.get(userId);
if (channel != null && channel.isActive()) {
channel.writeAndFlush(message);
}
}
public void broadcast(Object message) {
for (Channel channel : connections.values()) {
if (channel.isActive()) {
channel.writeAndFlush(message);
}
}
}
public int getConnectionCount() {
return connections.size();
}
}┌─────────────────────┬─────────────────┬─────────────────┬─────────────────┐
│ 特性 │ Netty │ Mina │ Vert.x │
├─────────────────────┼─────────────────┼─────────────────┼─────────────────┤
│ 开发方 │ JBoss/Netty │ Apache │ Eclipse │
├─────────────────────┼─────────────────┼─────────────────┼─────────────────┤
│ 开发语言 │ Java │ Java │ Java │
├─────────────────────┼─────────────────┼─────────────────┼─────────────────┤
│ 编程模型 │ 事件驱动/回调 │ 事件驱动/回调 │ 响应式/事件驱动│
├─────────────────────┼─────────────────┼─────────────────┼─────────────────┤
│ 性能 │ 极高 │ 高 │ 高 │
├─────────────────────┼─────────────────┼─────────────────┼─────────────────┤
│ 内存管理 │ ByteBuf（优秀）│ IoBuffer（一般）│ 依赖底层 │
├─────────────────────┼─────────────────┼─────────────────┼─────────────────┤
│ 协议支持 │ 非常丰富 │ 丰富 │ 较丰富 │
├─────────────────────┼─────────────────┼─────────────────┼─────────────────┤
│ 线程模型 │ EventLoop │ IoProcessor │ Event Loop │
├─────────────────────┼─────────────────┼─────────────────┼─────────────────┤
│ 生态 │ 最丰富 │ 一般 │ 较丰富 │
├─────────────────────┼─────────────────┼─────────────────┼─────────────────┤
│ 使用场景 │ RPC/IM/游戏 │ 传统网络应用 │ 微服务/Web │
├─────────────────────┼─────────────────┼─────────────────┼─────────────────┤
│ 活跃度 │ 非常活跃 │ 较低 │ 活跃 │
└─────────────────────┴─────────────────┴─────────────────┴─────────────────┘
1. 连接数优化：
- BossGroup：1个线程（Accept）
- WorkerGroup：CPU核心数*2
- 业务线程池：根据业务类型调整
2. TCP参数：
- SO_BACKLOG：1024+（半连接队列）
- TCP_NODELAY：true（禁用Nagle算法）
- SO_KEEPALIVE：true（心跳检测）
- SO_REUSEADDR：true（端口复用）
3. 缓冲区：
- SO_RCVBUF/SO_SNDBUF：根据网络环境调整
- WriteBufferWaterMark：防止内存溢出
4. 心跳机制：
- IdleStateHandler：检测空闲连接
- 及时关闭死连接，释放资源
1. 使用池化ByteBuf：
- PooledByteBufAllocator.DEFAULT
- 减少GC压力
2. 及时释放ByteBuf：
- ReferenceCountUtil.release(msg)
- 或使用SimpleChannelInboundHandler（自动释放）
3. 避免大内存分配：
- 控制消息大小
- 使用CompositeByteBuf组合小缓冲区
4. 监控内存泄漏：
- -Dio.netty.leakDetectionLevel=advanced
- 定期检查日志
5. JVM调优：
- -Xmx4g -Xms4g
- -XX:+UseG1GC
- -XX:MaxDirectMemorySize=2g
陷阱： 在EventLoop线程中执行数据库查询、RPC调用等耗时操作，阻塞IO线程，导致其他连接处理延迟。
最佳实践：
// 使用业务线程池处理耗时操作
EventExecutorGroup businessGroup = new DefaultEventExecutorGroup(16);
pipeline.addLast(businessGroup, new BusinessHandler());
// 或使用异步处理
@Override
public void channelRead0(ChannelHandlerContext ctx, Object msg) {
CompletableFuture.supplyAsync(() -> {
// 耗时操作
return process(msg);
}, businessExecutor).thenAccept(result -> {
ctx.writeAndFlush(result);
});
}陷阱： ByteBuf使用后未释放，引用计数不归零，导致内存泄漏。
最佳实践：
// 方式1：手动释放
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
ByteBuf buf = (ByteBuf) msg;
try {
// 处理消息
} finally {
ReferenceCountUtil.release(buf);
}
}
// 方式2：使用SimpleChannelInboundHandler（自动释放）
public class MyHandler extends SimpleChannelInboundHandler<ByteBuf> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {
// 处理消息，无需手动释放
}
}
// 方式3：启用内存泄漏检测
// -Dio.netty.leakDetectionLevel=advanced陷阱： TCP是流式协议，不保证消息边界，直接读取可能导致解析错误。
最佳实践：
// 使用LengthFieldBasedFrameDecoder（推荐）
pipeline.addLast(new LengthFieldBasedFrameDecoder(
65535, // 最大帧长度
0, // 长度字段偏移
4, // 长度字段长度
0, // 长度调整
4 // 跳过的字节数（头部长度）
));
// 或使用DelimiterBasedFrameDecoder（分隔符）
pipeline.addLast(new DelimiterBasedFrameDecoder(
8192, Unpooled.copiedBuffer("\n".getBytes())
));
// 或使用FixedLengthFrameDecoder（固定长度）
pipeline.addLast(new FixedLengthFrameDecoder(100));陷阱： Netty默认使用直接内存，大量连接或大消息时可能超出限制。
最佳实践：
# 设置直接内存上限
java -Dio.netty.maxDirectMemory=2147483648 # 2GB
# 或使用JVM参数
java -XX:MaxDirectMemorySize=2g// 监控直接内存使用
long usedDirectMemory = PlatformDependent.usedDirectMemory();
long maxDirectMemory = PlatformDependent.maxDirectMemory();
System.out.println("Direct Memory: " + usedDirectMemory + " / " + maxDirectMemory);答：
| IO模型 | 说明 | 优点 | 缺点 |
|---|---|---|---|
| BIO | 一个连接一个线程 | 编程简单 | 连接数受限，资源消耗大 |
| NIO | 多路复用，单线程处理多个连接 | 高并发，资源利用率高 | 编程复杂，需处理Selector |
| AIO | 异步IO，操作系统回调 | 性能最好 | Linux支持不完善，使用少 |
答： Netty使用主从Reactor多线程模型：
- Boss EventLoopGroup：1个线程，负责接收客户端连接
- Worker EventLoopGroup：N个线程（默认CPU核心数*2），负责处理读写事件
- EventLoop：一个EventLoop绑定一个线程，处理多个Channel的IO事件
- Channel：一个Channel的生命周期内只由一个EventLoop处理
关键设计：无锁串行化，一个Channel的所有IO操作由同一个线程执行，避免多线程竞争。
答：
- 读写分离：ByteBuf有readerIndex和writerIndex，不需要flip切换
- 自动扩容：写入超出容量时自动扩容
- 引用计数：通过retain()和release()管理内存生命周期
- 池化：PooledByteBufAllocator复用ByteBuf，减少GC压力
- 零拷贝：支持CompositeByteBuf组合多个缓冲区
- 更丰富的API：支持各种数据类型的读写
答：
- 原因：TCP是面向流的协议，不保留消息边界；Nagle算法合并小数据包；MTU限制导致分包
- 解决方案：
- 固定长度：FixedLengthFrameDecoder
- 分隔符：DelimiterBasedFrameDecoder（如换行符）
- 长度字段：LengthFieldBasedFrameDecoder（最灵活，推荐）
答： Netty使用PooledByteBufAllocator管理内存池：
- Tiny Pool：< 512B，16B为粒度
- Small Pool：512B ~ 8KB
- Normal Pool：8KB ~ 16MB，以Chunk（16MB）为单位分配
- 采用jemalloc算法的思想，减少内存碎片和GC频率
- 默认使用直接内存，避免JVM堆内外拷贝
答： Netty的零拷贝包括：
- CompositeByteBuf：组合多个ByteBuf，无需数据拷贝
- FileRegion：文件传输直接通过FileChannel.transferTo()发送到Socket
- Unpooled.wrappedBuffer()：包装现有byte[]，不拷贝数据
- slice() / duplicate()：共享底层内存，创建视图
答： JDK NIO的Epoll Bug（JDK 6之前）表现为Selector空轮询，CPU 100%。
Netty的解决方案：
- 重建Selector：检测到空轮询次数超过阈值后，重新创建Selector
- 计数器：记录空轮询次数
- 阈值：默认512次
// Netty源码中的处理逻辑
if (selectCnt >= MIN_PREMATURE_SELECTOR_RETURNS) {
// 重建Selector
selector = selectRebuildSelector(selectCnt);
}答： ChannelPipeline是ChannelHandler的双向链表：
- 入站事件（Inbound）：从Head向Tail传播
- channelRegistered、channelActive、channelRead、channelInactive等
- 出站事件（Outbound）：从Tail向Head传播
- bind、connect、write、flush、read、disconnect、close等
执行顺序：
- 入站事件：Head -> Decoder -> BusinessHandler -> Tail
- 出站事件：BusinessHandler -> Encoder -> Head
注意：ChannelHandler可以动态添加、删除、替换。
答：
| 特性 | Future | Promise |
|---|---|---|
| 定义 | 异步操作的结果 | 可写的Future |
| 状态 | 只读 | 可写 |
| 设置结果 | 不可设置 | 可以设置成功或失败 |
| 使用场景 | 消费者获取结果 | 生产者设置结果 |
关系：Promise是Future的子接口，增加了设置结果的能力。
示例：
// Future：只能获取结果
ChannelFuture future = channel.writeAndFlush(msg);
future.addListener(f -> {
if (f.isSuccess()) {
System.out.println("发送成功");
}
});
// Promise：可以设置结果
Promise<String> promise = new DefaultPromise<>(eventExecutor);
promise.setSuccess("结果");
promise.setFailure(new Exception("错误"));答： Netty提供四种解码器处理半包和粘包：
-
FixedLengthFrameDecoder：固定长度解码
- 适用：消息长度固定
- 缺点：浪费带宽
-
LineBasedFrameDecoder：换行符解码
- 适用：文本协议（如SMTP、FTP）
- 缺点：消息不能包含换行符
-
DelimiterBasedFrameDecoder：自定义分隔符解码
- 适用：消息有明确分隔符
-
LengthFieldBasedFrameDecoder：长度字段解码（推荐）
- 适用：二进制协议
- 最灵活，支持各种协议格式
答： Netty提供内存泄漏检测机制：
检测级别：
- DISABLED：禁用
- SIMPLE：简单检测（默认，1%采样）
- ADVANCED：高级检测（记录调用栈）
- PARANOID：偏执模式（100%采样，性能影响大）
启用方式：
-Dio.netty.leakDetectionLevel=advanced泄漏报告：
LEAK: ByteBuf.release() was not called before it's garbage-collected.
Recent access records:
#1:
io.netty.buffer.AdvancedLeakAwareByteBuf.writeBytes(...)
at com.example.MyHandler.channelRead(MyHandler.java:23)
答： Netty实现百万长连接的关键：
-
内存优化：
- 使用池化ByteBuf
- 控制单个连接内存占用（< 1MB）
- 及时释放资源
-
连接优化：
- BossGroup 1个线程
- WorkerGroup CPU核心数*2
- 业务线程池分离
-
心跳机制：
- IdleStateHandler检测空闲连接
- 及时清理死连接
-
操作系统调优：
- 调整文件描述符限制（ulimit -n 1000000）
- 调整TCP参数（tcp_tw_reuse、tcp_fin_timeout）
- 调整内核参数（net.ipv4.ip_local_port_range）
-
JVM调优：
- 堆内存合理分配
- G1或ZGC垃圾回收器
- 直接内存限制
答：
| 特性 | Bootstrap | ServerBootstrap |
|---|---|---|
| 用途 | 客户端 | 服务端 |
| EventLoopGroup | 1个 | 2个（Boss+Worker） |
| Channel类型 | SocketChannel | ServerSocketChannel |
| 方法 | connect() | bind() |
代码对比：
// 客户端Bootstrap
Bootstrap b = new Bootstrap();
b.group(group)
.channel(NioSocketChannel.class)
.handler(new ClientInitializer());
ChannelFuture f = b.connect(host, port).sync();
// 服务端ServerBootstrap
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ServerInitializer());
ChannelFuture f = b.bind(port).sync();答： Netty通过SslContext和SslHandler支持SSL/TLS：
// 服务端SSL配置
SslContext sslCtx = SslContextBuilder
.forServer(certFile, keyFile)
.build();
pipeline.addLast(sslCtx.newHandler(channel.alloc()));
// 客户端SSL配置
SslContext sslCtx = SslContextBuilder
.forClient()
.trustManager(certFile)
.build();
pipeline.addLast(sslCtx.newHandler(channel.alloc(), host, port));答：
1. 线程模型：
- BossGroup：1个线程
- WorkerGroup：CPU核心数*2
- 业务线程池：根据业务类型调整
2. TCP参数：
- SO_BACKLOG：1024+
- TCP_NODELAY：true
- SO_KEEPALIVE：true
- SO_REUSEADDR：true
3. 内存管理：
- 使用PooledByteBufAllocator
- 设置合理的WriteBufferWaterMark
- 启用内存泄漏检测（生产环境用SIMPLE）
4. 心跳机制：
- IdleStateHandler：检测空闲连接
- 及时关闭死连接
5. 监控：
- 连接数监控
- 流量监控
- 内存使用监控
- GC监控
6. 高可用：
- 多实例部署
- 负载均衡（LVS、Nginx、HAProxy）
- 优雅关闭（shutdownGracefully）
此文原创，转载请注明出处。