Java `Netty` `EventLoopGroup` `ChannelPipeline` `Custom Codec` 高性能网络编程 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

各位观众老爷，大家好！我是今天的主讲人，江湖人称“代码搬运工”，今天咱们就来聊聊Java Netty这玩意儿，保证让各位听完之后，感觉自己也能轻松驾驭高性能网络编程。

废话不多说，咱们直接上干货！

开场白：Netty 是个啥？

简单来说，Netty就是一个高性能、异步事件驱动的网络应用框架。想象一下，你开了一家咖啡馆，Netty就是你的服务员团队，他们负责处理顾客（网络请求）的点单、制作咖啡、送餐等等，而且效率奇高，能同时服务很多顾客，还不容易出错。

第一幕：EventLoopGroup – 咖啡馆的经理

EventLoopGroup，可以理解为咖啡馆的经理团队，负责管理整个咖啡馆的运作。他们会根据顾客的数量和需求，安排服务员去工作。Netty中，EventLoopGroup 主要负责两件事：

Acceptor Group (老板)：接受新的连接。就像咖啡馆门口的迎宾，负责招呼新来的顾客。
Worker Group (员工)：处理具体的I/O事件，比如读写数据。就像服务员，负责点餐、送餐等。

代码示例：

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;

public class NettyServer {
    public static void main(String[] args) throws Exception {
        // 创建Boss线程组，用于接收客户端连接
        EventLoopGroup bossGroup = new NioEventLoopGroup(1); // 通常设置为 CPU 核心数 * 2
        // 创建Worker线程组，用于处理网络I/O
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class) // 使用NioServerSocketChannel作为服务器通道实现
             .childHandler(new ChannelInitializer<SocketChannel>() { // 客户端连接建立后，执行ChannelPipeline的初始化
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     // 添加ChannelHandler，后续会详细讲解
                     ch.pipeline().addLast(new YourCustomHandler());
                 }
             });

            // 绑定端口，开始接收进来的连接
            ChannelFuture f = b.bind(8080).sync();

            // 等待服务器  socket 关闭 。
            f.channel().closeFuture().sync();

        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
}

解释一下：

NioEventLoopGroup：基于NIO实现的EventLoopGroup。
bossGroup：负责监听端口，接受新的连接。参数 1 表示只使用一个线程来处理连接请求，这个线程会不断地接受新的客户端连接。对于高并发场景，你可以根据 CPU 核心数进行调整，例如 new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2)。
workerGroup：负责处理实际的I/O操作，比如读取客户端发送的数据。如果不指定线程数，Netty 会默认使用 CPU 核心数 * 2 的线程数。
ServerBootstrap：用于启动服务器，配置各种参数。
channel(NioServerSocketChannel.class)：指定使用NIO的ServerSocketChannel。
childHandler()：为每个新连接的客户端设置ChannelPipeline。

第二幕：ChannelPipeline – 咖啡馆的流水线

ChannelPipeline，就像咖啡馆的流水线，每个步骤都有不同的工作人员负责。比如，有人负责接收订单，有人负责磨咖啡豆，有人负责调制咖啡，最后有人负责送餐。

Netty中，ChannelPipeline 是一系列的 ChannelHandler 组成的，每个 ChannelHandler 负责处理不同的事件，比如：

入站事件 (Inbound)：从客户端发送到服务器的数据。
出站事件 (Outbound)：从服务器发送到客户端的数据。

代码示例（接上面的代码）：

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

public class YourCustomHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 处理客户端发送的数据
        ByteBuf buf = (ByteBuf) msg;
        try {
            // 将数据读取到字节数组
            byte[] req = new byte[buf.readableBytes()];
            buf.readBytes(req);
            String body = new String(req, "UTF-8");
            System.out.println("Received : " + body);

            // 给客户端发送响应
            String currentTime = "Hello Client!";
            ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
            ctx.write(resp); // 将响应写入到缓冲区
        } finally {
            ReferenceCountUtil.release(msg); // 释放资源，防止内存泄漏
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush(); // 将缓冲区的数据发送到客户端
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close(); // 发生异常时，关闭连接
    }
}

解释一下：

ChannelInboundHandlerAdapter：用于处理入站事件的适配器。
channelRead()：当接收到客户端发送的数据时，会调用这个方法。
channelReadComplete()：当一次完整的读取操作完成后，会调用这个方法。
exceptionCaught()：当发生异常时，会调用这个方法。
ByteBuf：Netty提供的字节缓冲区，用于存储数据。
Unpooled.copiedBuffer()：创建一个新的ByteBuf，并将数据复制到其中。
ctx.write()：将数据写入到缓冲区。
ctx.flush()：将缓冲区的数据发送到客户端。
ReferenceCountUtil.release(msg)：释放资源，防止内存泄漏。Netty 使用引用计数来管理 ByteBuf 的生命周期，每次读取完数据后，必须手动释放资源。

第三幕：Custom Codec – 咖啡馆的翻译官

Custom Codec，就像咖啡馆的翻译官，负责将顾客的语言（二进制数据）翻译成咖啡师能理解的语言（Java对象），反之亦然。

Netty中，Codec 分为两种：

Encoder (编码器)：将 Java 对象编码成二进制数据。
Decoder (解码器)：将二进制数据解码成 Java 对象。

为什么要自定义 Codec 呢？因为网络传输的数据都是二进制的，而我们的业务逻辑通常是基于 Java 对象的。所以，我们需要一个 Codec 来完成这两者之间的转换。

代码示例：

首先，定义一个简单的Java对象：

public class Message {
    private int id;
    private String content;

    // 省略 getter 和 setter 方法

    public Message(int id, String content) {
        this.id = id;
        this.content = content;
    }

    public int getId() {
        return id;
    }

    public void setId(int id) {
        this.id = id;
    }

    public String getContent() {
        return content;
    }

    public void setContent(String content) {
        this.content = content;
    }
}

然后，创建一个Encoder：

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;

public class MessageEncoder extends MessageToByteEncoder<Message> {

    @Override
    protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) throws Exception {
        out.writeInt(msg.getId()); // 写入消息ID
        byte[] contentBytes = msg.getContent().getBytes("UTF-8");
        out.writeInt(contentBytes.length); // 写入消息内容长度
        out.writeBytes(contentBytes); // 写入消息内容
    }
}

接着，创建一个Decoder：

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

import java.util.List;

public class MessageDecoder extends ByteToMessageDecoder {

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        // 确保有足够的数据可读
        if (in.readableBytes() < 8) { // ID (4 bytes) + Content Length (4 bytes)
            return;
        }

        in.markReaderIndex(); // 标记当前读取位置

        int id = in.readInt(); // 读取消息ID
        int contentLength = in.readInt(); // 读取消息内容长度

        // 检查是否接收到完整的消息内容
        if (in.readableBytes() < contentLength) {
            in.resetReaderIndex(); // 重置读取位置，等待更多数据
            return;
        }

        byte[] contentBytes = new byte[contentLength];
        in.readBytes(contentBytes); // 读取消息内容
        String content = new String(contentBytes, "UTF-8");

        Message message = new Message(id, content);
        out.add(message); // 将解码后的消息添加到输出列表
    }
}

最后，将 Encoder 和 Decoder 添加到 ChannelPipeline 中：

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

public class NettyServerInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ch.pipeline().addLast(new MessageDecoder()); // 添加解码器
        ch.pipeline().addLast(new MessageEncoder()); // 添加编码器
        ch.pipeline().addLast(new YourBusinessHandler()); // 添加业务处理器
    }
}

解释一下：

MessageToByteEncoder：用于将消息编码成字节流的编码器。
ByteToMessageDecoder：用于将字节流解码成消息的解码器。
encode()：将 Java 对象编码成 ByteBuf。
decode()：将 ByteBuf 解码成 Java 对象。
in.readableBytes()：返回可读的字节数。
in.markReaderIndex()：标记当前读取位置。
in.resetReaderIndex()：重置读取位置。

第四幕：高性能秘诀 – 咖啡馆的运营策略

要想让咖啡馆生意兴隆，光有好的服务员和流水线还不够，还需要一些运营策略。Netty 的高性能也依赖于一些关键的策略：

异步非阻塞 I/O (Asynchronous Non-blocking I/O)：Netty 基于 NIO 实现，采用非阻塞 I/O 模型，可以同时处理大量的并发连接，而不会阻塞线程。
零拷贝 (Zero-Copy)：Netty 尽量减少数据的拷贝次数，提高数据传输效率。例如，使用 FileRegion 可以直接将文件内容发送到网络，而不需要将文件内容复制到内存中。
内存池 (Pooled ByteBuf)：Netty 使用内存池来管理 ByteBuf，避免频繁创建和销毁 ByteBuf，减少内存碎片，提高内存利用率。
线程模型 (Thread Model)：Netty 的线程模型设计合理，可以充分利用多核 CPU 的性能，提高并发处理能力。
无锁化编程 (Lock-Free Programming)：Netty 在一些关键的地方使用了无锁化编程技术，避免线程竞争，提高性能。

| 特性 | 描述 |
| 异步非阻塞 I/O | 使用 NIO 提供的非阻塞 I/O 操作，可以同时处理多个连接，而无需等待 I/O 操作完成。
| 零拷贝 | 通过减少数据在内核空间和用户空间之间的复制次数来提高性能。

发表回复 取消回复

发表回复取消回复