Netty 基础教程

从 NIO 到 Reactor 模型与编解码器，12 个单元 · 36 道经典实训题，左讲解右跟练。

12 个单元 47 道实训题基础 + 拔高

📋 跟练列表

单元1 · Netty快速入门

单元1 · Netty 快速入门

NIO 基础、Netty 特性、第一个服务端

服务端：ServerBootstrap + NioEventLoopGroup + NioServerSocketChannel。
处理：ChannelInitializer 装配 pipeline，SimpleChannelInboundHandler 处理消息。

实训1.1 引入 Netty 依赖

📝 题目描述

在 pom.xml 中引入 netty-all 依赖。

💡 解题思路

netty-all 聚合所有 Netty 模块，开发期使用方便。

<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.100.Final</version>
</dependency>

实训1.2 第一个 Netty 服务端

📝 题目描述

编写 EchoServer：绑定 8080 端口，将客户端消息原样返回。

💡 解题思路

ServerBootstrap 配置服务端；NioEventLoopGroup 管理线程；ChannelHandler 处理消息。

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;

public class EchoServer {

    public static void main(String[] args) throws Exception {
        EventLoopGroup boss = new NioEventLoopGroup(1);
        EventLoopGroup worker = new NioEventLoopGroup();

        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(boss, worker)
             .channel(NioServerSocketChannel.class)
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 protected void initChannel(SocketChannel ch) {
                     ch.pipeline().addLast(new EchoHandler());
                 }
             });
            ChannelFuture f = b.bind(8080).sync();
            System.out.println("EchoServer 启动，端口 8080");
            f.channel().closeFuture().sync();
        } finally {
            boss.shutdownGracefully();
            worker.shutdownGracefully();
        }
    }
}

实训1.3 ChannelHandler 处理消息

📝 题目描述

编写 EchoHandler 继承 SimpleChannelInboundHandler 处理字符串消息。

💡 解题思路

channelRead0 接收解码后的消息；writeAndFlush 写回客户端。

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

public class EchoHandler extends SimpleChannelInboundHandler<String> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) {
        System.out.println("收到：" + msg);
        ctx.writeAndFlush("回复：" + msg);
    }

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

单元2 · EventLoop 与线程模型

EventLoop、EventLoopGroup、线程分配

线程模型：boss 接受连接、worker 处理 IO；一个 Channel 绑定一个 EventLoop。
任务调度：execute 提交任务、schedule 延时任务。

实训2.1 Boss/Worker 线程模型

📝 题目描述

说明 NioEventLoopGroup 中 boss 与 worker 的分工。

💡 解题思路

boss 负责接受连接，worker 负责读写；默认 worker 数为 CPU 核数 * 2。

EventLoopGroup boss = new NioEventLoopGroup(1);
EventLoopGroup worker = new NioEventLoopGroup();

// boss：接受客户端连接
// worker：处理已建立连接的 IO 事件
ServerBootstrap b = new ServerBootstrap();
b.group(boss, worker);

实训2.2 EventLoop 绑定

📝 题目描述

说明一个 Channel 与其 EventLoop 的绑定关系。

💡 解题思路

一个 Channel 整个生命周期绑定到同一个 EventLoop（线程），避免并发竞争。

// 同一 Channel 的所有事件由同一个 EventLoop 串行处理
// 因此 ChannelHandler 中数据访问无需额外加锁
Channel ch = ctx.channel();
EventLoop loop = ch.eventLoop();
System.out.println("Channel 绑定的 EventLoop 线程：" + loop);

实训2.3 提交任务到 EventLoop

📝 题目描述

演示向 EventLoop 提交定时任务与普通任务。

💡 解题思路

execute/submit 提交普通任务；schedule 提交延时任务。

import io.netty.channel.EventLoop;
import java.util.concurrent.TimeUnit;

public void scheduleTask(EventLoop loop) {
    // 延时任务
    loop.schedule(() -> System.out.println("1 秒后执行"), 1, TimeUnit.SECONDS);

    // 周期任务
    loop.scheduleAtFixedRate(() -> System.out.println("每 2 秒执行"),
            0, 2, TimeUnit.SECONDS);
}

单元3 · Channel 与 ChannelPipeline

Channel 生命周期、pipeline 责任链

Channel：writeAndFlush 写、close 关、isActive 判断状态。
Pipeline：责任链模式，入站/出站 handler 按序执行。
传递：fireChannelRead 向后传递消息。

实训3.1 Channel 核心方法

📝 题目描述

演示 writeAndFlush、close、isActive 等 Channel 方法。

💡 解题思路

writeAndFlush 写数据；close 关闭连接；isActive 判断连接状态。

import io.netty.channel.Channel;

public void channelOps(Channel ch) {
    boolean active = ch.isActive();
    System.out.println("连接状态：" + active);

    ch.writeAndFlush("hello");
    // ch.close();  // 关闭连接
}

实训3.2 Pipeline 责任链

📝 题目描述

为 pipeline 添加多个 handler，说明执行顺序。

💡 解题思路

pipeline 按添加顺序执行；ChannelInboundHandler 处理入站、Outbound 处理出站。

import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;

public class ServerInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel ch) {
        ch.pipeline()
          .addLast("decoder", new MyDecoder())      // 入站：解码
          .addLast("handler", new EchoHandler())    // 入站：业务
          .addLast("encoder", new MyEncoder());     // 出站：编码
    }
}

实训3.3 Handler 上下文传递

📝 题目描述

演示 handler 中调用 ctx.fireChannelRead 向后传递消息。

💡 解题思路

fireChannelRead 将消息传递给下一个入站 handler；不调用则链终止。

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

public class FirstHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        System.out.println("FirstHandler 处理");
        // 传递给下一个 handler
        ctx.fireChannelRead(msg);
    }
}

单元4 · ByteBuf 详解

堆内存与直接内存、读写指针、池化

ByteBuf：读写指针分离，readableBytes 可读字节数。
池化：PooledByteBufAllocator 复用内存，引用计数管理生命周期。
Composite：组合多个缓冲区避免复制。

实训4.1 ByteBuf 读写

📝 题目描述

演示 ByteBuf 的 writeByte/readByte 与读写指针变化。

💡 解题思路

writerIndex 记录写位置，readerIndex 记录读位置；readableBytes 为可读字节数。

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;

public class ByteBufDemo {

    public static void main(String[] args) {
        ByteBuf buf = Unpooled.buffer(16);
        buf.writeInt(100);          // 写 4 字节
        buf.writeBytes("Hi".getBytes());

        System.out.println("可读字节：" + buf.readableBytes());
        System.out.println("读 int：" + buf.readInt());
        System.out.println("读字符串：" + buf.readCharSequence(2, java.nio.charset.StandardCharsets.UTF_8));
        buf.release();
    }
}

实训4.2 池化与引用计数

📝 题目描述

说明池化 ByteBuf 与引用计数的关系。

💡 解题思路

池化复用内存减少 GC；引用计数管理生命周期，使用后 release。

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;

public void allocDemo() {
    // 默认 allocator 为池化直接内存
    ByteBuf buf = ByteBufAllocator.DEFAULT.buffer(128);
    try {
        buf.writeInt(1);
        System.out.println("refCnt=" + buf.refCnt());
    } finally {
        buf.release();
        System.out.println("release 后 refCnt=" + buf.refCnt());
    }
}

实训4.3 CompositeByteBuf

📝 题目描述

演示 CompositeByteBuf 组合多个缓冲区为逻辑视图。

💡 解题思路

CompositeByteBuf 将多个 ByteBuf 组合，避免复制开销。

import io.netty.buffer.CompositeByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.util.CharsetUtil;

public class CompositeDemo {

    public static void main(String[] args) {
        CompositeByteBuf composite = Unpooled.compositeBuffer();
        composite.addComponent(true, Unpooled.copiedBuffer("Hel", CharsetUtil.UTF_8));
        composite.addComponent(true, Unpooled.copiedBuffer("lo", CharsetUtil.UTF_8));

        System.out.println(composite.toString(CharsetUtil.UTF_8));
        composite.release();
    }
}

单元5 · 编解码器

ByteToMessageDecoder、MessageToByteEncoder、Codec

解码器：ByteToMessageDecoder 累积字节产出消息。
编码器：MessageToByteEncoder 写出前编码。
Codec：CombinedChannelDuplexHandler 组合双向编解码。

实训5.1 自定义解码器

📝 题目描述

编写 IntegerDecoder 将字节流解码为 int 消息。

💡 解题思路

ByteToMessageDecoder 积累字节，readableBytes 足够时读出对象加入 out。

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import java.util.List;

public class IntegerDecoder extends ByteToMessageDecoder {

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        if (in.readableBytes() >= 4) {
            out.add(in.readInt());
        }
    }
}

实训5.2 自定义编码器

📝 题目描述

编写 StringEncoder 将字符串编码为字节写出。

💡 解题思路

MessageToByteEncoder 在写出前将消息编码到 ByteBuf。

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import java.nio.charset.StandardCharsets;

public class StringEncoder extends MessageToByteEncoder<String> {

    @Override
    protected void encode(ChannelHandlerContext ctx, String msg, ByteBuf out) {
        out.writeBytes(msg.getBytes(StandardCharsets.UTF_8));
    }
}

实训5.3 组合 Codec

📝 题目描述

编写 CombinedChannelDuplexHandler 组合解码器与编码器。

💡 解题思路

CombinedChannelDuplexHandler 将 decoder 与 encoder 合并为一个 handler。

import io.netty.channel.CombinedChannelDuplexHandler;

public class StringCodec extends CombinedChannelDuplexHandler<IntegerDecoder, StringEncoder> {

    public StringCodec() {
        super(new IntegerDecoder(), new StringEncoder());
    }
}

单元6 · 粘包与拆包

问题现象、LineBasedFrameDecoder、自定义协议

粘包拆包：TCP 字节流无边界导致。
分隔符：LineBasedFrameDecoder 按换行拆包。
长度头：LengthFieldBasedFrameDecoder 按长度字段拆包。

实训6.1 粘包拆包现象

📝 题目描述

说明 TCP 粘包与拆包产生的原因。

💡 解题思路

TCP 是字节流协议，多条消息可能粘在一起（粘包）或一条被拆开（拆包）。

// 粘包：发送方连续 send 多条，接收方一次读到多条
// 拆包：一条大消息被分多次读入
// 原因：TCP 无消息边界，以字节流传输
public class StickyNote {
    // 解决方案：定长、分隔符、长度字段
}

实训6.2 分隔符解码器

📝 题目描述

使用 LineBasedFrameDecoder 按换行符拆包。

💡 解题思路

LineBasedFrameDecoder(maxLength) 按 \n 或 \r\n 分割消息。

import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import java.nio.charset.StandardCharsets;

public class LineServerInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel ch) {
        ch.pipeline()
          .addLast(new LineBasedFrameDecoder(1024))
          .addLast(new StringDecoder(StandardCharsets.UTF_8))
          .addLast(new EchoHandler());
    }
}

实训6.3 长度字段协议

📝 题目描述

使用 LengthFieldBasedFrameDecoder 处理带长度头的协议。

💡 解题思路

lengthFieldOffset/lengthFieldLength 指定长度字段位置与大小。

import io.netty.handler.codec.LengthFieldBasedFrameDecoder;

// 协议：4 字节长度头 + 消息体
public class LengthDecoder extends LengthFieldBasedFrameDecoder {

    public LengthDecoder() {
        super(65535, 0, 4, 0, 4);
    }
}

单元7 · HTTP 服务器

HttpServerCodec、HTTP 请求处理

HTTP：HttpServerCodec + HttpObjectAggregator。
响应：FullHttpResponse 设置状态码与 Content-Type。

实训7.1 HTTP 服务端搭建

📝 题目描述

用 Netty 编写 HTTP 服务端，返回 Hello World。

💡 解题思路

HttpServerCodec 编解码 HTTP 报文；HttpObjectAggregator 聚合请求。

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.http.*;

public class HttpServer {

    public static void main(String[] args) throws Exception {
        EventLoopGroup group = new NioEventLoopGroup();
        ServerBootstrap b = new ServerBootstrap();
        b.group(group)
         .channel(NioServerSocketChannel.class)
         .childHandler(new ChannelInitializer<SocketChannel>() {
             @Override
             protected void initChannel(SocketChannel ch) {
                 ch.pipeline()
                   .addLast(new HttpServerCodec())
                   .addLast(new HttpObjectAggregator(65536))
                   .addLast(new HttpHandler());
             }
         });
        b.bind(8080).sync().channel().closeFuture().sync();
    }
}

实训7.2 HTTP 响应

📝 题目描述

编写 HttpHandler 返回 JSON 格式的 HTTP 响应。

💡 解题思路

FullHttpResponse 携带状态码、Content-Type 与内容字节。

import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.*;
import java.nio.charset.StandardCharsets;

public class HttpHandler extends SimpleChannelInboundHandler<FullHttpRequest> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest req) {
        String body = "{"message":"Hello Netty HTTP"}";
        FullHttpResponse resp = new DefaultFullHttpResponse(
                HttpVersion.HTTP_1_1,
                HttpResponseStatus.OK,
                Unpooled.copiedBuffer(body, StandardCharsets.UTF_8));
        resp.headers().set(HttpHeaderNames.CONTENT_TYPE, "application/json");
        resp.headers().setInt(HttpHeaderNames.CONTENT_LENGTH, resp.content().readableBytes());
        ctx.writeAndFlush(resp);
    }
}

实训7.3 路由与静态资源

📝 题目描述

简述基于 Netty 实现简单路由分发（/api、/static）的思路。

💡 解题思路

解析 URI 前缀分发到不同 handler，静态资源读取本地文件返回。

public class RouterHandler extends SimpleChannelInboundHandler<FullHttpRequest> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest req) {
        String uri = req.uri();
        if (uri.startsWith("/api/")) {
            handleApi(ctx, req);
        } else if (uri.startsWith("/static/")) {
            handleStatic(ctx, req);
        } else {
            ctx.writeAndFlush(new DefaultFullHttpResponse(
                    HttpVersion.HTTP_1_1, HttpResponseStatus.NOT_FOUND,
                    Unpooled.EMPTY_BUFFER));
        }
    }
}

单元8 · WebSocket

WebSocketServerProtocolHandler、消息推送

WebSocket：WebSocketServerProtocolHandler 协议升级。
消息：TextWebSocketFrame 封装文本，ChannelGroup 广播。

实训8.1 WebSocket 服务端

📝 题目描述

搭建 WebSocket 服务端并添加协议升级 handler。

💡 解题思路

WebSocketServerProtocolHandler(path) 处理握手与协议升级。

import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;

public class WsInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel ch) {
        ch.pipeline()
          .addLast(new HttpServerCodec())
          .addLast(new HttpObjectAggregator(65536))
          .addLast(new WebSocketServerProtocolHandler("/ws"))
          .addLast(new WsHandler());
    }
}

实训8.2 WebSocket 消息处理

📝 题目描述

处理 TextWebSocketFrame 消息并广播给所有客户端。

💡 解题思路

TextWebSocketFrame 封装文本消息；ChannelGroup 管理所有连接。

import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.concurrent.GlobalEventExecutor;

public class WsHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    private static final ChannelGroup clients =
            new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        clients.add(ctx.channel());
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) {
        clients.writeAndFlush(new TextWebSocketFrame("广播：" + msg.text()));
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        clients.remove(ctx.channel());
    }
}

实训8.3 心跳与关闭处理

📝 题目描述

在 WebSocket handler 中处理关闭帧与异常。

💡 解题思路

处理 CloseWebSocketFrame 关闭连接；exceptionCaught 清理异常连接。

import io.netty.channel.*;
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;

public class WsHandler extends SimpleChannelInboundHandler<Object> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) {
        if (msg instanceof TextWebSocketFrame) {
            // 业务消息
            ctx.writeAndFlush(new TextWebSocketFrame("收到"));
        } else if (msg instanceof CloseWebSocketFrame) {
            ctx.close();
        }
    }

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

单元9 · 心跳与空闲检测

IdleStateHandler、自定义心跳协议

空闲检测：IdleStateHandler 触发 IdleStateEvent。
心跳：Ping/Pong 机制，超时判定离线。

实训9.1 空闲检测配置

📝 题目描述

使用 IdleStateHandler 检测读空闲与写空闲。

💡 解题思路

IdleStateHandler(reader, writer, all) 触发 IdleStateEvent 交由用户处理。

import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import java.util.concurrent.TimeUnit;

public class HeartbeatInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel ch) {
        ch.pipeline()
          .addLast(new IdleStateHandler(5, 0, 0, TimeUnit.SECONDS))
          .addLast(new HeartbeatHandler());
    }
}

实训9.2 心跳超时处理

📝 题目描述

监听 IdleStateEvent，读超时则关闭连接。

💡 解题思路

userEventTriggered 捕获 IdleStateEvent，根据 state 类型处理。

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;

public class HeartbeatHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent e = (IdleStateEvent) evt;
            if (e.state() == IdleState.READER_IDLE) {
                System.out.println("读空闲，关闭连接");
                ctx.close();
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }
}

实训9.3 心跳应答机制

📝 题目描述

设计 Ping/Pong 心跳协议并说明客户端断线重连思路。

💡 解题思路

客户端定时发 Ping，服务端回 Pong；超时未收到则判定离线。

// 协议消息类型
public enum MsgType {
    PING, PONG, DATA
}

// 客户端：
// 1. 每 30 秒发送 PING
// 2. 收到 PONG 或任意数据视为连接正常
// 3. 连续 N 次无响应则主动重连

// 服务端：
// 1. 收到 PING 立即回 PONG
// 2. 读空闲超过阈值判定离线并移除
public class HeartbeatProtocol {
    // 以上为设计要点
}

单元10 · 断线重连与客户端

Bootstrap、重连策略

客户端：Bootstrap + NioSocketChannel + connect。
重连：channelInactive 触发重连，退避间隔。
回调：ChannelFutureListener 异步处理连接结果。

实训10.1 Netty 客户端

📝 题目描述

编写 Netty 客户端连接 8080 端口并发送消息。

💡 解题思路

Bootstrap 配置客户端；connect 发起连接，writeAndFlush 发送数据。

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

public class EchoClient {

    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) {
                     ch.pipeline().addLast(new ClientHandler());
                 }
             });
            Channel ch = b.connect("127.0.0.1", 8080).sync().channel();
            ch.writeAndFlush("Hello Netty");
            ch.closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }
}

实训10.2 自动重连

📝 题目描述

实现连接断开后自动重连的客户端。

💡 解题思路

channelInactive 中触发重连；重试间隔递增避免风暴。

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import java.util.concurrent.TimeUnit;

public class ReconnectHandler extends ChannelInboundHandlerAdapter {

    private final Bootstrap bootstrap;
    private final String host;
    private final int port;
    private int retries = 0;

    public ReconnectHandler(Bootstrap bootstrap, String host, int port) {
        this.bootstrap = bootstrap;
        this.host = host;
        this.port = port;
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        long delay = Math.min(1L << retries++, 30);
        System.out.println("连接断开，" + delay + " 秒后重连");
        ctx.channel().eventLoop().schedule(() -> {
            bootstrap.connect(host, port).addListener(f -> {
                if (f.isSuccess()) {
                    retries = 0;
                }
            });
        }, delay, TimeUnit.SECONDS);
    }
}

实训10.3 连接成功回调

📝 题目描述

演示 ChannelFutureListener 处理连接成功/失败。

💡 解题思路

connect 返回 ChannelFuture，添加 listener 异步回调处理结果。

ChannelFuture future = bootstrap.connect("127.0.0.1", 8080);
future.addListener((ChannelFutureListener) f -> {
    if (f.isSuccess()) {
        System.out.println("连接成功：" + f.channel().remoteAddress());
        f.channel().writeAndFlush("连接建立");
    } else {
        System.err.println("连接失败：" + f.cause());
    }
});

单元11 · 性能优化与高级特性

零拷贝、线程调优、内存池

零拷贝：FileRegion/sendfile、CompositeByteBuf、DirectBuffer。
调优：线程数、backlog、TCP_NODELAY。
内存池：PooledByteBufAllocator 减少 GC。

实训11.1 零拷贝

📝 题目描述

说明 Netty 中的零拷贝技术及典型应用。

💡 解题思路

FileRegion 文件传输、CompositeByteBuf 组合缓冲、DirectBuffer 减少拷贝。

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.DefaultFileRegion;

// 文件传输零拷贝：FileRegion 借助 sendfile 系统调用
public void sendFile(ChannelHandlerContext ctx, String path) throws Exception {
    java.io.File f = new java.io.File(path);
    java.io.FileInputStream in = new java.io.FileInputStream(f);
    ctx.writeAndFlush(new DefaultFileRegion(in.getChannel(), 0, f.length()));
}

实训11.2 线程参数调优

📝 题目描述

配置 boss/worker 线程数与连接 backlog。

💡 解题思路

backlog 控制等待队列长度；线程数按 CPU 与连接规模调整。

import io.netty.channel.nio.NioEventLoopGroup;

// worker 线程数：CPU 核数 * 2 为默认，IO 密集可适当增加
EventLoopGroup worker = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2);

ServerBootstrap b = new ServerBootstrap();
b.option(ChannelOption.SO_BACKLOG, 1024)
 .childOption(ChannelOption.TCP_NODELAY, true)
 .childOption(ChannelOption.SO_KEEPALIVE, true);

实训11.3 Allocator 与内存池

📝 题目描述

说明池化内存分配器的作用与配置方式。

💡 解题思路

PooledByteBufAllocator 复用内存块减少 GC；Direct 内存减少内核拷贝。

import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.ChannelOption;

// 显式指定池化分配器
ServerBootstrap b = new ServerBootstrap();
b.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);

// 关闭自动释放检测可提升性能（需手动管理引用计数）
b.childOption(ChannelOption.AUTO_READ, true);

单元12 · 综合项目实训

综合运用所学知识完成项目

综合应用：协议设计 + 编解码 + 反射调用 + 动态代理。

实训12.1 简易 RPC - 协议设计

📝 题目描述

设计 RPC 请求协议：魔数 + 版本 + 序列化方式 + 消息体长度。

💡 解题思路

自定义长度字段协议解决粘包，魔数校验合法性。

public class RpcHeader {
    public static final int MAGIC = 0xCAFE;
    public static final int VERSION = 1;

    // 魔数(4) + 版本(1) + 序列化方式(1) + 消息体长度(4)
    // 长度头：lengthFieldOffset=0, lengthFieldLength=4
    // 用 LengthFieldBasedFrameDecoder(64*1024, 6, 4, 0, 10) 解码
    private int magic;
    private byte version;
    private byte serializeType;
    private int bodyLength;
}

实训12.2 简易 RPC - 服务端

📝 题目描述

实现 RPC 服务端：解码请求、反射调用、编码响应。

💡 解题思路

pipeline 组合解码器/业务处理器/编码器，反射调用本地服务实现。

import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;

public class RpcServerInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel ch) {
        ch.pipeline()
          .addLast(new RpcDecoder())          // 解码请求
          .addLast(new RpcServerHandler())    // 反射调用服务
          .addLast(new RpcEncoder());         // 编码响应
    }
}

实训12.3 简易 RPC - 动态代理

📝 题目描述

客户端使用 JDK 动态代理隐藏网络调用细节。

💡 解题思路

代理拦截接口调用，序列化方法名与参数发送到服务端，返回结果。

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;

public class RpcProxy {

    @SuppressWarnings("unchecked")
    public static <T> T create(Class<T> interfaceClass, RpcClient client) {
        return (T) Proxy.newProxyInstance(
                interfaceClass.getClassLoader(),
                new Class[]{interfaceClass},
                (proxy, method, args) -> client.invoke(
                        interfaceClass.getName(), method.getName(), args));
    }
}