【Netty】Netty模型

Netty模型

NioEventLoopGroup

Netty抽象出两组线程池：BossGroup，WorkerGroup

这两个线程组都是NioEventLoopGroup实例，只是分工不同；

• Boss Group：专门负责客户端的连接；
• Worker Group：专门负责网络的读写；

每个线程池中，含有多个NioEventLoop线程（默认线程数：CPU * 2）；

Boss组下的循环线程（NioEventLoop）：

1. selector轮询accept事件（只关注连接事件）；
2. 处理accept事件，与客户端建立连接，并将其注册到Worker组下的某个NioEventLoop中的selector上；
3. runAllTasks处理任务队列；

Worker组下的循环线程（NioEventLoop）：

1. selector轮询read/write事件（只关注注册之后的读写事件）；
2. 根据key，获取对应的SocketChannel处理IO事件；
3. runAllTasks处理任务队列；

NioEventLoop

本质就是一个不断执行循环任务的线程；

含有组件：Selector，TaskQueue，Executor等等；

内部采用串行化设计：读取--->解码--->处理--->编码--->发送

一个NioChannel，只会绑定唯一的NioEventLoop，并拥有自己的ChannelPipeline；

pipeline

是一个保存了多个ChannelHandler的双向链表管道；

• 一个Channel包含一个Channelpipeline；
• ChannelPipeline中包含由多个ChannelHandlerContext组成的双向链表；
• 每个ChannelHandlerContext维护一个ChannelHandler;

入站事件：由Head传递到Tail；

出站事件：由tail传递到head；

两种类型事件，互不干扰；

可以从ChannelHandlerContext获取很多相关信息：

Channel channel();    		// 获取对应Channel
EventExecutor executor();	// 获取使用的执行器
ChannelHandler handler();	// 获取对应的ChannelHandler
ChannelPipeline pipeline();	// 获取当前的pipeline
// 底层调用pipeline.writeAndFlush，从tail---flush
ChannelFuture writeAndFlush(Object msg);	

编解码器

Socket即网络；网络中的数据，都是字节流；

• 读取接受网络数据，即入栈，入栈首先要解码：将字节流转化为原本的数据格式；
• 发送数据到网络，即出栈，出栈首先要编码：将原本的数据格式转化为字节流；

ChannelHandler

大量存在于pipeline中，数据在管道中传输，经由多个Handler进行处理；

ChannelHandler是顶层父类；

子类：ChannelOutboundHandler：出栈，ChannelInboundHandler：入栈

可以自定义，需要继承ChannelOutboundHandlerAdapter，ChannelInboundHandlerAdapter

然后重写多种方法，来对数据进行处理：

channelRead			// 读事件
channelReadComplete	// 读完毕事件
channelActive		// 通道就绪事件
exceptionCaught		// 通道关闭事件
.....

TaskQueue

比如：ServerHandler在处理某个client的IO事件时，非常耗时，可能阻塞，我们不希望它阻塞，希望它异步执行，就可以将此channel提交给NioEventLoop中的TaskQueue；

三种典型场景：

1. 用户程序自定义的普通任务--->execute(）

   ctx.channel().eventLoop().execute(() -> {});
   

2. 用户自定义的定时任务--->schedule(）

   ctx.channel().eventLoop().schedule()
   

3. 推送Channel到不同的业务线程中

ChannelFuture

是netty的异步模型，建立在：future和 callback之上；

Bind、Write、Connect等操作会简单的返回一个ChannelFuture；

通过Future-Listener机制，用户可以方便的主动获取或者通过通知机制获得IO 操作结果;

ChannelFuture cf = serverBootstrap.bind(6669).sync();
// 绑定成功后，立刻回调
cf.addListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture future) throws Exception {
        // 实现监听
        if (cf.isSuccess()){...}
        if (cf.isDone()){...}
        ...
    }
});

异步提交channel到TaskQueue也可以进行异步监听：

ctx.channel().eventLoop().execute(() -> {
    try {
        Thread.sleep(10 * 1000);
        ChannelFuture channelFuture = ctx.writeAndFlush(Unpooled.copiedBuffer("{-----execute发送消息-----}", CharsetUtil.UTF_8));
        // 此线程执行完毕会返回结果（10s，发送成功后执行）
        channelFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (channelFuture.isSuccess()) {
                    System.out.println("------TaskQueue监听成功------");
                } else
                    System.out.println("------TaskQueue监听失败------");
            }
        });
    } catch (Exception e) {
        System.out.println("Exception：" + e.getMessage());
    }
});

Netty的buffer

buf工具类：Unpooled；

Netty中的buf，不需要反转（flip）

底层维护了两个索引：readerIndex（ridx），writerIndex（widx）

ridx只能读取widx之前的数据；

创建buf：

// 初始化一个buf：ridx：0，widx：0，cap：10
ByteBuf byteBuf = ByteBuf buffer(int initialCapacity, int maxCapacity)
// 创建一个buf：ridx：0，widx：12，cap：36
ByteBuf byteBuf = Unpooled.copiedBuffer("hello server", CharsetUtil.UTF_8);

读写：

// 写入buf，修改 writerIndex
buf.writeByte(i);
// 读取buf数据，修改 readerIndex
buf.readByte();
// 获取buf数据，不修改 readerIndex
buf.getByt(i);