Netty从入门到进阶（二）

二、Netty入门

1. 概述

1.1 Netty是什么

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

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

1.2 Netty的作者

他还是另一个著名网络应用框架Mina的重要贡献者。

1.3 Netty的地位

Netty在Java网络应用框架中的地位就好比：Spring框架在JavaEE开发中的地位

以下框架都使用了Netty，因为它们有网络通信需求：

Cassandra - nosql 数据库
Spark - 大数据分布式计算框架
Hadoop - 大数据分布式存储框架
RocketMQ - ali 开源的消息队列
ElasticSearch - 搜索引擎
gRPC - rpc 框架
Dubbo - rpc 框架
Spring 5.x - flux api 完全抛弃了 tomcat ，使用 netty 作为服务器端
Zookeeper - 分布式协调框架

1.4 Netty的优势

Netty vs NIO，工作量大，bug多

NIO 需要自己构建协议
Netty 解决了TCP传输问题，如粘包、半包
NIO epoll空轮询导致 CPU 100%
Netty 对API进行增强，使之更易用，如FastThreadLocal => ThreadLocal, ByteBuf => ByteBuffer

Netty vs 其它网络应用框架

Mina由apache维护，将来3.x版本可能会有较大重构，破坏API向下兼容性，而Netty的开发迭代更迅速，API更简洁、文档更优秀

久经考验，16年，Netty版本

2.x 2004
3.x 2008
4.x 2013
5.x 已废弃（没有明显的性能提升，维护成本高）

2. Hello World

2.1 目标

开发一个简单的服务器端和客户端

客户端向服务器端发送hello, world
服务器仅接收，不返回

2.2 服务器端

package cn.itcast.netty.c2;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.logging.LoggingHandler;

public class HelloServer {
    public static void main(String[] args) {
        // 1. 启动器，负责组装 netty 组件，启动服务器
        new ServerBootstrap()
                // 2. BossEventLoop, WorkerEventLoop(selector,thread), group 组
                .group(new NioEventLoopGroup())
                // 3. 选择 服务器的 ServerSocketChannel 实现
                .channel(NioServerSocketChannel.class) // OIO BIO
                // 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 LoggingHandler());
                                ch.pipeline().addLast(new StringDecoder()); // 5. 将 ByteBuf 转换为字符串
                                ch.pipeline().addLast(new ChannelInboundHandlerAdapter() { // 6. 自定义 handler
                                    @Override // 读事件
                                    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                        System.out.println(msg); // 打印上一步转换好的字符串
                                    }
                                });
                            }
                        })
                // 4. 绑定监听端口
                .bind(8080);
    }
}

代码解读

1处，创建NioEventLoopGroup，可以简单理解为线程池 + selector
2处，选择服务Socket实现类，其中有NioServerSocketChannel表示基于NIO的服务器端实现，其它实现还有

3处，为啥方法叫childHandler，是接下来添加的处理器都是给SocketChannel用的，而不是给ServerSocketChannel。ChannelInitializer处理器（仅执行一次），它的作用是待客户端SocketChannel建立连接后，执行initChannel以便添加更多的处理器
4处，ServerSocketChannel绑定的监听端口
5处，SocketChannel的处理器，解码 ByteBuf => String
6处，SocketChannel的业务处理器，使用上一个处理器的处理结果

输出

15:48:34 [DEBUG] [nioEventLoopGroup-2-2] i.n.h.l.LoggingHandler - [id: 0x30203006, L:/127.0.0.1:8080 - R:/127.0.0.1:49555] REGISTERED
15:48:34 [DEBUG] [nioEventLoopGroup-2-2] i.n.h.l.LoggingHandler - [id: 0x30203006, L:/127.0.0.1:8080 - R:/127.0.0.1:49555] ACTIVE
15:48:34 [DEBUG] [nioEventLoopGroup-2-2] i.n.h.l.LoggingHandler - [id: 0x30203006, L:/127.0.0.1:8080 - R:/127.0.0.1:49555] READ: 12B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 68 65 6c 6c 6f 2c 20 77 6f 72 6c 64 |hello, world |
+--------+-------------------------------------------------+----------------+
hello, world
15:48:34 [DEBUG] [nioEventLoopGroup-2-2] i.n.h.l.LoggingHandler - [id: 0x30203006, L:/127.0.0.1:8080 - R:/127.0.0.1:49555] READ COMPLETE

2.3 客户端

package cn.itcast.netty.c2;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;

import java.net.InetSocketAddress;

public class HelloClient {
    public static void main(String[] args) throws InterruptedException {
        // 启动类
        new Bootstrap()
            // 1. 添加 EventLoop
            .group(new NioEventLoopGroup())
            // 2. 选择客户端 channel 实现
            .channel(NioSocketChannel.class)
            // 3. 添加处理器
            .handler(new ChannelInitializer<NioSocketChannel>() {
                @Override // 在连接建立后被调用
                protected void initChannel(NioSocketChannel ch) throws Exception {
                    ch.pipeline().addLast(new StringEncoder());  // 8. 将 字符串 转换为 ByteBuf
                }
            })
            // 4. 连接到服务器
            .connect(new InetSocketAddress("localhost", 8080))
            .sync()  // 5. 阻塞方法，直到连接建立
            .channel()  // 6. 代表客户端与服务器端的连接对象
            // 7. 向服务器发送数据
            .writeAndFlush("hello, world");  // 然后调用StringEncoder()方法，把字符串转为ByteBuf
    }
}

代码解读

1处，创建NioEventLoopGroup，同Server
2处，选择客户Socket实现类，NioSocketChannel表示基于NIO的客户端实现，其它还有

3处，添加SocketChannel的处理器，ChannelInitializer处理器（仅执行一次），它的作用是待客户端SocketChannel建立连接后，执行initChannel以便添加更多的处理器
4处，指定要连接的服务器和端口
5处，Netty中很多方法都是异步的，如connect，这时需要使用sync方法等待connect建立连接完毕
6处，获取channel对象，它即为通道抽象，可以进行数据读写操作
7处，写入消息并清空缓冲区
8处，消息会经过通道handler处理，这里是将 String => ByteBuf 发出
数据经过网络传输，到达服务器端，服务器端5和6处的handler都被触发，走完一个流程

2.4 流程梳理

💡提示

一开始需要树立正确的观念

把channel理解为数据的通道
把msg理解为流动的数据，最开始输入是ByteBuf，但经过pipeline的加工，会变成其它类型对象，最后输出又变成ByteBuf
把handler理解为数据的处理工序
- 工序有多道，合在一起就是pipeline，pipeline负责发布事件（读、读取完成...）传播给每个handler，handler对自己感兴趣的事件进行处理（重写了相应事件处理方法）
- handler份Inbound（入站）和 Outbound（出站）两类
把eventLoop理解为处理数据的工人
- 工人可以管理多个channel的io操作，并且一旦工人负责了某个channel，就要负责到底（绑定）
- 工人既可以执行io操作，也可以进行任务处理，每位工人有任务队列，队列力可以堆放多个channel的待处理任务，任务分为普通任务、定时任务
- 工人按照pipeline顺序，依次按照handler的规划（代码）处理数据，可以为每道工序指定不同的工人。

3. 组件

3.1 EventLoop

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

它的继承关系比较复杂：

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

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

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

示例代码：

package cn.itcast.netty.c2;

import io.netty.channel.DefaultEventLoop;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.TimeUnit;

@Slf4j
public class TestEventLoop {

    public static void main(String[] args) {
        // 1. 创建事件循环组
        EventLoopGroup group = new NioEventLoopGroup(2);  // IO事件、普通任务、定时任务
//        EventLoopGroup group2 = new DefaultEventLoop();  // 普通任务、定时任务

        // 2. 获取下一个事件循环对象
        System.out.println(group.next());
        System.out.println(group.next());
        System.out.println(group.next());

        // 3. 执行普通任务（可以用来执行耗时较长的任务）
        group.next().submit(() -> {
            try {
                Thread.sleep(1);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            log.debug("ok");
        });

        // 4. 执行定时任务
        group.next().scheduleAtFixedRate(() -> {
            log.debug("定时任务!");
        }, 0, 1, TimeUnit.SECONDS);

        log.debug("main");
    }
}

输出

io.netty.channel.nio.NioEventLoop@553f17c
io.netty.channel.nio.NioEventLoop@4f7d0008
io.netty.channel.nio.NioEventLoop@553f17c
17:12:20 [DEBUG] [nioEventLoopGroup-2-2] c.i.n.c.TestEventLoop - 定时任务!
17:12:20 [DEBUG] [nioEventLoopGroup-2-1] c.i.n.c.TestEventLoop - ok
17:12:20 [DEBUG] [main] c.i.n.c.TestEventLoop - main
17:12:21 [DEBUG] [nioEventLoopGroup-2-2] c.i.n.c.TestEventLoop - 定时任务!
17:12:22 [DEBUG] [nioEventLoopGroup-2-2] c.i.n.c.TestEventLoop - 定时任务!
17:12:23 [DEBUG] [nioEventLoopGroup-2-2] c.i.n.c.TestEventLoop - 定时任务!

💡优雅关闭

优雅关闭shutdownGracefully方法。该方法会首先切换EventLoopGroup到关闭状态从而拒绝新的任务的加入，然后在任务队列的任务都处理完成后，停止线程的运行。从而确保整体应用是在正常有序的状态下退出。

演示NioEventLoop处理IO事件

服务器端两个 nio worker工人：

package cn.itcast.netty.c2;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.Charset;

@Slf4j
public class EventLoopServer {
    public static void main(String[] args) {
        new ServerBootstrap()
                // boss 和 worker
                // boss只负责SeverSocketChannel上的accept事件，worker只负责socketChannel上的读写操作
                .group(new NioEventLoopGroup(), new NioEventLoopGroup(2))
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                ByteBuf buf = (ByteBuf) msg;
                                log.debug(buf.toString(Charset.defaultCharset()));
                            }
                        });
                    }
                })
                .bind(8080);
    }
}

客户端：

package cn.itcast.netty.c2;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;

import java.net.InetSocketAddress;

public class EventLoopClient {
    public static void main(String[] args) throws InterruptedException {
        // 1. 启动类
        Channel channel = new Bootstrap()
                // 2. 添加 EventLoop
                .group(new NioEventLoopGroup())
                // 3. 选择客户端 channel 实现
                .channel(NioSocketChannel.class)
                // 4. 添加处理器
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override // 在连接建立后被调用
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new StringEncoder());  // 将 字符串 转换为 ByteBuf
                    }
                })
                // 5. 连接到服务器
                .connect(new InetSocketAddress("localhost", 8080))
                .sync()  // 阻塞方法，直到连接建立
                .channel();  // 代表客户端与服务器端的连接对象
    }
}

两个工人轮流处理channel，但工人与channel之间进行了绑定

增加两个非 nio 工人：

package cn.itcast.netty.c2;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.Charset;

@Slf4j
public class EventLoopServer {
    public static void main(String[] args) {
        // 细分2：创建一个独立的EventLoopGroup（增加两个非nio工人）
        EventLoopGroup group = new DefaultEventLoop();

        new ServerBootstrap()
                // boss 和 worker
                // 细分1：boss只负责SeverSocketChannel上的accept事件，worker只负责socketChannel上的读写操作
                .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(group, "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);
    }
}

可以看到nio工人和非nio工人也分别绑定了channel，而我们自己的handler由非nio工人执行。

💡handler执行中如何换人？

关键代码：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()) {  // 当前handler中的线程，是否和eventLoop是同一个线程
        next.invokeChannelRead(m);  // 直接调用
    } 
    // 不是，将要执行的代码作为任务提交给下一个事件循环处理（换人）
    else {
        executor.execute(new Runnable() {  // 下一个handler线程
            @Override
            public void run() {
                next.invokeChannelRead(m);
            }
        });
    }

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

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

3.2 Channel

channel的主要作用：

close()可以用来关闭channel
closeFuture()用来处理channel的关闭
- sync方法作用是同步等待channel关闭
- 而addListener方法是异步等待channel关闭
pipeline()方法添加处理器
write()方法将数据写入（要调用flush方法才会从缓冲区发出）
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!");

1处返回的是ChannelFuture对象，它的作用是利用channel()方法来获取Channel对象

注意：connect方法是异步的，意味着不等连接建立，方法执行就返回了。因此ChannelFuture对象中不能【立刻】获得到正确的Channel对象。

实验如下：

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
channelFuture.sync(); // 2
System.out.println(channelFuture.channel()); // 3

执行到1时，连接未建立，打印：[id: 0x2e1884dd]
执行到2时，sync方法是同步等待连接建立完成
执行到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
channelFuture.addListener((ChannelFutureListener) future -> {
    System.out.println(future.channel()); // 2
});

执行到1时，连接未建立，打印：[id: 0x749124ba]
ChannelFutureListener会在连接建立时被调用（其中operationComplete方法），因此执行到2时，连接肯定建立了，打印：[id: 0x749124ba, L:/127.0.0.1:57351 - R:/127.0.0.1:8080]

package cn.itcast.netty.c2;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.Date;

@Slf4j
public class EventLoopClient {
    public static void main(String[] args) throws InterruptedException {
        // 带有Future，Promise的类型都是和异步方法配套使用，用来处理结果
        ChannelFuture channelFuture = new Bootstrap()
                .group(new NioEventLoopGroup())
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new StringEncoder());  // 将 字符串 转换为 ByteBuf
                    }
                })
                // 1. 连接到服务器
                // 异步非阻塞, main发起了调用，真正执行connect连接的是 nio 线程
                .connect(new InetSocketAddress("localhost", 8080));

        // 2.1 使用sync方法同步处理结果
        /*channelFuture
                .sync()  // 阻塞当前线程，直到nio线程连接建立完毕
                .channel()
                .writeAndFlush(new Date() + ": hello world!");*/

        // 2.2 使用addListener(回调对象)方法异步处理结果
        channelFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                // 在nio线程连接建立好之后，会调用operationComplete
                Channel channel = future.channel();
                log.debug("{}", channel);
                channel.writeAndFlush(new Date() + "hello, world!");
            }
        });

    }
}

CloseFuture

package cn.itcast.netty.c2;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.Scanner;

@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();  // 异步操作
//                    log.debug("处理关闭之后的操作");  // 不能在这里善后
                    break;
                }
                channel.writeAndFlush(line);
            }
        }, "input").start();

        // 获取CloseFuture对象， 1）同步处理关闭 2）异步处理关闭
        ChannelFuture closeFuture = channel.closeFuture();

        // 1）同步处理关闭
        /*log.debug("waiting close...");
        closeFuture.sync();  // 阻塞
        log.debug("处理关闭之后的操作");*/

        // 2) 异步处理关闭
        closeFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                log.debug("处理关闭之后的操作");  // nio线程调用
                group.shutdownGracefully();  // 优雅停止
            }
        });
    }
}

💡异步提升的是什么

有些同学看到这里会有疑问：为什么不在一个线程中去执行建立连接、去执行关闭 channel，那样不是也可以吗？非要用这么复杂的异步方式：比如一个线程发起建立连接，另一个线程去真正建立连接
还有同学会笼统地回答，因为 netty 异步方式用了多线程、多线程就效率高。其实这些认识都比较片面，多线程和异步所提升的效率并不是所认为的

思考下面的场景：4个医生给人看病，每个病人花费20分钟，而且医生看病的过程中是以病人为单位的，一个病人看完了，才能看下一个病人。假设病人源源不断地来，可以计算一下4个医生一天工作8小时，处理的病人总数是 4 * 8 * 3 = 96。

经研究发现，看病可以细分为四个步骤，经拆分后每个步骤需要5分钟，如下：

因此，可以做如下优化，只有一开始，医生2、3、4分别需要等待5、10、15分钟才能执行工作，但只要后续病人源源不断地来，他们就能够满负荷工作，并且处理病人的能力提高到了4 * 8 * 12（20分钟一个 -> 5分钟一个），效率几乎是原来的4倍。

要点

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

3.3 Future & Promise

在异步处理时，经常用到这两个接口。首先要说明netty中的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
addListener		添加回调，异步接收结果
setSuccess			设置成功结果
setFailure			设置失败结果

例1：jdk中Future

package cn.itcast.netty.c2;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.*;

@Slf4j
public class TestJdkFuture {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1. 线程池
        ExecutorService service = Executors.newFixedThreadPool(2);

        // 2. 提交任务
        Future<Integer> future = service.submit(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                log.debug("执行计算...");
                Thread.sleep(1000);
                return 50;
            }
        });

        // 3. 主线程通过future获取结果
        log.debug("等待结果...");
        log.debug("结果是: {}", future.get());  // 阻塞等待
    }
}

输出

09:28:35 [DEBUG] [main] c.i.n.c.TestJdkFuture - 等待结果...
09:28:35 [DEBUG] [pool-1-thread-1] c.i.n.c.TestJdkFuture - 执行计算...
09:28:36 [DEBUG] [main] c.i.n.c.TestJdkFuture - 结果是: 50

例2：Netty中的Future

package cn.itcast.netty.c2;

import io.netty.channel.EventLoop;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ExecutionException;

@Slf4j
public class TestNettyFuture {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        NioEventLoopGroup group = new NioEventLoopGroup();
        EventLoop eventLoop = group.next();

        Future<Integer> future = eventLoop.submit(() -> {
            log.debug("执行计算...");
            Thread.sleep(1000);
            return 70;
        });

        log.debug("等待结果...");
        log.debug("getNow(), 结果是: {}", future.getNow());  // 非阻塞
        log.debug("结果是: {}", future.get());  // 阻塞

        future.addListener(new GenericFutureListener<Future<? super Integer>>() {
            @Override
            public void operationComplete(Future<? super Integer> future) throws Exception {
                log.debug("接收结果: {}", future.getNow());
            }
        });
    }
}

输出

09:35:47 [DEBUG] [main] c.i.n.c.TestNettyFuture - 等待结果...
09:35:47 [DEBUG] [nioEventLoopGroup-2-1] c.i.n.c.TestNettyFuture - 执行计算...
09:35:47 [DEBUG] [main] c.i.n.c.TestNettyFuture - getNow(), 结果是: null
09:35:48 [DEBUG] [main] c.i.n.c.TestNettyFuture - 结果是: 70
09:35:48 [DEBUG] [nioEventLoopGroup-2-1] c.i.n.c.TestNettyFuture - 接收结果: 70

例3：Netty中的Promise

package cn.itcast.netty.c2;

import io.netty.channel.EventLoop;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.DefaultPromise;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ExecutionException;

@Slf4j
public class TestNettyPromise {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1. 创建EventLoop对象
        EventLoop eventLoop = new NioEventLoopGroup().next();

        // 2. 主动创建Promise，结果容器
        DefaultPromise<Integer> promise = new DefaultPromise<>(eventLoop);

        new Thread(() -> {
            // 3. 任意一个线程执行计算，计算完毕后向promise填充结果
            log.debug("开始计算...");

            try {
                int i = 1 / 0;
                Thread.sleep(1000);
                promise.setSuccess(80);
            } catch (InterruptedException e) {
                e.printStackTrace();
                promise.setFailure(e);
            }
        }).start();

        // 4. 接收结果
        log.debug("等待结果...");
        log.debug("结果是: {}", promise.get());
    }
}

输出

09:43:06 [DEBUG] [Thread-0] c.i.n.c.TestNettyPromise - 开始计算...
09:43:06 [DEBUG] [main] c.i.n.c.TestNettyPromise - 等待结果...
Exception in thread "Thread-0" java.lang.ArithmeticException: / by zero

3.4 Handler & Pipeline

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

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

打个比喻，每个Channel是一个产品的加工车间，Pipeline是车间中的流水线，ChannelHandler就是流水线上的各道工序，而ByteBuf就是原材料，经过很多工序的加工：先经过一道道入站工序，再经过一道道出站工序最终变成产品。

服务器端：

package cn.itcast.netty.c2;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class TestPipeline {
    public static void main(String[] args) {
        new ServerBootstrap()
                .group(new NioEventLoopGroup())
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        // 1. 通过channel拿到pipeline
                        ChannelPipeline pipeline = ch.pipeline();
                        // 2. 添加处理器 head ⇄ h1 ⇄ h2 ⇄ h3 ⇄ h4 ⇄ h5 ⇄ h6 ⇄ tail
                        pipeline.addLast("h1", new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                log.debug("1");
                                super.channelRead(ctx, msg);  // 1
                                // ctx.fireChannelRead(msg);
                            }
                        });
                        pipeline.addLast("h2", new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                log.debug("2");
                                super.channelRead(ctx, msg);  // 2
                            }
                        });
                        pipeline.addLast("h3", new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                log.debug("3");
                                super.channelRead(ctx, msg);  
                                ch.writeAndFlush(ctx.alloc().buffer().writeBytes("server...".getBytes()));  // 3
                            }
                        });

                        pipeline.addLast("h4", new ChannelOutboundHandlerAdapter() {
                            @Override
                            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                                log.debug("4");
                                super.write(ctx, msg, promise);  // 4
                            }
                        });
                        pipeline.addLast("h5", new ChannelOutboundHandlerAdapter() {
                            @Override
                            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                                log.debug("5");
                                super.write(ctx, msg, promise);  // 5
                            }
                        });
                        pipeline.addLast("h6", new ChannelOutboundHandlerAdapter() {
                            @Override
                            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                                log.debug("6");
                                super.write(ctx, msg, promise);  // 6
                            }
                        });
                    }
                })
                .bind(8080);
    }
}

客户端：

package cn.itcast.netty.c2;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.Scanner;

@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();  // 异步操作
//                    log.debug("处理关闭之后的操作");  // 不能在这里善后
                    break;
                }
                channel.writeAndFlush(line);
            }
        }, "input").start();

        // 获取CloseFuture对象， 1）同步处理关闭 2）异步处理关闭
        ChannelFuture closeFuture = channel.closeFuture();

        // 1）同步处理关闭
        /*log.debug("waiting close...");
        closeFuture.sync();  // 阻塞
        log.debug("处理关闭之后的操作");*/

        // 2) 异步处理关闭
        closeFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                log.debug("处理关闭之后的操作");  // nio线程调用
                group.shutdownGracefully();  // 优雅停止
            }
        });
    }
}

客户端输出：

aaa
10:15:55 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x9576c7dc, L:/127.0.0.1:51165 - R:localhost/127.0.0.1:8080] WRITE: 3B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 61 61 |aaa |
+--------+-------------------------------------------------+----------------+
10:15:55 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x9576c7dc, L:/127.0.0.1:51165 - R:localhost/127.0.0.1:8080] FLUSH
10:15:55 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x9576c7dc, L:/127.0.0.1:51165 - R:localhost/127.0.0.1:8080] READ: 9B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 73 65 72 76 65 72 2e 2e 2e |server... |
+--------+-------------------------------------------------+----------------+
10:15:55 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x9576c7dc, L:/127.0.0.1:51165 - R:localhost/127.0.0.1:8080] READ COMPLETE

服务器端输出：先进后出

10:15:55 [DEBUG] [nioEventLoopGroup-2-2] c.i.n.c.TestPipeline - 1
10:15:55 [DEBUG] [nioEventLoopGroup-2-2] c.i.n.c.TestPipeline - 2
10:15:55 [DEBUG] [nioEventLoopGroup-2-2] c.i.n.c.TestPipeline - 3
10:15:55 [DEBUG] [nioEventLoopGroup-2-2] c.i.n.c.TestPipeline - 6
10:15:55 [DEBUG] [nioEventLoopGroup-2-2] c.i.n.c.TestPipeline - 5
10:15:55 [DEBUG] [nioEventLoopGroup-2-2] c.i.n.c.TestPipeline - 4

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

入站处理器中，ctx.fireChannelRead(msg)是调用下一个入站处理器
- 如果注释掉1处代码，则仅会打印 1
- 如果注释掉2处代码，则仅会打印 1 2
3处的ctx.writeAndFlush(ctx.alloc().buffer().writeBytes("server...".getBytes()))会从尾部开始触发后续出站处理器的执行
- 如果注释掉3处代码，则仅会打印 1 2 3
类似的，出站处理器中，super.write(ctx, msg, promise)的调用会触发上一个出站处理器
- 如果注释掉6处代码，则仅会打印 1 2 3 6
ctx.channel().write(msg) vs ctx.write(msg)
- 都是触发出站处理器的执行
- ctx.channel().write(msg)从尾部开始查找出站处理器
- ctx.write(msg)是从当前节点找上一个出站处理器
- 3处的cxt.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()是从尾部tail开始查找，结果又是节点6自己

借助EmbeddedChannel进行测试

package cn.itcast.netty.c2;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelOutboundHandlerAdapter;
import io.netty.channel.ChannelPromise;
import io.netty.channel.embedded.EmbeddedChannel;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class TestEmbeddedChannel {
    public static void main(String[] args) {
        ChannelInboundHandlerAdapter h1 = new ChannelInboundHandlerAdapter() {
            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                log.debug("1");
                super.channelRead(ctx, msg);
            }
        };
        ChannelInboundHandlerAdapter h2 = new ChannelInboundHandlerAdapter() {
            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                log.debug("2");
                super.channelRead(ctx, msg);
            }
        };

        ChannelOutboundHandlerAdapter h3 = new ChannelOutboundHandlerAdapter() {
            @Override
            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                log.debug("3");
                super.write(ctx, msg, promise);
            }
        };
        ChannelOutboundHandlerAdapter h4 = new ChannelOutboundHandlerAdapter() {
            @Override
            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                log.debug("4");
                super.write(ctx, msg, promise);
            }
        };

        EmbeddedChannel channel = new EmbeddedChannel(h1, h2, h3, h4);
        // 模拟入站操作
        channel.writeInbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("hello".getBytes()));  // 1 2
        // 模拟出站操作
        channel.writeOutbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("world".getBytes()));  // 4 3
    }
}

输出

10:54:04 [DEBUG] [main] c.i.n.c.TestEmbeddedChannel - 1
10:54:04 [DEBUG] [main] c.i.n.c.TestEmbeddedChannel - 2
10:54:04 [DEBUG] [main] c.i.n.c.TestEmbeddedChannel - 4
10:54:04 [DEBUG] [main] c.i.n.c.TestEmbeddedChannel - 3

3.5 ByteBuf

ByteBuf是对字节数据的封装

1）创建

package cn.itcast.netty.c2;

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

import static io.netty.buffer.ByteBufUtil.appendPrettyHexDump;
import static io.netty.util.internal.StringUtil.NEWLINE;

public class TestByteBuf {

    public static void main(String[] args) {
        ByteBuf buf = ByteBufAllocator.DEFAULT.buffer();  // 可以动态扩容 基于直接内存的ByteBuf
//        System.out.println(buf);  // PooledUnsafeDirectByteBuf(ridx: 0, widx: 0, cap: 256)
        log(buf);

        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < 300; i++) {
            sb.append("a");
        }

        buf.writeBytes(sb.toString().getBytes());
//        System.out.println(buf);  // PooledUnsafeDirectByteBuf(ridx: 0, widx: 300, cap: 512)
        log(buf);
    }

    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());
    }
}

输出

read index:0 write index:0 capacity:256

read index:0 write index:300 capacity:512
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000010| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000020| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000030| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000040| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000050| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000060| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000070| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000080| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000090| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|000000a0| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|000000b0| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|000000c0| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|000000d0| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|000000e0| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|000000f0| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000100| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000110| 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaaaaaa|
|00000120| 61 61 61 61 61 61 61 61 61 61 61 61 |aaaaaaaaaaaa |
+--------+-------------------------------------------------+----------------+

2）直接内存 vs 堆内存

堆内存的分配效率较高，但读写效率较低，且受到垃圾回收影响；而直接内存分配效率较低，读写效率较高（零拷贝）。

可以使用下面的代码来创建池化基于堆的ByteBuf：

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

也可以使用下面的代码来创建池化基于直接内存的bytebuf：

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

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

3）池化 vs 非池化

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

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

池化功能是否开启，可以通过下面的系统环境变量或虚拟机参数来设置：

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

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

4）组成

ByteBuf由四部分组成：

最开始读写指针都在0位置；
废弃字节是指已经读取过的部分；
最大容量为Integet.MAX_VALUE（即2^31 -1，约2GB)，底层限制：
- 堆内存：受JVM堆大小限制（可通过-Xmx调整）；
- 直接内存：受系统内存和-XX:MaxDirectMemorySize限制

5）写入

方法签名	含义	备注
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开头的一系列方法，也可以写入数据，但不会改变写指针位置。

6）扩容

这时再写入一个int整数时，容量不够了（初始容量是10），这时会扩容

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

扩容规则是：

如果写入后数据大小未超过512，则选择下一个16的整数倍，例如写入后大小为12，则扩容后capacity是16；
如果写入后数据大小超过512，则选择下一个2^n，例如写入后大小为513，则扩容后capacity是2^10=1024（z^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 |............ |
+--------+-------------------------------------------------+----------------+

7）读取

例如读了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。

8）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

9）slice

【零拷贝】的体现之一，对原始ByteBuf进行切片成多个ByteBuf，切片后的ByteBuf并没有发生内存复制，还是使用原始ByteBuf的内存，切片后的ByteBuf维护独立的read和write指针。

示例1：

package cn.itcast.netty.c2;

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

import static cn.itcast.netty.c2.TestByteBuf.log;

public class TestSlice {
    public static void main(String[] args) {
        ByteBuf buf = ByteBufAllocator.DEFAULT.buffer(10);
        buf.writeBytes(new byte[]{'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j'});
        System.out.println("-----------------------有参slice--------------------------------");
        log(buf);

        // 在切换过程中没有发生数据复制
        ByteBuf f1 = buf.slice(0, 5);
        log(f1);
        ByteBuf f2 = buf.slice(5, 5);
        log(f2);

        System.out.println("----------------------更改分片内容---------------------------------");
        // 更改slice的内容，则原始ByteBuf也会受影响，因为底层都是同一款块内存
        f1.setByte(0,  'b');
        log(f1);
        log(buf);

        System.out.println("-----------------------无参slice--------------------------------");
        ByteBuf origin = ByteBufAllocator.DEFAULT.buffer(10);
        origin.writeBytes(new byte[]{1, 2, 3, 4});
        log(origin);
        origin.readByte();
        log(origin);
        // 无参slice是从原始ByteBuf的read index到write index之间的内容进行切片，切片后的max capacity被固定为这个区间的大小，不能追加write
        ByteBuf slice = origin.slice();
        log(slice);
//        slice.writeBytes(5);  // 会报错，IndexOutOfBoundsException异常
        System.out.println("-----------------------原始ByteBuf再次读操作--------------------------------");
        origin.readByte();
        log(origin);
        log(slice);

        System.out.println("-----------------------释放原始ByteBuf的内存--------------------------------");
        slice.retain();
        origin.release();
        log(origin);  //  如果没有slice.retain() -> io.netty.util.IllegalReferenceCountException
        log(slice);
    }
}

输出

-----------------------有参slice--------------------------------
read index:0 write index:10 capacity:10
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 62 63 64 65 66 67 68 69 6a |abcdefghij |
+--------+-------------------------------------------------+----------------+
read index:0 write index:5 capacity:5
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 62 63 64 65 |abcde |
+--------+-------------------------------------------------+----------------+
read index:0 write index:5 capacity:5
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 66 67 68 69 6a |fghij |
+--------+-------------------------------------------------+----------------+
----------------------更改分片内容---------------------------------
read index:0 write index:5 capacity:5
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 62 62 63 64 65 |bbcde |
+--------+-------------------------------------------------+----------------+
read index:0 write index:10 capacity:10
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 62 62 63 64 65 66 67 68 69 6a |bbcdefghij |
+--------+-------------------------------------------------+----------------+
-----------------------无参slice--------------------------------
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 |.... |
+--------+-------------------------------------------------+----------------+
read index:1 write index:4 capacity:10
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 02 03 04 |... |
+--------+-------------------------------------------------+----------------+
read index:0 write index:3 capacity:3
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 02 03 04 |... |
+--------+-------------------------------------------------+----------------+
-----------------------原始ByteBuf再次读操作--------------------------------
read index:2 write index:4 capacity:10
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 03 04 |.. |
+--------+-------------------------------------------------+----------------+
read index:0 write index:3 capacity:3
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 02 03 04 |... |
+--------+-------------------------------------------------+----------------+
-----------------------释放原始ByteBuf的内存--------------------------------
read index:2 write index:4 capacity:10
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 03 04 |.. |
+--------+-------------------------------------------------+----------------+
read index:0 write index:3 capacity:3
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 02 03 04 |... |
+--------+-------------------------------------------------+----------------+

10）duplicate

【零拷贝】的体现之一，就好比截取了原始ByteBuf所有内容，并且没有max capacity的限制，也是与原始ByteBuf使用同一块底层内存，只是读写指针是独立的。

示例1：

package cn.itcast.netty.c2;

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

import static cn.itcast.netty.c2.TestByteBuf.log;

public class TestDuplicate {
    public static void main(String[] args) {
        ByteBuf buf = ByteBufAllocator.DEFAULT.buffer(10);
        buf.writeBytes(new byte[]{'a', 'b', 'c', 'd'});
        log(buf);

        ByteBuf duplicate = buf.duplicate();
        log(duplicate);

        buf.writeBytes(new byte[]{'x', 'y'});
        log(buf);
        duplicate.writeBytes(new byte[]{'e', 'f', 'g', 'h', 'i', 'j', 'k'});
        log(buf);
        log(duplicate);
    }
}

输出

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| 61 62 63 64 |abcd |
+--------+-------------------------------------------------+----------------+
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| 61 62 63 64 |abcd |
+--------+-------------------------------------------------+----------------+
read index:0 write index:6 capacity:10
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 62 63 64 78 79 |abcdxy |
+--------+-------------------------------------------------+----------------+
read index:0 write index:6 capacity:64
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 62 63 64 65 66 |abcdef |
+--------+-------------------------------------------------+----------------+
read index:0 write index:11 capacity:64
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 62 63 64 65 66 67 68 69 6a 6b |abcdefghijk |
+--------+-------------------------------------------------+----------------+

11）copy

会将底层内存数据进行深拷贝，因此无论读写，都与原始ByteBuf无关。

12）CompositeByte

【零拷贝】的体现之一，可以将多个ByteBuf合并为一个逻辑上的ByteBuf，避免拷贝。

示例：

package cn.itcast.netty.c2;

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

import static cn.itcast.netty.c2.TestByteBuf.log;

public class TestCompositeByteBuf {
    public static void main(String[] args) {
        ByteBuf buf1 = ByteBufAllocator.DEFAULT.buffer();
        buf1.writeBytes(new byte[]{1, 2, 3, 4, 5});

        ByteBuf buf2 = ByteBufAllocator.DEFAULT.buffer();
        buf2.writeBytes(new byte[]{6, 7, 8, 9, 10});

        /*ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
        buffer.writeBytes(buf1).writeBytes(buf2);
        log(buffer);*/

        CompositeByteBuf buffer = ByteBufAllocator.DEFAULT.compositeBuffer();
        buffer.addComponents(true, buf1, buf2);
        log(buffer);
    }
}

输出

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

CompositeByteBuf是一个组合的ByteBuf，它内部维护了一个Component数组，每个Component管理一个ByteBuf，记录了这个ByteBuf相对于整体偏移量等信息，代表着整体中某一段的数据。

优点，对外是一个虚拟视图，组合这些ByteBuf不会产生内存复制；
缺点，复杂了很多，多次操作会带来性能的损耗。

13）Unpooled

Unpooled是一个工具类，类如其名，提供了非池化的ByteBuf创建、组合、复制等操作。

这里仅介绍其跟【零拷贝】相关的wrappedBuffer方法，可以用来包装ByteBuf

package cn.itcast.netty.c2;

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

import static cn.itcast.netty.c2.TestByteBuf.log;

public class TestUnpooled {
    public static void main(String[] args) {
        ByteBuf buf1 = ByteBufAllocator.DEFAULT.buffer(5);
        buf1.writeBytes(new byte[]{1, 2, 3, 4, 5});
        ByteBuf buf2 = ByteBufAllocator.DEFAULT.buffer(5);
        buf2.writeBytes(new byte[]{6, 7, 8, 9, 10});

        // 当包装ByteBuf个数超过1个时，底层使用了CompositeByteBuf
        ByteBuf buf3 = Unpooled.wrappedBuffer(buf1, buf2);
        log(buf3);

        // 也可以用来包装普通字节数组，底层也不会有拷贝操作
        ByteBuf buf4 = Unpooled.wrappedBuffer(new byte[]{1, 2, 3}, new byte[]{4, 5, 6});
        System.out.println(buf4.getClass());
        log(buf4);
    }
}

输出

read index:0 write index:10 capacity:10
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 05 06 07 08 09 0a |.......... |
+--------+-------------------------------------------------+----------------+
class io.netty.buffer.CompositeByteBuf
read index:0 write index:6 capacity:6
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 05 06 |...... |
+--------+-------------------------------------------------+----------------+

ByteBuf优势

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

4. 双向通信

需求：实现一个echo server

服务器端：

package cn.itcast.netty.c2;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.nio.charset.Charset;

public class EchoServer {
    public static void main(String[] args) throws InterruptedException {
        new ServerBootstrap()
                .group(new NioEventLoopGroup())
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                try {
                                    ByteBuf buffer = (ByteBuf) msg;
                                    System.out.println("Server received: " + buffer.toString(Charset.defaultCharset()));

                                    // 创建响应并使用相同的消息内容
                                    ByteBuf reponse = ctx.alloc().buffer();
                                    reponse.writeBytes(buffer);
                                    ctx.writeAndFlush(reponse);

                                    // 注意：这里不需要手动释放buffer，因为Netty会在pipeline处理完成后自动释放
                                } catch (Exception e) {
                                    e.printStackTrace();
                                    ctx.close();
                                }
                            }

                            @Override
                            public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
                                cause.printStackTrace();
                                ctx.close();
                            }
                        });
                    }
                }).bind(8080).sync().channel().closeFuture().sync();
    }
}

客户端：

package cn.itcast.netty.c2;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.ReferenceCountUtil;

import java.nio.charset.Charset;
import java.util.Scanner;

public class EchoClient {
    public static void main(String[] args) throws InterruptedException {
        NioEventLoopGroup group = new NioEventLoopGroup();
        try {
            Channel channel = new Bootstrap()
                    .group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<NioSocketChannel>() {
                        @Override
                        protected void initChannel(NioSocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new StringEncoder(Charset.defaultCharset()));
                            ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                                @Override
                                public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                    try {
                                        ByteBuf buffer = (ByteBuf) msg;
                                        System.out.println("Client received: " + buffer.toString(Charset.defaultCharset()));

                                        // 注意：这里不需要手动释放buffer，因为Netty会在pipeline处理完成后自动释放
                                    } finally {
                                        ReferenceCountUtil.release(msg);  // 更安全的做法
                                    }
                                }

                                @Override
                                public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
                                    cause.printStackTrace();
                                    ctx.close();
                                }
                            });
                        }
                    }).connect("localhost", 8080).sync().channel();

            channel.closeFuture().addListener(future -> {
                group.shutdownGracefully();  // 优雅关闭
            });

            Scanner sc = new Scanner(System.in);
            while (true) {
                String line = sc.nextLine();
                if("q".equals(line)) {
                    channel.close();
                    break;
                }
                channel.writeAndFlush(line);
            }
        } finally {
            group.shutdownGracefully();  // 优雅关闭
        }
    }
}

客户端输出

hello server
Client received: hello server
hello, world!
Client received: hello, world!
q

服务器端输出

Server received: hello server
Server received: hello, world!

关键点说明

1. 资源释放问题：

在简单的echo服务中，通常不需要手动释放ByteBuf，因为Netty会自动管理接收到的消息的内存。
但最佳实践是：
- 如果你消费了消息（如转发了它），则不需要释放；
- 如果你没有消费消息，应该调用ReferenceCountUtil.release(msg)
- 或者在handler前添加SimpleChannelInboundHandler，它会自动释放

2. 异常处理：