Netty核心技术

read() 从通道读取数据并放入到缓冲区

write() 把缓存区的数据写入到通道中

transferFrom() 从目标通道中复制数据到当前通道

transferTo() 把数据从当前通道复制给目标通道

mark

position

limit

capacity

open() 得到一个选择器对象

select() 监听所有注册的通道,当其中有IO操作可以进行时, 将对应的SelectorKey 加入到内部集合中返回, 参数用来设置超时时间

selectedKeys() 返回此选择器的选定键集

selectNow() 不阻塞,立马返回

wakeup() 唤醒selector

keys() 返回此选择器的键集

采用阻塞IO模式获取输入的数据

每个连接都需要独立的线程完成数据的输入,业务处理

当并发数很大, 就会创建大量的线程, 占用很大系统资源

连接创建后, 如果当前线程暂时没有数据可读, 该线程会阻塞在 Read 操作, 造成线程资源浪费

1), Reactor 模型,通过一个或多个输入同时传递给服务处理器的模型(基于事件驱动)

2), 服务器端程序会处理传入的多个请求,并将它们同步分派到相应的处理线程,因此 Reactor 模式也叫 Dispathcher 模式

3), Reactor 模式使用 IO 复用监听事件, 收到事件后分发给某个线程,这点就是网络高并发处理关键

1), Reactor : Reactor 在一个单独的线程中运行, 负责监听和分发事件, 分发给适当处理程序来对 IO 事件做出反应, 它像公司的电话接线员, 他接听来自客户电话并将线路转到适当的联系人

2), Handlers : 处理程序执行IO 事件要完成的实际事件, 类似于 客户想要与之交谈的公司中的实际官员, Reactor 通过适当的处理程序来响应 IO 事件, 处理程序执行非阻塞的操作

 children[i]元素的类型都是 NioEventLoop , NioEventLoop 实现了 EventLoop 接口 ,其中 EventLoop 继承 EventExecutor

children 的类型是 EventExecutor 所以可以放入NioEventLoop

channel = channelFactory.newChannel();

init(channel);

channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);

AbstractChannelHandlerContext 类的  next.invokeBind(localAddress, promise)

DefaultChannelPipeline 类的  unsafe.bind(localAddress, promise)

NioServerSocketChannel 类的 doBind 方法

1), 通过工具类,调用NioServerSocketChannel 内部封装, serverSocketChannel 的accept 方法,这是Nio做法

2), 获取到一个JDK的 SocketChanenl 然后使用 NIO 进行封装,最后添加到List容器中

3), 这样容器buf 就有了 NioSocketChannel 

协议解码器--将二进制数据转为java对象

协议编码器--将java对象转为二进制数据

业务处理程序 --执行实例业务逻辑

DefaultChannelPipeline(Channel channel)

1), pipeline 添加 handler ,参数是线程池, name是null, handler 是我们活着系统传入的 handler ,Netty 为了防止多个线程导致安全问题, 同步了这段代码

2), 检查这个 handler 实例是否共享的,如果不是 ,并且已经被别的 pipeline 使用了, 则抛出异常

3), 调用 newContext(group, filterName(name, handler), handler); 方法,创建一个 Context 从这里可以看出, 每次添加一个 handler 都会创建一个关联的 Context

4), 调用 addLast方法 ,将 context 追加到链表中

5), 如果这个通道还没有注册到 selecor 上, 就将这个 Context 添加到这个 pipeline 的代办任务中,就注册好了以后, 就会调用 callHandlerAdded0(newCtx); --默认是什么都不做,用户可以实现这个方法

到这里,针对3创建过程, 了解的差不多了, 和最初说的一样, 没当创建 ChannelSelecor 的时候都会创建一个绑定的 pipeline ,一对一的关系, 创建 pipeline 的时候也会创建 tail 和 head 节点,形成了最初的链表, tail 是入栈 inboud类型的handler ,head 是出栈outbound 的类型, 在调用pipeline 的 addLast 方法的时候, 会根据给定的 handler 创建一个 context 然后将这个context 插入到链表的尾端tail前面即可

idStateHandler

ReadTimeoutHandler

WriteTimeoutHandler

private final boolean observeOutput; //是否考虑出栈时较慢的情况,默认是 false

private final long readerIdleTimeNanos; //读事件空闲时间,0是禁用该事件

private final long writerIdleTimeNanos;//写事件空闲时间,0是禁用该事件

private final long allIdleTimeNanos;//读或写空闲时间,0是禁用该事件

只要给定的参数大于0, 就创建一个定时任务, 每个事件都创建,同时,将 State 状态设置为1,防止重复初始化,调用 initOutputChanged 方法, 监控出栈数据属性

1), 得到用户设置超时时间

2), 如果读操作结束了(执行了 ChannelReadComplete 方法设置) , 就用当前时间减去给定时间最后一次读操作的时间(执行 channelReadComplete 方法设置),如果小于0, 就触发事件, 反之继续放入队列, 间隔时间是新的计算时间

3), 触发的逻辑是,首先将任务再次放到队列, 时间是刚开始设置的时间, 返回一个 promise 对象, 用于取消操作,然后设置 first 属性为  false表示, 下一次读取不在是第一次了,这个属性在 channelRead 方法会被改成true

4), 创建一个 IdStateEvent 类型的写事件对象, 将次对象传递给用户的 userEventTriggered 方法完成触发事件的操作

5), 总的来说, 每次读取操作都会记录一个时间, 定时任务时间到了, 会计算当前时间和最后一次读的时间的间隔, 如果间隔超过了设置的时间,就触发 userEventTriggered 方法

写任务的代码逻辑基本和读任务的代码逻辑一样, 唯一不同的就是有一个针对 出栈较慢数据的判断 hasOutputChanged(ctx, first)

1), 表示这个监控所有的事件,当读写事件发生时, 都会记录, 代码了逻辑和写事件基本一致

2), 当前时间减去最后一次读或写,若大于0 说明超时了

3), 这里的时间计算是取最多大值来的, 然后像写事件一样, 判断是否发生了写的慢的情况

1), handler 中加入handler

2), handler 中添加线程池

3), context 中添加线程池

第一种在handler 中添加异步, 可以更加自由, 比如如果需要访问数据库,那我就异步, 如果不需要,则不异步, 异步会拖长接口响应时间, 因为需要将任务放入 mpscTask 中,如果IO时间很短, task 很多可能一个循环下来, 都没时间执行整个 task 导致响应时间不达标

第二种方式是 Netty 标准方式, 但是,那么做会将整个 handler 都交给业务线程池, 不论耗时不耗时, 都加入到队列中, 不够灵活

各有优势劣势,从灵活性考虑,第一种较好