深入分析netty（三）
文章目录
3.大名鼎鼎的Eventloop
3.1.1 关于Reactor 的线程模型
首先我们来看一下Reactor的线程模型
Reactor的线程模型有三种:
1、单线程模型
2、多线程模型
3、主从多线程模型
首先来看一下单线程模型:
所谓单线程，即acceptor处理和handler处理都在一个线程中处理.这个模型的坏处显而易见:当其中某个handler阻塞时，会导致其他所有的client 的handler 都得不到执行，并且更严重的是，handler的阻塞也会导致整个服务不能接收新的client请求(因为acceptor也被阻塞了)．因为有这么多的缺陷，因此单线程Reactor模型用的比较少.
那么什么是多线程模型呢? Reactor的多线程模型与单线程模型的区别就是 acceptor是一个单独的线程处理，并且有一组特定的NIO线程来负责各个客户端连接的IO操作. Reactor个单独的线程处理，并且有一组特定的NIO线程来负责各个客户端连接的IO操作. Reactor多线程模型如下:
Reactor多线程模型有如下特点: 1、有专门一个线程，即Acceptor线程用于监听客户端的TCP连接请求.
2、客户端连接的IO操作都是由一个特定的NIO线程池负责．每个客户端连接都与一个特定的NIO线程绑定，因此在这个客户端连接中的所有IO操作都是在同一个线程中完成的. 3、客户端连接有很多，但是NIO线程数是比较少的，因此一个NIO线程可以同时绑定到多个客户端连接中.
接下来我们再来看一下Reactor的主从多线程模型.
一般情况下，Reactor的多线程模式已经可以很好的工作了，但是我们考虑一下如下情况:
如果我们的服务器需要同时处理大量的客户端连接请求或我们需要在客户端连接时，权限的检查，那么单线程的 Acceptor很有可能就处理不过来，造成了大量的客户端不能连接到服务器.
Reactor的主从多线程模型就是在这样的情况下提出来的，它的特点是:服务器端接收客户端的连接请求不再是一个线程，而是由一个独立的线程池组成．它的线程模型如下:
可以看到，Reactor的主从多线程模型和Reactor多线程模型很类似，只不过Reactor的主从多线程模型的acceptor使用了线程池来处理大量的客户端请求.
3.1.2NioEventLoopGroup 与Reactor线程模型的对应
我们介绍了三种Reactor的线程模型，那么它们和NioEventLoopGroup又有什么关系呢?其实，不同的设置NioEventLoopGroup的方式就对应了不同的Reactor的线程模型.
单线程模型
来看一下下面的例子:
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
serverBootstrap server = new ServerBootstrap();
server.group(bossGroup);注意，我们实例化了一个NioEventLoopGroup，然后接着我们调用server.group(bossGroup)设置了服务器端的EventLoopGroup.有人可能会有疑惑:我记得在启动服务器端的Netty程序时，是需要设置bossGroup和workerGroup的，为什么这里就只有一个bossGroup? 其实很简单，ServerBootstrap重写了group方法:
public ServerBootstrap group(EventLoopGroup group) {
return group(group，group);
}因此当传入一个 group时，那么bossGroup和workerGroup 就是同一个NioEventLoopGroup了. 这时候呢，因为bossGroup和workerGroup就是同一个NioEventLoopGroup，并且这个NioEventLoopGroup只有一个线程，这样就会导致Netty中的acceptor和后续的所有客户端连接的IO操作都是在一个线程中处理的.那么对应到Reactor的线程模型中，我们这样设置
NioEventLoopGroup时，就相当于Reactor单线程模型.
多线程模型 同理，再来看一下下面的例子:
EventLoopGroup bossGroup = new NioEventLoopGroup(128);
ServerBootstrap server = new ServerBootstrap();
server.group(bossGroup);将bossGroup的参数就设置为大于1的数,其实就是Reactor多线程模型.
主从线程模型 相信同学们都想到了，实现主从线程模型的例子如下:
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup()﹔
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup，workerGroup);bossGroup 为主线程，而workerGroup中的线程是CPU核心数乘以2，因此对应的到Reactor线程模型中，我们知道，这样设置的NioEventLoopGroup其实就是Reactor主从多线程模型.
3.1.3 NioEventLoopGroup类层次结构
3.1.4 NioEventLoopGroup实例化过程
在前面的章节中我们已经简单地介绍了一下NioEventLoopGroup 的初始化过程，这里再回顾一 下:
即: 1、EventLoopGroup(其实是MultithreadEventExecutorGroup)内部维护一个类型为EventExecutor children数组，其大小是nThreads，这样就构成了一个线程池 2、如果我们在实例化 NioEventLoopGroup时，如果指定线程池大小，则nThreads就是指定的值，反之是处理器核心数* 2
3、MultithreadEventExecutorGroup中会调用newChild 抽象方法来初始化children数组 4、抽象方法newChild是在NioEventLoopGroup中实现的，它返回一个NioEventLoop实例. 5、NioEventLoop属性: SelectorProvider provider属性:NioEventLoopGroup构造器中通过 SelectorProvider. provider()获取一个SelectorProvider Selector selector属性: NioEventLoop构造器中通过调用通过selector= provider. openSelector(获聪一个selector对象.
3.1.5 NioEventLoop类层次结构
NioEventLoop继承于SingleThreadEventLoop，而SingleThreadEventLoop又继承于SingleThreadEventExecutor. SingleThreadEventExecutor是Netty中对本地线程的抽象，它内部有一个Thread thread属性，存储了一个本地Java线程．因此我们可以认为，一个NioEventLoop 其实和一个特定的线程绑定，并且在其生命周期内，绑定的线程都不会再改变.
NioEventLoop的类层次结构图还是比较复杂的，不过我们只需要关注几个重要的点即可．首先NioEventLoop的继承链如下: NioEventLoop -〉SingleThreadEventLoop -〉SingleThreadEventExecutor->AbstractScheduledEventExecutor 在AbstractScheduledEventExecutor中，Netty实现了NioEventLoop 的schedule功能，即我们可以通过调用一个NioEventLoop实例的schedule方法来运行一些定时任务．而在SingleThreadEventLoop中，又实现了任务队列的功能，通过它，我们可以调用一个NioEventLoop实例的execute方法来向任务队列中添加一个task，并由NioEventLoop进行调度执行. 通常来说，NioEventLoop肩负着两种任务，第一个是作为IO线程，执行与Channel相关的IO操作，包括调用select等待就绪的IO事件、读写数据与数据的处理等;而第二个任务是作为任务队列，执行taskQueue中的任务，例如用户调用eventLoop.schedule提交的定时任 务也是这个线程执行的.
3.1.6 NioEventLoop的实例化过程
从上图可以看到，SingleThreadEventExecutor有一个名为thread 的Thread类型字段，这个字段就代表了与SingleThreadEventExecutor关联的本地线程.
我们看看thread在哪里赋的值:
private void doStartThread() {
assert thread == null;
executor.execute(new Runnable() {
@Override
public void run() {
thread = Thread.currentThread();
if (interrupted) {
thread.interrupt();
}
boolean success = false;
updateLastExecutionTime();
try {
SingleThreadEventExecutor.this.run();
success = true;
} catch (Throwable t) {
logger.warn("Unexpected exception from an event executor: ", t);
} finally {
for (;;) {
int oldState = STATE_UPDATER.get(SingleThreadEventExecutor.this);
if (oldState >= ST_SHUTTING_DOWN || STATE_UPDATER.compareAndSet(
SingleThreadEventExecutor.this, oldState, ST_SHUTTING_DOWN)) {
break;
}
}
// Check if confirmShutdown() was called at the end of the loop.
if (success && gracefulShutdownStartTime == 0) {
logger.error("Buggy " + EventExecutor.class.getSimpleName() + " implementation; " +
SingleThreadEventExecutor.class.getSimpleName() + ".confirmShutdown() must be called " +
"before run() implementation terminates.");
}
try {
// Run all remaining tasks and shutdown hooks.
for (;;) {
if (confirmShutdown()) {
break;
}
}
} finally {
try {
cleanup();
} finally {
STATE_UPDATER.set(SingleThreadEventExecutor.this, ST_TERMINATED);
threadLock.release();
if (!taskQueue.isEmpty()) {
logger.warn(
"An event executor terminated with " +
"non-empty task queue (" + taskQueue.size() + ')');
}
terminationFuture.setSuccess(null);
}
}
}
}
});
}SingleThreadEventExecutor启动时会调用dostartThread方法,然后执行executor.execute方法，将当前线程赋值给thread.在这个线程中所做的事情主要就是调用SingleThreadEventExecutor.this.run()方法，而因为NioEventLoop实现了这个方法，因此根据多态性，其实调用的是NioEventLoop.run(方法.
3.1.7 EventLoop 与Channel的关联
Netty中，每个Channel都有且仅有一个EventLoop与之关联，它们的关联过程如下:
从上图中我们可以看到，当调用了AbstractChannel$AbstractUnsafe.register后，就完成了Channel和EventLoop的关联. register实现如下:
@Override
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
if (eventLoop == null) {
throw new NullPointerException("eventLoop");
}
if (isRegistered()) {
promise.setFailure(new IllegalStateException("registered to an event loop already"));
return;
}
if (!isCompatible(eventLoop)) {
promise.setFailure(
new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));
return;
}
AbstractChannel.this.eventLoop = eventLoop;
if (eventLoop.inEventLoop()) {
register0(promise);
} else {
try {
eventLoop.execute(new Runnable() {
@Override
public void run() {
register0(promise);
}
});
} catch (Throwable t) {
logger.warn(
"Force-closing a channel whose registration task was not accepted by an event loop: {}",
AbstractChannel.this, t);
closeForcibly();
closeFuture.setClosed();
safeSetFailure(promise, t);
}
}
}在AbstractChannel$AbstractUnsafe.register中，会将一个EventLoop赋值给AbstractChannel内部的eventLoop 字段，到这里就完成了EventLoop 与 Channel的关联过程.
3.1.7 EventLoop 的启动
在前面我们已经知道了，NioEventLoop本身就是一个SingleThreadEventExecutor，因此NioEventLoop的启动，其实就是NioEventLoop 所绑定的本地 Java线程的启动.依照这个思想，我们只要找到在哪里调用了SingleThreadEventExecutor的thread 字段的 start(方法就可以知道是在哪里启动的这个线程了. 从代码中搜索，thread.start()被封装到SingleThreadEventExecutor.startThread()方法中了:
private void startThread() {
if (STATE_UPDATER.get(this) == ST_NOT_STARTED) {
if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {
doStartThread();
}
}
}STATE_UPDATER是SingleThreadEventExecutof内部维护的一个属性，它的作用是标识当前的thread的状态．在初始的时候，STATE_UPDATER== sT_NOT_STARTED，因此第一次调用startThread()方法时，就会进入到if语句内，进而调用到thread.start(). 而这个关键的startThread()方法又是在哪里调用的呢?经过方法调用关系搜索，我们发现，startThread是SingleThreadEventExecutor.execute方法中调用的:
@Override
public void execute(Runnable task) {
if (task == null) {
throw new NullPointerException("task");
}
boolean inEventLoop = inEventLoop();
if (inEventLoop) {
addTask(task);
} else {
startThread();
addTask(task);
if (isShutdown() && removeTask(task)) {
reject();
}
}
if (!addTaskWakesUp && wakesUpForTask(task)) {
wakeup(inEventLoop);
}
}既然如此，那现在我们的工作就变为了寻找―在哪里第一次调用了SingleThreadEventExecutor.execute()方法. 如果留心的同学可能已经注意到了，我们在前面EventLoop 与Channel的关联这一小节时，有提到到在注册channel 的过程中，会在 AbstractChannel
AbstractUnsafe.register方法，整个代码都是在主线程中运行的，因此上面的eventLoop. inEventLoop()就为false，于是进入到else分支，在这个分支中调用了eventLoop.execute. eventLoop 是一个 NioEventLoop的实例，而NioEventLoop没有实现execute方法，因此调用的是SingleThreadEventExecutor.execute:
我们已经分析过了，inEventLoop == false，因此执行到else分支，在这里就调用了startThread()方法来启动SingleThreadEventExecutor内部关联的Java本地线程了. 总结一句话，当EventLoop.execute第一次被调用时，就会触发startThread()的调用，进而导致了EventLoop所对应的 Java线程的启动. 我们将EventLoop与Channel 的关联小节中的时序图补全后，就得到了EventLoop启动过程的时序图:
4.Promise与Future双子星的秘密
java.util.concurrent.Future是Java提供的接口，表示异步执行的状态，Future的get方法会判断任务是否执行完成，如果完成就返回结果，否则阻塞线程，直到任务完成。 Netty扩展了Java的Future，最主要的改进就是增加了监听器Listener接口，通过监听器可以让异步执行更加有效率，不需要通过get来等待异步执行结束，而是通过监听器回调来精确地控制异步执行结束的时间点。
ChannelFuture接口扩展了Netty的Future接口，表示一种没有返回值的异步调用，同时关联了Channel，跟一个Channel绑定
Promise接口也扩展了Future接口，它表示一种可写的Future，就是可以设置异步执行的结果
ChannelPromise接口扩展了Promise和ChannelFuture，绑定了Channel，又可写异步执行结构，又具备了监听者的功能，是Netty实际编程使用的表示异步执行的接口
DefaultChannelPromise是ChannelPromise的卖现类，它是实际运行时的Promoise实例。Netty 使用addListener的方式来回调异步执行的结果。 看一下DefaultPromise的addListener方法，它判断异步任务执行的状态，如果执行完成，就理解通知监听者，否则加入到监听者队列通知监听者就是找一个线程来执行调用监听的回调函数。
@Override
public Promise<V> addListener(GenericFutureListener<? extends Future<? super V>> listener) {
checkNotNull(listener, "listener");
synchronized (this) {
addListener0(listener);
}
if (isDone()) {
notifyListeners();
}
return this;
}
private void addListener0(GenericFutureListener<? extends Future<? super V>> listener) {
if (listeners == null) {
listeners = listener;
} else if (listeners instanceof DefaultFutureListeners) {
((DefaultFutureListeners) listeners).add(listener);
} else {
listeners = new DefaultFutureListeners((GenericFutureListener<? extends Future<V>>) listeners, listener);
}
}
private void notifyListeners() {
EventExecutor executor = executor();
if (executor.inEventLoop()) {
final InternalThreadLocalMap threadLocals = InternalThreadLocalMap.get();
final int stackDepth = threadLocals.futureListenerStackDepth();
if (stackDepth < MAX_LISTENER_STACK_DEPTH) {
threadLocals.setFutureListenerStackDepth(stackDepth + 1);
try {
notifyListenersNow();
} finally {
threadLocals.setFutureListenerStackDepth(stackDepth);
}
return;
}
}
safeExecute(executor, new Runnable() {
@Override
public void run() {
notifyListenersNow();
}
});
}再来看监听者的接口，就一个方法，即等异步任务执行完成后，拿到Future结果，执行回调的 逻辑
public interface GenericFutureListener<F extends Future<?>> extends EventListener {
/**
* Invoked when the operation associated with the {@link Future} has been completed.
*
* @param future the source {@link Future} which called this callback
*/
void operationComplete(F future) throws Exception;
}5. Handler的各种姿势
5.1 Channe lHandlerContext
每个ChannelHandler被添加到ChannelPipeline后，都会创建一个Channe lHandlerContext并与之创建的Channe lHandler关联绑定。Channe lHandl erContext允许ChannelHandler与其他的ChannelHandler实现进行交互。ChannelHandlerContext不会改变添加到其中的ChannelHandler，因此它是安全的。 下图显示了Channe lHandl erContext、Channe lHandler、ChannelPipeline 的关系:
5.2 Channel的状态模型
Netty有一个简单但强大的状态模型，并完美映射到Channel InboundHandler的各个方法。下面是Channel生命周期四个不同的状态:
- channe lUnregi stered
- channe lRegistered
- channelActive
- channelInactive
Channel的状态在其生命周期中变化，因为状态变化需要触发，下图显示了Channel状态变化:
5.2 ChannelHandler 和其子类
先看一张Handler的类继承图
5.3 ChannelHandler中的方法
Netty定义了良好的类型层次结构来表示不同的处理程序类型，所有的类型的父类是Channe lHandler。ChannelHandler 提供了在其生命周期内添加或从ChannelPipeline中删除的方法。
- handl erAdded, ChannelHandler 添加到实际上下文中准备处理事件
- handlerRemoved, 将ChannelHandler从实际上下文中删除，不再处理事件
- except ionCaught,处理抛出的异常
Netty还提供了一个实现了ChannelHandler的抽象类ChannelHandl erAdapter。 ChannelHandlerAdapter实现了父类的所有方法，基本上就是传递事件到ChannelPipeline中的下一个ChannelHandler直到结束。我们也可以直接继承于Channe lHandlerAdapter，然后重写里面的方法。 5.4 ChannelInboundHandler Channel InboundHandler提供了一些方法再接收数据或Channel状态改变时被调用。下面是ChannelInboundHandler的一些方法:
- channelRegi stered, ChannelHandl erContext的Channel被注册到EventLoop;
- channe lUnregistered, Channe lHandlerContext的Channel从EventLoop中注销
- channe lActive, ChannelHandlerContext的Channel已激活
- channe l Inactive, Channe lHanderContxt的Channel结束生命周期
- channelRead, 从当前Channel的对端读取消息
- channelReadComplete， 消息读取完成后执行运。
- userLventTriggered, 一个用户事件被触发
- channe lWri tabilityChanged,改变通道的可写状态，可以使用Channel. isWritable()检查
- except ionCaught，重写父类ChannelHandler的方法，处理异常
Netty提供了一个实现了Channel InboundHandler接口并继承ChannelHandlerAdapter的类: Channel InboundHandlerAdapter Channe lInboundHandlerAdapter实现了Channel InboundHandler的所有方法，作用就是处理消息并将消息转发到ChannelPipeline中的下一个Channe lHandler。ChannelInboundHandlerAdapter的channelRead方法处理完消息后不会自动释放消息，若想自动释放收到的消息，可以使用SimpleChannelInboundHandler。 看下面的代码:
public class UnreleaseHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead (ChannelHandlerContext ctx, Object msg) throws Exception {
//手动释放消息
ReferenceCountUtil.release(msg);
}
}SimpleChannelInboundHandler会自动释放消息
public class ReleaseHandler extends SimpleChannelInboundHandler<Object> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
//不需要手动释放
}
}ChannelInitializer用来初始化Channellandfer，将自定义的各种ChannelHlandler添加到ChannelPipeline中。
6.数据翻译官编码和解码
6.1TCP黏包/拆包
TCP是一个“流”协议，所谓流，就是没有界限的一长串二进制数据。TCP作为传输层协议并不了解上层业务数据的具体含义，它会根据TCP缓冲区的实际情况进行数据包的划分，所以在业务上认为是一个完整的包，可能会被CP拆分成多个包进行发送，也有可能把多个小的包封装成一个大的数据包发送，这就是所谓的TCP粘包和拆包问题。
6.2粘包问题的解决策略
由于底层的TCP无法理解上层的业务数据，所以在底层是无法保证数据包不被拆分和重组的，这个问题只能通过上层的应用协议栈设计来解决。业界的主流协议的解决方案，可以归纳如下:
- 消息定长，报文大小固定长度，例如每个报文的长度固定为200字节，如果不够空位补空格;
- 包尾添加特殊分隔符，例如每条报文结束都添加回车换行符（例如FTP协议）或者指定特殊字符作为报文分隔符，接收方通过特殊分隔符切分报文区分;
- 将消息分为消息头和消息体，消息头中包含表示信息的总长度（或者消息体长度）的字段;
- 更复杂的自定义应用层协议。
6.3编、解码技术
通常我们也习惯将编码(Encode）称为序列化(serialization)，它将对象序列化为字节数组，用于网络传输、数据持久化或者其它用途。 反之，解码(Decode)/反序列化(deserialization）把从网络、磁盘等读取的字节数组还原成原始对象（通常是原始对象的拷贝），以方便后续的业务逻辑操作。 进行远程跨进程服务调用时（例如RPC调用），需要使用特定的编解码技术，对需要进行网络传输的对象做编码或者解码，以便完成远程调用。
6.4 Netty为什么要提供编解码框架
作为一个高性能的异步、NIO通信框架，编解码框架是Netty的重要组成部分。尽管站在微内核的角度看，编解码框架并不是Netty微内核的组成部分，但是通过ChannelHandler定制扩展出的编解码框架却是不可或缺的。 然而，我们已经知道在Netty中，从网络读取的Inbound消息，需要经过解码，将二进制的数据报转换成应用层协议消息或者业务消息，才能够被上层的应用逻辑识别和处理;同理，用户发送到网络的Outbound业务消息，需要经过编码转换成二进制字节数组（对于Netty就是ByteBuf) 才能够发送到网络对端。编码和解码功能是NIO框架的有机组成部分，无论是由业务定制扩展实现，还是NIO框架内置编解码能力，该功能是必不可少的。 为了降低用户的开发难度，Netty对常用的功能和API做了装饰，以屏蔽底层的实现细节。编解码功能的定制，对于熟悉Netty底层实现的开发者而言，直接基于Channellandler扩展开发，难度并不是很大。但是对于大多数初学者或者不愿意去了解底层实现细节的用户，需要提供给他们更简单的类库和API，而不是ChannelHandler。 Netty在这方面做得非常出色，针对编解码功能，它既提供了通用的编解码框架供用户扩展，又提供了常用的编解码类库供用户直接使用。在保证定制扩展性的基础之上，尽量降低用户的开发工作量和开发门槛，提升开发效率。 Netty预置的编解码功能列表如下: base64、Protobuf、JBoss Marshalling、spdy等。
6.5Netty粘包和拆包解决方案
6.5.1 Netty中常用的解码器
Netty 提供了多个解码器，可以进行分包的操作，分别是:
- LineBasedFrameDecoder
- DelimiterBasedFrameDecoder（添加特殊分隔符报文来分包）
- FixedLengthFrameDecoder（使用定长的报文来分包）
- LengthFieldBasedFrameDecoder
腾讯云开发者
