netty

Netty

  • Netty提供异步的、事件驱动的网络应用程序框架和工具，用以快速开发高性能、高可靠性的网络服务器和客户端程序。

  • Netty 是一个基于NIO的客户、服务器端编程框架，使用Netty 可以确保你快速和简单的开发出一个网络应用，例如实现了某种协议的客户，服务端应用。Netty相当简化和流线化了网络应用的编程开发过程，例如，TCP和UDP的socket服务开发。

  • image-20210220222827506

  • Pipeline是一个双向链表

Channel

  • 它代表一个到实体（ 如一个硬件设备、一个文件、一个网络套接字或者一个能够执行一个或者多个不同的I/O操作的程序组件） 的开放连接，如读操作和写操作

  • 会为每个Channel分配一个EventHandler

  • 在netty中，每个Channel中都有且仅有一个ChannelPipeline与之对应.其持有一个 ChannelHandler的实例链。在默认的情况下， ChannelHandler 会把对它的方法的调用转发给链中的下一个 ChannelHandler。

  • img

ChannelPipeline

  • 一个 Channel 包含了一个 ChannelPipeline， 而 ChannelPipeline 中又维护了一个由 ChannelHandlerContext组成的双向链表， 并且每个 ChannelHandlerContext 中又关联着一个 ChannelHandler

ChannelHandlerContext

  • ChannelHandlerContext 参数贯穿 ChannelPipeline，将信息传递给每个 ChannelHandler，是个合格的“通讯员”。

channelHandler

  • ChannelPipeline中可以有多个handler,多个channel共享一个handler可能会有并发安全问题，需要加入@shared注解

ChannelOption

  • Netty 在创建 Channel 实例后,一般都需要设置 ChannelOption 参数。
  • 常见的设置
    • ChannelOption.SO_BACKLOG
      对应 TCP/IP 协议 listen 函数中的 backlog 参数， 用来初始化服务器可连接队列大小。 服
      务端处理客户端连接请求是顺序处理的， 所以同一时间只能处理一个客户端连接。 多个客户
      端来的时候， 服务端将不能处理的客户端连接请求放在队列中等待处理， backlog 参数指定
      了队列的大小。
    • ChannelOption.SO_KEEPALIVE
      一直保持连接活动状态

ChannelFuture

  • 可以通过 ChannelFuture 接口的 addListener() 方法注册一个 ChannelFutureListener，当操作执行成功或者失败时，监听就会自动触发返回结果。
  • 还可以通过 ChannelFuture 接口的 sync()方法让异步的操作变成同步的

EventLoop 和 EventLoopGroup

  • 在 Netty 中每个 Channel 都会被分配到一个 EventLoop。一个 EventLoop 可以服务于多个 Channel。
  • 每个 EventLoop 会占用一个 Thread，同时这个 Thread 会处理 这个EventLoop 上面发生的所有 IO 操作和事件（Netty 4.0）。
  • EventLoopGroup 是用来生成 EventLoop，EventLoopGroup 要做的就是创建一个新的 Channel，并且给它分配一个 EventLoop。默认EventLoopGroup 的线程数为cpu核心数*2

任务队列

  • scheduleTaskQueue.schedule
  • taskQueue.execute

Netty心跳检测机制

  • 心跳机制的工作原理是: 在 client 与 server 之间在一定时间内没有数据交互时, 即处于 idle 状态时, 客户端或服务器就会发送一个特殊的数据包给对方, 当接收方收到这个数据报文后, 也立即发送一个特殊的数据报文, 回应发送方, 此即一个 PING-PONG 交互。所以, 当某一端收到心跳消息后, 就知道了对方仍然在线, 这就确保 TCP 连接的有效性.
  • 通过 Netty 实现心跳机制的话，核心类是 IdleStateHandler 。

ByteBuf

  • Netty 提供了两种 ByteBufAllocator 的实现
    • PooledByteBufAllocator，实现了 ByteBuf 的对象的池化，提高性能减少内存碎片。
    • Unpooled-ByteBufAllocator，没有实现对象的池化，每次会生成新的对象实例。

Netty 的 Bootstrap

  • Bootstarp 和 ServerBootstrap 被称为引导类，指对应用程序进行配置，并使他运行起来的过程。Netty处理引导的方式是使你的应用程序和网络层相隔离。

Bootstrap

  • Bootstrap 是客户端的引导类，Bootstrap 在调用 bind()（连接UDP）和 connect()（连接TCP）方法时，会新创建一个 Channel，仅创建一个单独的、没有父 Channel 的 Channel 来实现所有的网络交换。

ServerBootstrap

  • ServerBootstrap 是服务端的引导类，ServerBootstarp 在调用 bind() 方法时会创建一个 ServerChannel 来接受来自客户端的连接，并且该 ServerChannel 管理了多个子 Channel 用于同客户端之间的通信。

Netty拆包粘包

使用netty自带的解码器

  • LineBasedFrameDecoder : 发送端发送数据包的时候，每个数据包之间以换行符作为分隔，LineBasedFrameDecoder 的工作原理是它依次遍历 ByteBuf 中的可读字节，判断是否有换行符，然后进行相应的截取。
  • DelimiterBasedFrameDecoder : 可以自定义分隔符解码器，**LineBasedFrameDecoder** 实际上是一种特殊的 DelimiterBasedFrameDecoder 解码器。
  • FixedLengthFrameDecoder: 固定长度解码器，它能够按照指定的长度对消息进行相应的拆包。
  • **LengthFieldBasedFrameDecoder**：

自定义序列化编码解码器

  • 通常情况下，我们使用 Protostuff、Hessian2、json 序列方式比较多

Netty的零拷贝

  • Netty的 Zero-coyp 完全是在用户态(Java 层面)的, 它的 Zero-copy 的更多的是偏向于 优化数据操作 这样的概念.

特点

  • Netty 提供了 CompositeByteBuf 类, 它可以将多个 ByteBuf 合并为一个逻辑上的 ByteBuf, 避免了各个 ByteBuf 之间的拷贝.
  • 通过 wrap 操作, 我们可以将 byte[] 数组、ByteBuf、ByteBuffer等包装成一个 Netty ByteBuf 对象, 进而避免了拷贝操作.
  • ByteBuf 支持 slice 操作, 因此可以将 ByteBuf 分解为多个共享同一个存储区域的 ByteBuf, 避免了内存的拷贝.
  • 通过 FileRegion 包装的FileChannel.tranferTo 实现文件传输, 可以直接将文件缓冲区的数据发送到目标 Channel, 避免了传统通过循环 write 方式导致的内存拷贝问题.

通过 CompositeByteBuf 实现零拷贝

  • ByteBuf header = ...
ByteBuf body = ...

CompositeByteBuf compositeByteBuf = Unpooled.compositeBuffer();
//第一个true表表示当添加新的 ByteBuf 时, 自动递增 CompositeByteBuf 的 writeIndex。如果不添加
//true参数， 因此此时我们就不可能从 compositeByteBuf 中读取到数据。
compositeByteBuf.addComponents(true, header, body);
//还可以使用这种方法
ByteBuf allByteBuf = Unpooled.wrappedBuffer(header, body);

    1
    2
    3
    4
    5
    6
    7
    +  虽然看起来 CompositeByteBuf 是由两个 ByteBuf 组合而成的, 不过在 CompositeByteBuf 内部, 这两个 ByteBuf 都是单独存在的, CompositeByteBuf 只是逻辑上是一个整体.
    #### 通过 wrap 操作实现零拷贝
    + 例如我们有一个 byte 数组, 我们希望将它转换为一个 ByteBuf 对象, 以便于后续的操作, 那么传统的做法是将此 byte 数组拷贝到 ByteBuf 中, 即:
      ```java
      byte[] bytes = ...
      ByteBuf byteBuf = Unpooled.buffer();
      byteBuf.writeBytes(bytes);
  • 显然这样的方式也是有一个额外的拷贝操作的, 我们可以使用 Unpooled 的相关方法, 包装这个 byte 数组, 生成一个新的 ByteBuf 实例, 而不需要进行拷贝操作. 上面的代码可以改为:
    1
    2
    byte[] bytes = ...
    ByteBuf byteBuf = Unpooled.wrappedBuffer(bytes);
  • 我们通过 Unpooled.wrappedBuffer 方法来将 bytes 包装成为一个 UnpooledHeapByteBuf 对象, 而在包装的过程中, 是不会有拷贝操作的. 即最后我们生成的生成的 ByteBuf 对象是和 bytes 数组共用了同一个存储空间, 对 bytes 的修改也会反映到 ByteBuf 对象中。

通过 slice 操作实现零拷贝

  • slice 操作和 wrap 操作刚好相反, Unpooled.wrappedBuffer 可以将多个 ByteBuf 合并为一个, 而 slice 操作可以将一个 ByteBuf 切片 为多个共享一个存储区域的 ByteBuf 对象.
  • //等同于 buf.slice(buf.readerIndex(), buf.readableBytes()) 调用
public ByteBuf slice();
//
public ByteBuf slice(int index, int length);

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    + 用 `slice` 方法产生 header 和 body 的过程是没有拷贝操作的, header 和 body 对象在内部其实是共享了 byteBuf 存储空间的不同部分而已. 
    #### 通过 FileRegion 实现零拷贝
    + Netty 中使用 FileRegion 实现文件传输的零拷贝, 不过在底层 FileRegion 是依赖于 Java NIO `FileChannel.transfer` 的零拷贝功能.
    +  使用了 `FileChannel` 后, 我们就可以直接将源文件的内容直接拷贝(`transferTo`) 到目的文件中
    + ```java
      public static void copyFileWithFileChannel(String srcFileName, String destFileName) throws Exception {
          RandomAccessFile srcFile = new RandomAccessFile(srcFileName, "r");
          FileChannel srcFileChannel = srcFile.getChannel();
      
          RandomAccessFile destFile = new RandomAccessFile(destFileName, "rw");
          FileChannel destFileChannel = destFile.getChannel();
      
          long position = 0;
          long count = srcFileChannel.size();
      
          srcFileChannel.transferTo(position, count, destFileChannel);
      }
  • 具体使用
    • @Override
public void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
    RandomAccessFile raf = null;
    long length = -1;
    try {
        // 1. 通过 RandomAccessFile 打开一个文件.
        raf = new RandomAccessFile(msg, "r");
        length = raf.length();
    } catch (Exception e) {
        ctx.writeAndFlush("ERR: " + e.getClass().getSimpleName() + ": " + e.getMessage() + '\n');
        return;
    } finally {
        if (length < 0 && raf != null) {
            raf.close();
        }
    }

    ctx.write("OK: " + raf.length() + '\n');
    if (ctx.pipeline().get(SslHandler.class) == null) {
        // SSL not enabled - can use zero-copy file transfer.
        // 2. 调用 raf.getChannel() 获取一个 FileChannel.
        // 3. 将 FileChannel 封装成一个 DefaultFileRegion
        ctx.write(new DefaultFileRegion(raf.getChannel(), 0, length));
    } else {
        // SSL enabled - cannot use zero-copy file transfer.
        ctx.write(new ChunkedFile(raf));
    }
    ctx.writeAndFlush("\n");
}

Protobuf

  • 轻便高效的结构化数据存储格式，可以用于结构化数据串行化，很适合做数据存储或rpc的数据交换格式
11技术栈
netty
https://x-leonidas.github.io/2025/10/26/11技术栈/netty/
作者
听风
发布于
2025年10月26日
更新于
2022年5月5日
许可协议