Netty入门

概述

Netty is an asynchronous event-driven network application framework
for rapid development of maintainable high performance protocol servers & clients.

Netty是一个异步的、基于事件驱动的网络应用框架，用于快速开发可维护、高性能的网络服务端和客户端

Hello World

实现目标：简单的服务端和客户端通信

先加入依赖

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

Server

//启动器，负责组装netty组件，启动服务器
new ServerBootstrap()
        //BossEventLoop，WorkerEventLoop，group
        .group(new NioEventLoopGroup())
        //选择服务器的 ServerSocketChannel 的实现
        .channel(NioServerSocketChannel.class)// NIO BIO OIO
        //boss 处理连接 worker(child)处理读写，决定了worker（child）能做什么(handler)
        .childHandler(
            //Channel 代表和客户端进行数据读写的通道，Initializer 初始化器（负责添加别的handler）
            new ChannelInitializer<NioSocketChannel>() {
            @Override
            protected void initChannel(NioSocketChannel ch) throws Exception {
                //添加具体的handler
                ch.pipeline().addLast(new StringDecoder());//将byteBuff转换为String
                ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {//自定义handler
                    @Override //读事件
                    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                        System.out.println(msg);
                    }
                });
            }
        })
        //绑定端口
        .bind(8080);

1. 创建 NioEventLoopGroup，可以简单理解为 线程池 + Selector
2. 选择服务 Socket 实现类，其中 NioServerSocketChannel 表示基于NIO的服务端实现
3. 负责给 SocketChannel 添加处理器，而不是给 ServerSocketChannel，ChannelInitializer处理器只执行一次，它的作用是待客户端 SocketChannel 建立连接后，执行 initChannel 以添加更多的处理器
4. StringDecoder 处理器 ，解码 ByteBuf => String
5. ChannelInboundHandlerAdapter 业务处理器，使用上一个处理器的结果
6. bind 绑定端口

Click

//1.启动类
      new Bootstrap()
              //2.添加EventLoop
              .group(new NioEventLoopGroup())
              //3.选择客户端channel实现
              .channel(NioSocketChannel.class)
              //4.添加处理器
              .handler(new ChannelInitializer<NioSocketChannel>() {
                  @Override //在建立链接后被调用
                  protected void initChannel(NioSocketChannel sc) throws Exception {
                      sc.pipeline().addLast(new StringEncoder());
                  }
              })
              //5.链接到服务器
              .connect(new InetSocketAddress("localhost",8080))
              //阻塞方法，直到链接建立
              .sync()
              //代表连接对象
              .channel()
              //6.发送消息
              .writeAndFlush("hello world");

---

组件

EventLoop

EventLoop 中文可以解读为 事件循环对象

EventLoop本质是一个单线程执行器（同时维护了一个Selector），里面有run方法处理Channel上源源不断的io事件。

它的继承关系比较复杂：

• 一条线是继承了j.u.c.ScheduledExecutorService因此包含了线程池中所有的方法
• 另一条是继承了netty的OrderedEventExecutor
  • 提供了boolean inEventLoop(Thread thread)方法判断一个线程是否属于此EventLoop
  • 提供了parent方法来看看自己属于哪个EventLoopGroup

EventLoopGroup中文可以解读为 事件循环组

EventLoopGroup就是一组EventLoop，Channel一般会调用EventLoopGroup的 register 方法绑定其中一个EventLoop，后续这个Channel上的io事件都由此EventLoop来处理（保证了io事件处理时的线程安全）

• 继承自netty的EventExecutorGroup
  • 实现了iterable接口提供遍历EventLoop的能力
  • 另有next方法获取集合中下一个EventLoop

简单案例：

// 内部创建了两个 EventLoop, 每个 EventLoop 维护一个线程
DefaultEventLoopGroup group = new DefaultEventLoopGroup(2);
System.out.println(group.next());
System.out.println(group.next());
System.out.println(group.next());

// for循环输出
DefaultEventLoopGroup group = new DefaultEventLoopGroup(2);
for (EventExecutor eventLoop : group) {
    System.out.println(eventLoop);
}

输出

io.netty.channel.DefaultEventLoop@60f82f98
io.netty.channel.DefaultEventLoop@35f983a6
io.netty.channel.DefaultEventLoop@60f82f98

NioEventLoop处理 io 事件

服务端 EventLoopGroup 长度为 2

new ServerBootstrap()
    //这里对工作内容做了细分 跟NIO中的差不多 分成了 Boss(第一个参数) 和 Worker(第二个参数) 两个
    //细分：boss只负责 ServerSockerChannel 上的 asspet 事件，worker 只负责 socketChannel 上的读写
    //boss长度一般填 1 即可，worker 根据实际情况来
    .group(new NioEventLoopGroup(1), new NioEventLoopGroup(2))
    .channel(NioServerSocketChannel.class)
    .childHandler(new ChannelInitializer<NioSocketChannel>() {
        @Override
        protected void initChannel(NioSocketChannel ch) {
            ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                @Override
                public void channelRead(ChannelHandlerContext ctx, Object msg) {
                    ByteBuf byteBuf = msg instanceof ByteBuf ? ((ByteBuf) msg) : null;
                    if (byteBuf != null) {
                        byte[] buf = new byte[16];
                        ByteBuf len = byteBuf.readBytes(buf, 0, byteBuf.readableBytes());
                        log.debug(new String(buf));
                    }
                }
            });
        }
    }).bind(8080).sync();

客户端，启动多个，发送内容随意(不相同即可)

public static void main(String[] args) throws InterruptedException {
    Channel channel = new Bootstrap()
            .group(new NioEventLoopGroup(1))
            .handler(new ChannelInitializer<NioSocketChannel>() {
                @Override
                protected void initChannel(NioSocketChannel ch) throws Exception {
                    ch.pipeline().addLast(new StringEncoder());
                }
            })
            .channel(NioSocketChannel.class).connect("localhost", 8080)
            .sync()
            .channel();

    channel.writeAndFlush(ByteBufAllocator.DEFAULT.buffer().writeBytes("AAA".getBytes()));
    Thread.sleep(2000);
    channel.writeAndFlush(ByteBufAllocator.DEFAULT.buffer().writeBytes("AAA".getBytes()));
}

服务端最后输出

22:03:34 [DEBUG] [nioEventLoopGroup-3-1] c.i.o.EventLoopTest - AAA       
22:03:36 [DEBUG] [nioEventLoopGroup-3-1] c.i.o.EventLoopTest - AAA       
22:05:36 [DEBUG] [nioEventLoopGroup-3-2] c.i.o.EventLoopTest - BBB           
22:05:38 [DEBUG] [nioEventLoopGroup-3-2] c.i.o.EventLoopTest - BBB           
22:06:09 [DEBUG] [nioEventLoopGroup-3-1] c.i.o.EventLoopTest - CCC        
22:06:11 [DEBUG] [nioEventLoopGroup-3-1] c.i.o.EventLoopTest - CCC    

可以看到两个 EventLoop 轮流处理 channel，但工人与 channel 之间进行了绑定

多EventLoopGroup

多 EventLoopGroup 用于更细一部的细分，你可以将某些耗时较大的操作，交给另一个 EventLoopGroup

客户端代码不变，服务端代码做修改：

//细分，创建一个独立的 EventLoopGroup
DefaultEventLoopGroup defaultGroup = new DefaultEventLoopGroup();

new ServerBootstrap()
    //这里对工作内容做了细分 跟NIO中的差不多 分成了 Boss(第一个参数) 和 Worker(第二个参数) 两个
    //细分：boss只负责 ServerSockerChannel 上的 asspet 事件，worker 只负责 socketChannel 上的读写
    //boss长度一般填 1 即可，worker 根据实际情况来
    .group(new NioEventLoopGroup(),new NioEventLoopGroup(2))
    .channel(NioServerSocketChannel.class)
    .childHandler(new ChannelInitializer<NioSocketChannel>() {
        @Override
        protected void initChannel(NioSocketChannel ch) throws Exception {
            ch.pipeline()
                .addLast("handler1",new ChannelInboundHandlerAdapter(){
                    @Override
                    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                        ByteBuf buf = (ByteBuf) msg;
                        log.debug(buf.toString(Charset.defaultCharset()));
                        ctx.fireChannelRead(msg); //将消息传递给下一个handler
                    }
                })
                .addLast(defaultGroup,"handler2",new ChannelInboundHandlerAdapter(){
                    @Override
                    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                        ByteBuf buf = (ByteBuf) msg;
                        log.debug(buf.toString(Charset.defaultCharset()));
                    }
                });
        }
    })
    .bind(8080);

输出

15:12:28.241 [nioEventLoopGroup-4-2] DEBUG cn.itcast.netty.c3.EventLoopServer - 001
15:12:28.241 [defaultEventLoopGroup-2-4] DEBUG cn.itcast.netty.c3.EventLoopServer - 001
15:12:34.529 [nioEventLoopGroup-4-1] DEBUG cn.itcast.netty.c3.EventLoopServer - 002
15:12:34.530 [defaultEventLoopGroup-2-5] DEBUG cn.itcast.netty.c3.EventLoopServer - 002

handler 执行中的切换

也就是上图中 h1 -> h2 的切换

关键代码io.netty.channel.AbstractChannelHandlerContext#invokeChannelRead()

static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
    final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);
    // 下一个 handler 的事件循环是否与当前的事件循环是同一个线程
    EventExecutor executor = next.executor();
    
    // 当前 handler 的线程是否和 eventLoop 是同一个线程
    // 是，直接调用
    if (executor.inEventLoop()) {
        next.invokeChannelRead(m);
    } 
    // 不是，将要执行的代码作为任务提交给下一个事件循环处理（换人）
    else {
        executor.execute(new Runnable() {
            @Override
            public void run() {
                next.invokeChannelRead(m);
            }
        });
    }
}

• 如果两个 handler 绑定的是同一个线程，那么就直接调用
• 否则，把要调用的代码封装为一个任务对象，由下一个 handler 的线程来调用

NioEventLoop处理普通任务

NioEventLoop 除了可以处理 io 事件，同样可以向它提交普通任务

NioEventLoopGroup nioWorkers = new NioEventLoopGroup(2);

log.debug("server start...");
Thread.sleep(2000);
nioWorkers.execute(()->{
    log.debug("normal task...");
});

输出

22:30:36 [DEBUG] [main] c.i.o.EventLoopTest2 - server start...
22:30:38 [DEBUG] [nioEventLoopGroup-2-1] c.i.o.EventLoopTest2 - normal task...

可以用来执行耗时较长的任务

NioEventLoop处理定时任务

NioEventLoopGroup nioWorkers = new NioEventLoopGroup(2);

log.debug("server start...");
Thread.sleep(2000);
//定时任务 参数1::执行对象,参数2::延迟执行时间,参数3::间隔执行时间,参数4::时间单位
nioWorkers.scheduleAtFixedRate(() -> {
    log.debug("running...");
}, 0, 1, TimeUnit.SECONDS);

输出

22:35:15 [DEBUG] [main] c.i.o.EventLoopTest2 - server start...
22:35:17 [DEBUG] [nioEventLoopGroup-2-1] c.i.o.EventLoopTest2 - running...
22:35:18 [DEBUG] [nioEventLoopGroup-2-1] c.i.o.EventLoopTest2 - running...
22:35:19 [DEBUG] [nioEventLoopGroup-2-1] c.i.o.EventLoopTest2 - running...
22:35:20 [DEBUG] [nioEventLoopGroup-2-1] c.i.o.EventLoopTest2 - running...
...

可以用来执行定时任务

---

Channel

channel的主要方法：

• close() 可以用来关闭 channel
• closeFuture() 用来处理 channel 的关闭
  • sync 方法作用是同步等待 channel 关闭
  • 而 addListener 方法是异步等待 channel 关闭
• pipeline() 方法是添加处理器
• write() 方法是数据写入
• writeAndFlush() 方法是数据写入并立刻发送

ChannelFuture

之前客户端代码

new Bootstrap()
    .group(new NioEventLoopGroup())
    .channel(NioSocketChannel.class)
    .handler(new ChannelInitializer<Channel>() {
        @Override
        protected void initChannel(Channel ch) {
            ch.pipeline().addLast(new StringEncoder());
        }
    })
    .connect("127.0.0.1", 8080)
    .sync()
    .channel()
    .writeAndFlush(new Date() + ": hello world!");

把它们拆开

ChannelFuture channelFuture = new Bootstrap()
    .group(new NioEventLoopGroup())
    .channel(NioSocketChannel.class)
    .handler(new ChannelInitializer<Channel>() {
        @Override
        protected void initChannel(Channel ch) {
            ch.pipeline().addLast(new StringEncoder());
        }
    })
    .connect("127.0.0.1", 8080); // 1

channelFuture.sync().channel().writeAndFlush(new Date() + ": hello world!");

connect() 返回一个 ChannelFuture 对象，它的作用是利用 channel() 方法获取 Channel 对象

注意：connect() 是异步方法，意味着不等待连接建立，方法执行就返回了，因此 channelFuture 无法立刻获取返回的 Channel 对象，尝试如下：

System.out.println(channelFuture.channel()); // 1 [id: 0x2e1884dd]
channelFuture.sync(); // 2 sync 方法是同步等待连接建立完成
System.out.println(channelFuture.channel()); // 3 [id: 0x2e1884dd, L:/127.0.0.1:57191 - R:/127.0.0.1:8080]
//这种方法是同步的

除了 sync() 方法，也可以让异步操作执行，还可以使用回调：

ChannelFuture channelFuture = new Bootstrap()
    .group(new NioEventLoopGroup())
    .channel(NioSocketChannel.class)
    .handler(new ChannelInitializer<Channel>() {
        @Override
        protected void initChannel(Channel ch) {
            ch.pipeline().addLast(new StringEncoder());
        }
    })
    .connect("127.0.0.1", 8080);
System.out.println(channelFuture.channel()); // 1 [id: 0x749124ba]
channelFuture.addListener((ChannelFutureListener) future -> {
    System.out.println(future.channel()); // 2 ChannelFutureListener 会在连接建立时被调用（其中 operationComplete 方法），因此执行到此时，连接肯定建立了，打印 [id: 0x749124ba, L:/127.0.0.1:57351 - R:/127.0.0.1:8080]
});

CloseFuture

CloseFuture 可以用来关闭连接

@Slf4j
public class CloseFutureClient {
    public static void main(String[] args) throws InterruptedException {
        NioEventLoopGroup group new NioEventLoopGroup();
        ChannelFuture channelFuture = new Bootstrap()
                .group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override // 在连接建立后被调用
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                        ch.pipeline().addLast(new StringEncoder());
                    }
                })
                .connect(new InetSocketAddress("localhost", 8080));
        Channel channel = channelFuture.sync().channel();
        log.debug("{}", channel);
        new Thread(()->{
            Scanner scanner = new Scanner(System.in);
            while (true) {
                String line = scanner.nextLine();
                if ("q".equals(line)) {
                    channel.close(); // close 异步操作 1s 之后
					//log.debug("处理关闭之后的操作"); // 不能在这里善后
                    break;
                }
                channel.writeAndFlush(line);
            }
        }, "input").start();

        // 获取 CloseFuture 对象， 1) 同步处理关闭， 2) 异步处理关闭
        ChannelFuture closeFuture = channel.closeFuture();
        /*log.debug("waiting close...");
        closeFuture.sync();
        log.debug("处理关闭之后的操作");*/
        closeFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                log.debug("处理关闭之后的操作");
                group.shutdownGracefully();
            }
        });
    }
}

问题

为什么不在同一个线程中去执行 链接建立、关闭Channel，这样不是也可以吗？非要这么复杂吗。

答案：

其实不仅仅是异步多线程效率高，在Netty中异步不一定提升效率，而是提高了吞吐量

• 单线程没法异步提高效率，必须配合多线程、多核 cpu 才能发挥异步的优势
• 异步并没有缩短响应时间，反而有所增加
• 合理进行任务拆分，也是利用异步的关键

Future & Promise

异步处理时，经常用到这两个接口

Future 需要注意，在 JDK 有相同的 Future，但是它们是两个接口，Netty 的 Future 继承自 JDK 的 Future，而 Promise 又对 Netty Future 进行了扩展。

• JDK Future 只能同步等待任务结束（成功 或 失败）才能得到结果
• Netty Future 可以同步等待任务结束得到结果，也可以异步的方式得到结果，但都要等任务结束
• Netty Promise 不仅有 Netty Future 的功能，而且脱离了任务独立存在，只作为两个线程间传递结果的容器

功能/名称	jdk Future	netty Future	Promise
cancel	取消任务	-	-
isCanceled	任务是否取消	-	-
isDone	任务是否完成，不能区分成功失败	-	-
get	获取任务结果，阻塞等待	-	-
getNow	-	获取任务结果，非阻塞，还未产生结果时返回 null	-
await	-	等待任务结束，如果任务失败，不会抛异常，而是通过 isSuccess 判断	-
sync	-	等待任务结束，如果任务失败，抛出异常	-
isSuccess	-	判断任务是否成功	-
cause	-	获取失败信息，非阻塞，如果没有失败，返回null	-
addLinstener	-	添加回调，异步接收结果	-
setSuccess	-	-	设置成功结果
setFailure	-	-	设置失败结果

Handler & Pipeline

ChannelHandler 用来处理 Channel 上的各种事件，分为入站、出站两种，所有 ChannelHandler 被连成一串，就是 Pipeline

• 入站处理器通常是 ChannelInboundHandlerAdapter 的子类，主要读取客户端数据，写返回结果
• 出站处理器通常是 ChannelOutboundHandlerAdapter 的子类，主要是对写返回结果进行加工

其基本结果大概就是 责任链模式，代码示例：

服务端：

new ServerBootstrap()
    .group(new NioEventLoopGroup())
    .channel(NioServerSocketChannel.class)
    .childHandler(new ChannelInitializer<NioSocketChannel>() {
        protected void initChannel(NioSocketChannel ch) {
            ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){
                @Override
                public void channelRead(ChannelHandlerContext ctx, Object msg) {
                    System.out.println(1);
                    ctx.fireChannelRead(msg); // 1
                }
            });
            ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){
                @Override
                public void channelRead(ChannelHandlerContext ctx, Object msg) {
                    System.out.println(2);
                    ctx.fireChannelRead(msg); // 2
                }
            });
            ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){
                @Override
                public void channelRead(ChannelHandlerContext ctx, Object msg) {
                    System.out.println(3);
                    ctx.channel().write(msg); // 3
                }
            });
            ch.pipeline().addLast(new ChannelOutboundHandlerAdapter(){
                @Override
                public void write(ChannelHandlerContext ctx, Object msg, 
                                  ChannelPromise promise) {
                    System.out.println(4);
                    ctx.write(msg, promise); // 4
                }
            });
            ch.pipeline().addLast(new ChannelOutboundHandlerAdapter(){
                @Override
                public void write(ChannelHandlerContext ctx, Object msg, 
                                  ChannelPromise promise) {
                    System.out.println(5);
                    ctx.write(msg, promise); // 5
                }
            });
            ch.pipeline().addLast(new ChannelOutboundHandlerAdapter(){
                @Override
                public void write(ChannelHandlerContext ctx, Object msg, 
                                  ChannelPromise promise) {
                    System.out.println(6);
                    ctx.write(msg, promise); // 6
                }
            });
        }
    })
    .bind(8080);

客户端：

new Bootstrap()
    .group(new NioEventLoopGroup())
    .channel(NioSocketChannel.class)
    .handler(new ChannelInitializer<Channel>() {
        @Override
        protected void initChannel(Channel ch) {
            ch.pipeline().addLast(new StringEncoder());
        }
    })
    .connect("127.0.0.1", 8080)
    .addListener((ChannelFutureListener) future -> {
        future.channel().writeAndFlush("hello,world");
    });

服务端打印结果：

1
2
3
6
5
4

可以看到，ChannelInboundHandlerAdapter（入站）是按照 addLast 的顺序执行的，而 ChannelOutboundHandlerAdapter （出站）是按照 addLast 的逆序执行的，ChannelPipeline 的实现是一个 ChannelHandlerContext（包装了 ChannelHandler） 组成的双向链表

• 入站处理器中，ctx.fireChannelRead(msg) 是 调用下一个入站处理器,也可以调用 super.channelRead(ctx, msg);

  • 如果注释掉 1 处代码，则仅会打印 1
  • 如果注释掉 2 处代码，则仅会打印 1 2

• 3处 ctx.channel().write(msg) 会从尾部开始触发后续出站处理器的执行

  • 如果注释掉 3 处代码，则仅会打印 1 2 3

• 类似的，出站处理器中，ctx.write(msg, promise) 的调用也会 触发上一个出站处理器

  • 如果注释掉 6 处代码，则仅会打印 1 2 3 6

• ctx.channel().write(msg) vs ctx.write(msg)

  • 都是触发出站处理器的执行
  • ctx.channel().write(msg) 从尾部开始查找出站处理器
  • ctx.write(msg) 是从当前节点找上一个出站处理器
  • 3 处的 ctx.channel().write(msg) 如果改为 ctx.write(msg) 仅会打印 1 2 3，因为节点3 之前没有其它出站处理器了
  • 6 处的 ctx.write(msg, promise) 如果改为 ctx.channel().write(msg) 会打印 1 2 3 6 6 6… 因为 ctx.channel().write() 是从尾部开始查找，结果又是节点6 自己

服务端流程图

ByteBuf

ByteBuf 是对字节数据的封装

创建

ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(10);
//上面代码创建了一个默认的 ByteBuf（池化基于直接内存的 ByteBuf），初始容量是 10
log(buffer);
//read index:0 write index:0 capacity:10

private static void log(ByteBuf buffer) {
    int length = buffer.readableBytes();
    int rows = length / 16 + (length % 15 == 0 ? 0 : 1) + 4;
    StringBuilder buf = new StringBuilder(rows * 80 * 2)
        .append("read index:").append(buffer.readerIndex())
        .append(" write index:").append(buffer.writerIndex())
        .append(" capacity:").append(buffer.capacity())
        .append(NEWLINE);
    appendPrettyHexDump(buf, buffer);
    System.out.println(buf.toString());
}

直接内存 VS 堆内存

ByteBuf 是存储是分直接内存和堆内存的

创建堆内存 ByteBuf

ByteBuf buffer = ByteBufAllocator.DEFAULT.heapBuffer(10);

创建直接内存 ByteBuf

ByteBuf buffer = ByteBufAllocator.DEFAULT.directBuffer(10);

• 直接内存创建和销毁的代价昂贵，但读写性能高（少一次内存复制），适合配合池化功能一起用
• 直接内存对 GC 压力小，因为这部分内存不受 JVM 垃圾回收的管理，但也要注意及时主动释放

池化 VS 非池化

池化的最大意义在于可以重用 ByteBuf，优点：

• 没有池化，则每次都得创建新的 ByteBuf 实例，这个操作对直接内存代价昂贵，就算是堆内存，也会增加 GC 压力
• 有了池化，则可以重用池中 ByteBuf 实例，并且采用了与 jemalloc 类似的内存分配算法提升分配效率
• 高并发时，池化功能更节约内存，减少内存溢出的可能

池化功能默认开启，也可以通过环境变量设置

-Dio.netty.allocator.type={unpooled|pooled}

• 4.1 以后，非 Android 平台默认启用池化实现，Android 平台启用非池化实现
• 4.1 之前，池化功能还不成熟，默认是非池化实现

组成

ByteBuf 由四部分组成

最开始的时候，读写指针都在 0 的位置

写入

常用方法：

方法签名	含义	备注
writeBoolean(boolean value)	写入 boolean 值	用一字节 01|00 代表 true|false
writeByte(int value)	写入 byte 值
writeShort(int value)	写入 short 值
writeInt(int value)	写入 int 值	Big Endian，即 0x250，写入后 00 00 02 50
writeIntLE(int value)	写入 int 值	Little Endian，即 0x250，写入后 50 02 00 00
writeLong(long value)	写入 long 值
writeChar(int value)	写入 char 值
writeFloat(float value)	写入 float 值
writeDouble(double value)	写入 double 值
writeBytes(ByteBuf src)	写入 netty 的 ByteBuf
writeBytes(byte[] src)	写入 byte[]
writeBytes(ByteBuffer src)	写入 nio 的 ByteBuffer
int writeCharSequence(CharSequence sequence, Charset charset)	写入字符串

注意

• 这些方法的未指明返回值的，其返回值都是 ByteBuf，意味着可以链式调用
• 网络传输，默认习惯是 Big Endian

先写入 4 个字节

buffer.writeBytes(new byte[]{1, 2, 3, 4});
log(buffer);

结果

read index:0 write index:4 capacity:10
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04                                     |....            |
+--------+-------------------------------------------------+----------------+

再写入一个 int 整数，也是 4 个字节

buffer.writeInt(5);
log(buffer);

结果

read index:0 write index:8 capacity:10
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 00 00 00 05                         |........        |
+--------+-------------------------------------------------+----------------+

还有一类方法是 set 开头的一系列方法，也可以写入数据，但不会改变写指针位置

扩容

在写入一个 int 整数时，容量不够了（初始容量为 10），这时候会引发扩容

buffer.writeInt(6);
log(buffer);

扩容规则：

• 如果写入后数据未超过512，则选择下一个 16 的整数倍，例如写入后大小为 12，则扩容后 capacity 为 16
• 如果写入后数据超过512，则选择下一个 2^n，例如写入后大小为 513，则扩容为 2^10=1024（2^9=512 已经不够了）
• 扩容不能超过 max capacity 会报错

结果

read index:0 write index:12 capacity:16
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 00 00 00 05 00 00 00 06             |............    |
+--------+-------------------------------------------------+----------------+

读取

例如读取 4 次，每次一个字节

System.out.println(buffer.readByte());
System.out.println(buffer.readByte());
System.out.println(buffer.readByte());
System.out.println(buffer.readByte());
log(buffer);

读过的内容，就属于废弃部分了，再读只能读取剩余未读取的部分

1
2
3
4
read index:4 write index:12 capacity:16
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 00 00 05 00 00 00 06                         |........        |
+--------+-------------------------------------------------+----------------+

如果需要重复读取 int 整数 5，怎么办？

可以在 read 前先做个标记 mark

buffer.markReaderIndex();
System.out.println(buffer.readInt());
log(buffer);

结果

5
read index:8 write index:12 capacity:16
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 00 00 06                                     |....            |
+--------+-------------------------------------------------+----------------+

这时要重复读取的话，重置到标记位置 reset

buffer.resetReaderIndex();
log(buffer);

结果

read index:4 write index:12 capacity:16
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 00 00 05 00 00 00 06                         |........        |
+--------+-------------------------------------------------+----------------+

还有种办法是采用 get 开头的一系列方法，这些方法不会改变 read index

retain & release

内存释放，由于 Netty 中有堆外内存的 ByteBuf 实现，堆外内存最好是手动释放，而不是等待 GC 回收

• UnpooledHeapByteBuf 使用 JVM 内存，只需要等待 GC 回收即可
• UnpooledDirectByteBuf 使用的就是直接内存，需要特殊方法回收
• PooledByteBuf 和它的子类使用了池化机制，需要更复杂的方式回收

Netty 采用了 引用计数法 来控制回收内存，每个 ByteBuf 都实现了 ReferenceCounted 接口

• 每个 ByteBuf 对象初始计数为 1
• 调用 release 方法计数减 1，如果计数为 0，ByteBuf 内存被回收
• 调用 retain 方法计数加 1，表示调用者没调用完之前，其他 handler 即使调用了 release 也不会造成回收
• 当计数为 0 时，底层内存会被回收，即使 ByteBuf 对象还在，也将无法正常使用

---

谁来负责调用 release 呢？

不是我们想象的（一般情况下）

ByteBuf buf = ...
try {
    ...
} finally {
    buf.release();
}

因为 pipeline 的存在，一般需要将 ByteBuf 传递给下一个 ChannelHandler，如果在 finally 中 release 了，就失去了传递性（当然，如果在这个 ChannelHandler 内这个 ByteBuf 已完成了它的使命，那么便无须再传递）

基本规则是，谁是最后使用者，谁负责 release，如下

• 起点，对于 NIO 实现来讲，在 io.netty.channel.nio.AbstractNioByteChannel.NioByteUnsafe#read 方法中首次创建 ByteBuf 放入 pipeline（line 163 pipeline.fireChannelRead(byteBuf)）
• 入站 ByteBuf 处理原则
  • 对原始 ByteBuf 不做处理，调用 ctx.fireChannelRead(msg) 向后传递，这时无须 release
  • 将原始 ByteBuf 转换为其它类型的 Java 对象，这时 ByteBuf 就没用了，必须 release
  • 如果不调用 ctx.fireChannelRead(msg) 向后传递，那么也必须 release
  • 注意各种异常，如果 ByteBuf 没有成功传递到下一个 ChannelHandler，必须 release
  • 假设消息一直向后传，那么 TailContext 会负责释放未处理消息（原始的 ByteBuf）
• 出站 ByteBuf 处理原则
  • 出站消息最终都会转为 ByteBuf 输出，一直向前传，由 HeadContext flush 后 release
• 异常处理原则
  • 有时候不清楚 ByteBuf 被引用了多少次，但又必须彻底释放，可以循环调用 release 直到返回 true

ByteBuf优势

• 池化 - 可以重用池中 ByteBuf 实例，更节约内存，减少内存溢出的可能
• 读写指针分离，不需要像 ByteBuffer 一样切换读写模式
• 可以自动扩容
• 支持链式调用，使用更流畅
• 很多地方体现零拷贝，例如 slice、duplicate、CompositeByteBuf