一、背景
看过之前文章的同学都知道,Netty 是一个异步事件驱动的网络应用程序框架,用于快速开发可维护的高性能协议服务器和客户端。
上一篇文章也聊了Reactor线程模型的设计,主要是偏理论方向的,那么今天我们聊一聊实战,就是netty是如何实现Reactor模式的,一方面帮助我们了解netty的底层实现,另外一方面当我们遇到同样的业务场景时,也可以自己实现一个Reactor模式。
对Reactor不熟悉的,可以看看上一篇:
二、Reactor核心角色
前文得知,netty使用的是主从Reactor多线程模型,话不多说,直接上图:

从图中也可以看出,核心角色:
1、MainReactor、SubReactor:主从Reactor线程,两者结构相似,功能不同。
MainReactor通过select监控建立连接事件,收到事件后通过Acceptor接收,处理建立连接事件,然后把建立好的连接分配给子线程SubReactor处理
SubReactor将连接加入连接队列进行监听,并创建一个Handler用于处理各种连接事件
2、Acceptor:处理客户端连接请求
3、Handler:执行非阻塞读/写,read读取数据后,会分发给后面的Worker线程池进行业务处理
4、Worker:Worker线程池会分配独立的线程完成真正的业务处理,如何将响应结果发给Handler进行处理
主要是以上四块,下面我们看看Netty是如何实现的,这里以netty源码为主,但是只考虑主逻辑,一些异常分支会省略
三、Netty实现的Reactor
如下图:

Boss Group 对应的是 MainReactor 角色,Work Group 对应的是 SubReactor 角色
Boss Group 和 Work Group 都是 NioEventLoopGroup, 只是负责的分工不同。Boss Group 下的 EventLoop 处理 Accept 事件,Work Group 下的 EventLoop 处理 Read / Write 等事件
下面我们详细了解几个很核心的概念
先看看下面 EventLoopGroup、EventLoop 的类图,看个大概就行,类太多,没有理解的时候怕没有耐心看完。

简单总结一下:
EventLoop 继承 EventLoopGroup
最上面是 JDK 的 JUC 包里的 Executor,说明 EventLoop 、EventLoopGroup 都是线程池
EventLoopGroup:
下面直接从代码层面出发,看看是 EventLoopGroup 的结构及工作流程,我们以NioEventLoopGroup 为例。
在 Netty 服务端中,我们会设置的两个组,如下:
this.bossGroup = new NioEventLoopGroup();this.workerGroup = new NioEventLoopGroup();
先看看NioEventLoopGroup类的构造函数
public NioEventLoopGroup() {this(0);}public NioEventLoopGroup(int nThreads) {this(nThreads, (Executor) null);}
是一层一层嵌套的,每层都会添加一个默认参数
下图是全参数的构造函数:
public NioEventLoopGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory,final SelectorProvider selectorProvider,final SelectStrategyFactory selectStrategyFactory,final RejectedExecutionHandler rejectedExecutionHandler) {super(nThreads, executor, chooserFactory, selectorProvider, selectStrategyFactory, rejectedExecutionHandler);}
看到参数这么多,先尝试着简单解释一下:
nThreads: 线程池中的线程数,也就是 NioEventLoop 的实例数量;默认值:CPU * 2
executor: 这个就比较奇怪了,我们本身就是个线程池,为啥还要传一个线程池实例?(后面讲)
chooserFactory:当任务提交时,线程池选择一个线程来执行这个任务,就是选择策略
selectorProvider:selector提供者,因为每个NioEventLoop都有一个selector,所以需要一个provider
selectStrategyFactory:selector策略工厂,每个NioEventLoop都有一个selector,所以需要一个selector的策略工厂,说白了就是工作流程
rejectedExecutionHandler: 拒绝策略
这么多参数,具体干嘛用的,一路追查下去,我们找到他的父类的父类,看看构造函数:
protected MultithreadEventExecutorGroup(int nThreads, Executor executor,EventExecutorChooserFactory chooserFactory, Object... args) {if (nThreads <= 0) {throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));}if (executor == null) {executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());}children = new EventExecutor[nThreads];for (int i = 0; i < nThreads; i ++) {boolean success = false;try {children[i] = newChild(executor, args);success = true;} catch (Exception e) {// TODO: Think about if this is a good exception typethrow new IllegalStateException("failed to create a child event loop", e);} finally {if (!success) {for (int j = 0; j < i; j ++) {children[j].shutdownGracefully();}for (int j = 0; j < i; j ++) {EventExecutor e = children[j];try {while (!e.isTerminated()) {e.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS);}} catch (InterruptedException interrupted) {// Let the caller handle the interruption.Thread.currentThread().interrupt();break;}}}}}chooser = chooserFactory.newChooser(children);final FutureListener<Object> terminationListener = new FutureListener<Object>() {@Overridepublic void operationComplete(Future<Object> future) throws Exception {if (terminatedChildren.incrementAndGet() == children.length) {terminationFuture.setSuccess(null);}}};for (EventExecutor e: children) {e.terminationFuture().addListener(terminationListener);}Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);Collections.addAll(childrenSet, children);readonlyChildren = Collections.unmodifiableSet(childrenSet);}
仔细看一下这里的逻辑:
a、children 先赋值一个EventExecutor数组,大小就是传递进来的值,若没有传递,默认值为当前CPU * 2
children = new EventExecutor[nThreads];
b、循环实例化children,异常逻辑就不看了
for (int i = 0; i < nThreads; i ++) {boolean success = false;try {children[i] = newChild(executor, args);success = true;} catch (Exception e) {// TODO: Think about if this is a good exception typethrow new IllegalStateException("failed to create a child event loop", e);} finally {。。。略}}}
具体看看实例化方法,把NioEventLoopGroup带着的参数传递进去
protected EventLoop newChild(Executor executor, Object... args) throws Exception {return new NioEventLoop(this, executor, (SelectorProvider) args[0],((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]);}
原来EventLoopGroup的参数,大部分是为例传递给EventLoop
c、选择执行任务的线程
chooser = chooserFactory.newChooser(children);
d、后面就是为每个 EventLoop 加销毁的 listener 和 设置 EventLoop 为只读Set
final FutureListener<Object> terminationListener = new FutureListener<Object>() {@Overridepublic void operationComplete(Future<Object> future) throws Exception {if (terminatedChildren.incrementAndGet() == children.length) {terminationFuture.setSuccess(null);}}};for (EventExecutor e: children) {e.terminationFuture().addListener(terminationListener);}Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);Collections.addAll(childrenSet, children);readonlyChildren = Collections.unmodifiableSet(childrenSet);
到这里EventLoopGroup的流程就差不多了,下面我们通过newChild方法创建一个EventLoop实例
@Overrideprotected EventLoop newChild(Executor executor, Object... args) throws Exception {return new NioEventLoop(this, executor, (SelectorProvider) args[0],((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]);}
EventLoop:
这里我们以NioEventLoop为主
下面是NioEventLoop类自身的主要属性:
private Selector selector;private Selector unwrappedSelector;private SelectedSelectionKeySet selectedKeys;private final SelectorProvider provider;private final AtomicBoolean wakenUp = new AtomicBoolean();private final SelectStrategy selectStrategy;private volatile int ioRatio = 50;private int cancelledKeys;private boolean needsToSelectAgain;
再看看NioEventLoop的构造方法:
NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider,SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler) {super(parent, executor, false, DEFAULT_MAX_PENDING_TASKS, rejectedExecutionHandler);if (selectorProvider == null) {throw new NullPointerException("selectorProvider");}if (strategy == null) {throw new NullPointerException("selectStrategy");}provider = selectorProvider;final SelectorTuple selectorTuple = openSelector();selector = selectorTuple.selector;unwrappedSelector = selectorTuple.unwrappedSelector;selectStrategy = strategy;}
父类方法下面看
核心就是selector初始化,具体openSelector方法下面再讲
看完自身的属性和构造函数,我们再去看看他的父类SingleThreadEventLoop构造函数:
protected SingleThreadEventLoop(EventLoopGroup parent, Executor executor,boolean addTaskWakesUp, int maxPendingTasks,RejectedExecutionHandler rejectedExecutionHandler) {super(parent, executor, addTaskWakesUp, maxPendingTasks, rejectedExecutionHandler);tailTasks = newTaskQueue(maxPendingTasks);}
这里初始化一个tailTasks,先略过
再看SingleThreadEventLoop的父类SingleThreadEventExecutor构造函数
protected SingleThreadEventExecutor(EventExecutorGroup parent, Executor executor,boolean addTaskWakesUp, int maxPendingTasks,RejectedExecutionHandler rejectedHandler) {super(parent);this.addTaskWakesUp = addTaskWakesUp;this.maxPendingTasks = Math.max(16, maxPendingTasks);this.executor = ObjectUtil.checkNotNull(executor, "executor");taskQueue = newTaskQueue(this.maxPendingTasks);rejectedExecutionHandler = ObjectUtil.checkNotNull(rejectedHandler, "rejectedHandler");}
看到这里简单总结一下EventLoop核心属性:
一个Selector选择器
一个线程安全的LinkedBlockingQueue任务队列
一个单线程的线程池
其它概念先有个大概了解,后续涉及相关文章会详细讲解
Channel: Netty 网络通信的主体,由它负责同对端进行网络通信、注册和数据操作等功能。每个 Channel 会绑定一个 EventLoop,绑定后 Channel 所有的操作,都由该 EventLoop 实例来完成。
channelhandler:具体处理数据的地方
ChannelPipeline:链接各个 ChannelHandler,通过 ChannelHandlerContext传递上下文
2、具体工作流程
先大概看看下面的流程图

核心逻辑就在EventLoop里面,接下来我们以NioEventLoopGroup
为入手点,把EventLoop(NioEventLoop实例)流程梳理一下
既然EventLoop是线程池,那么看一下启动方法execute:
public void execute(Runnable task) {if (task == null) {throw new NullPointerException("task");}// 判断是否为当前线程执行的eventLoopboolean inEventLoop = inEventLoop();if (inEventLoop) {// 添加任务到队列, 默认16addTask(task);} else {// 先启动线程startThread();// 再添加任务队列addTask(task);if (isShutdown() && removeTask(task)) {reject();}}if (!addTaskWakesUp && wakesUpForTask(task)) {wakeup(inEventLoop);}}
继续看看如何启动线程的
private void startThread() {if (STATE_UPDATER.get(this) == ST_NOT_STARTED) {// CAS赋值if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) { doStartThread();}}}private void doStartThread() {assert thread == null;// 线程池创建一个线程实例executor.execute(new Runnable() {@Overridepublic void run() {// 设置线程为当前线程thread = Thread.currentThread();if (interrupted) {thread.interrupt();}boolean success = false;updateLastExecutionTime();try {// 执行SingleThreadEventExecutor 的run方法SingleThreadEventExecutor.this.run();success = true;} catch (Throwable t) {logger.warn("Unexpected exception from an event executor: ", t);} finally {...略}}});}
线程启动后,会执行SingleThreadEventExecutor(NioEventLoop)的run()方法,终于来到核心的地方了,先贴段代码:
protected void run() {// 首先是个死循环,处理IO事件和异步任务for (;;) {try {// 根据是否有任务,来判断具体操作// 有任务,执行selectNow(),检查当前是否有就绪的 IO 事件, 如果有, 则返回就绪 IO 事件的个数; 如果没有, 则返回0,该方法不会阻塞// 没有任务,执行select(wakenUp.getAndSet(false)),会阻塞,下面会详细讲switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {case SelectStrategy.CONTINUE:continue;case SelectStrategy.SELECT:select(wakenUp.getAndSet(false));if (wakenUp.get()) {selector.wakeup();}default:// fallthrough}cancelledKeys = 0;needsToSelectAgain = false;// IO操作与任务处理时间占比,默认为50%,可以在辅助类ServerBootstrap配置启动参数时设置// 值越大,处理IO操作占比越大,当为100时,不考虑时间,先处理IO事件,再执行异步任务final int ioRatio = this.ioRatio;if (ioRatio == 100) {try {// 查询就绪的 IO 事件,下面详细讲processSelectedKeys();} finally {// 任务处理机制,下面详细讲runAllTasks();}} else {final long ioStartTime = System.nanoTime();try {processSelectedKeys();} finally {// Ensure we always run tasks.final long ioTime = System.nanoTime() - ioStartTime;runAllTasks(ioTime * (100 - ioRatio) / ioRatio);}}} catch (Throwable t) {handleLoopException(t);}// Always handle shutdown even if the loop processing threw an exception.try {if (isShuttingDown()) {closeAll();if (confirmShutdown()) {return;}}} catch (Throwable t) {handleLoopException(t);}}}
到这里,我们终于找到了NioEventLoop事件循环的核心,下面我们再来分析一下(select、processSelectedKeys、runAllTasks)这三个核心的方法。
先看看select源码:
//这个方法解决了Nio中臭名昭著的bug:selector的select方法导致空轮询 cpu100%private void select(boolean oldWakenUp) throws IOException {Selector selector = this.selector;try {int selectCnt = 0;long currentTimeNanos = System.nanoTime();/* delayNanos(currentTimeNanos):计算延迟任务队列中第一个任务的到期执行时间(即最晚还能延迟多长时间执行),默认返回1s。每个SingleThreadEventExecutor都持有一个延迟执行任务的优先队列PriorityQueue,启动线程时,往队列中加入一个任务。*/long selectDeadLineNanos = currentTimeNanos + delayNanos(currentTimeNanos);for (;;) {/* 如果延迟任务队列中第一个任务的最晚还能延迟执行的时间小于500000纳秒,且selectCnt == 0(selectCnt 用来记录selector.select方法的执行次数和标识是否执行过selector.selectNow()),则执行selector.selectNow()方法并立即返回。*/long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L;if (timeoutMillis <= 0) {if (selectCnt == 0) {selector.selectNow();selectCnt = 1;}break;}if (hasTasks() && wakenUp.compareAndSet(false, true)) {selector.selectNow();selectCnt = 1;break;}// 超时阻塞selectint selectedKeys = selector.select(timeoutMillis);selectCnt ++;System.out.println(selectCnt);// 有事件到来 | 被唤醒 | 有内部任务 | 有定时任务时,会返回if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks() || hasScheduledTasks()) {break;}long time = System.nanoTime();if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {// 阻塞超时后没有事件到来,重置selectCntselectCnt = 1;} else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 &&selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {// Selector重建rebuildSelector();selector = this.selector;// Select again to populate selectedKeys.selector.selectNow();selectCnt = 1;break;}currentTimeNanos = time;}} catch (CancelledKeyException e) {// Harmless exception - log anyway}}
说一下Nio中臭名昭著的bug:
简单来说就是 旧版本的Java NIO在Linux Epoll实现上存在bug,(jvm)Selector.select方法可能在没有任何就绪事件的情况下返回,导致CPU空转,利用率飙升到100%
看看netty的解决策略:
1、根据该BUG的特征,首先侦测该BUG是否发生(空转计数)
2、将问题Selector上注册的Channel转移到新建的Selector上(rebuildSelector)
3、老的问题Selector关闭,使用新建的Selector替换
再看一下processSelectedKeys:
private void processSelectedKeys() {if (selectedKeys != null) {processSelectedKeysOptimized();} else {processSelectedKeysPlain(selector.selectedKeys());}}
根据selectedKeys是否为空来确定调用哪个方法
先看看:
private void processSelectedKeysOptimized() {// 迭代 selectedKeys 获取就绪的 IO 事件for (int i = 0; i < selectedKeys.size; ++i) {final SelectionKey k = selectedKeys.keys[i];// null out entry in the array to allow to have it GC'ed once the Channel close// See https://github.com/netty/netty/issues/2363// 方便GCselectedKeys.keys[i] = null;final Object a = k.attachment();if (a instanceof AbstractNioChannel) {// 为每个事件都调用 processSelectedKey 来处理它processSelectedKey(k, (AbstractNioChannel) a);} else {@SuppressWarnings("unchecked")NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;processSelectedKey(k, task);}if (needsToSelectAgain) {// null out entries in the array to allow to have it GC'ed once the Channel close// See https://github.com/netty/netty/issues/2363selectedKeys.reset(i + 1);selectAgain();i = -1;}}}
其实核心就是遍历所有就绪的IO事件,调用processSelectedKey来处理,再看看processSelectedKey函数
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();...
try {int readyOps = k.readyOps();// 连接事件if ((readyOps & SelectionKey.OP_CONNECT) != 0) {// 连接就绪事件只需要处理一次就行了,否则后续的 select()操作会一直立刻返回int ops = k.interestOps();ops &= ~SelectionKey.OP_CONNECT;k.interestOps(ops);// 最后调用 SocketChannel 的 finishConnect()方法unsafe.finishConnect();}// 可写事件if ((readyOps & SelectionKey.OP_WRITE) != 0) {// Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to writech.unsafe().forceFlush();}// 可读事件 | accepter 事件if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {unsafe.read();}} catch (CancelledKeyException ignored) {unsafe.close(unsafe.voidPromise());}}
到这里,如果看过nio selector源码的同学肯定会非常熟悉了,篇幅有限,就不详细介绍每个事件源码了,简单介绍一下事件
ACCEPT事件:连接建立好之后将该连接的channel注册到workGroup中某个NIOEventLoop的selector中
READ事件:从channel中读取数据,存放到byteBuf中,触发后续的ChannelHandler来处理数据
WRITE事件:正常情况下一般是不会注册写事件的,如果Socket发送缓冲区中没有空闲内存时,在写入会导致阻塞,此时可以注册写事件,当有空闲内存(或者可用字节数大于等于其低水位标记)时,再响应写事件,并触发对应回调
CONNECT事件:该事件是client触发的,由主动建立连接这一侧触发的
processSelectedKeysPlain方法和processSelectedKeysOptimized类似,核心都是调用processSelectedKey函数来处理,就不再展开。
最后看看runAllTasks函数,其实有两个方法,一个是有参数的,一个是没有参数的,我们来看看有参数的实现:
protected boolean runAllTasks(long timeoutNanos) {// 获取已到期的定时任务fetchFromScheduledTaskQueue();Runnable task = pollTask();if (task == null) {afterRunningAllTasks();return false;}final long deadline = ScheduledFutureTask.nanoTime() + timeoutNanos;long runTasks = 0;long lastExecutionTime;for (;;) {// 执行任务safeExecute(task);runTasks ++;// 每64个任务检查一次是否超时,相对来说nanoTime()是比较耗时的操作,所以不是每次都检查if ((runTasks & 0x3F) == 0) {lastExecutionTime = ScheduledFutureTask.nanoTime();if (lastExecutionTime >= deadline) {break;}}// 取下一个任务task = pollTask();if (task == null) {lastExecutionTime = ScheduledFutureTask.nanoTime();break;}}// 预留的扩展方法afterRunningAllTasks();this.lastExecutionTime = lastExecutionTime;return true;}
简单小结 eventLoop 的执行策略:
1、根据hasTasks() 的结果来决定是执行 selectNow() 还是 select(),如果有任务正在等待,那么应该使用无阻塞的 selectNow(),如果没有任务在等待,那么就可以使用带阻塞的 select() 操作
2、IO操作与执行任务的占比,即上述processSelectedKeys()与runAllTasks()方法的执行策略,ioRate是线程分配给 IO 操作所占的时间比(即运行 processSelectedKeys 耗时在整个循环中所占用的时间)
四、总结
1、概念回顾
一个 Selector。
一个任务队列(mpsc_queue: 多生产者单消费者 lock-free)。
一个延迟任务队列(delay_queue: 一个二叉堆结构的优先级队列,复杂度为O(log n))。
EventLoop 绑定了一个 Thread,这直接避免了pipeline 中的线程竞争。
Boss 和 Worker 共用 EventLoop 的代码逻辑,Boss 处理 accept 事件,Worker 处理 read,write 等事件。
Boss 监听并 accept 连接(channel)后以轮训的方式将 channel 交给 Worker,Worker 负责处理此 channel 后续的read/write 等 IO 事件。
在不 bind 多端口的情况下 BossEventLoopGroup 中只需要包含一个 EventLoop,也只能用上一个,多了没用。
WorkerEventLoopGroup 中一般包含多个 EventLoop,经验值一般为 cpu * 2
Channel 分两大类 ServerChannel 和 Channel,ServerChannel 对应着监听套接字(ServerSocketChannel),Channel 对应着一个网络连接。
2、源码启发
在大部分场景下,并行多线程处理可以提升系统的并发性能,我们日常工作中也是采用多线程来提升性能,使用锁来保证并发安全。
但是Netty的“无锁化”设计理念,是之前没有考虑过的另一种高性能之道。
简单对比一下两者的区别:
数据分配(多线程+锁):数据产生后统一放在数据容器中,由数据消费线程自己来获取数据进行处理,这里的获取动作需要保证是安全的,一般通过锁机制来保护,比如Java线程池中线程从阻塞队列中获取任务进行执行,就是由阻塞队列保证线程安全
数据隔离(EventLoop):数据隔离就是数据产生后就提交给不同的线程来处理,线程内部一般有一个数据容器来保存待处理的数据,这里的提交动作需要保证是安全的,比如Netty的BossGroup将建立好的连接注册到WorkerGroup时,是由内核来保证线程安全的(比如Linux就是通过epoll_ctl方法,该方法是线程安全的)
当然这两者没有说哪种策略就是最好的,软件工程没有银弹,只有更适合的场景,比如数据隔离(EventLoop),没有带来的复杂性,也没有线程切换带来的开销,但是存在一个处理数据不均衡问题。
比如每个channel都绑定了一个EventLoop,那么就肯定会存在,某个EventLoop上面所有绑定的channel多,另外一个EventLoop绑定的channel少,无法做到负载均衡,但是对于数据分配(多线程+锁)来说,就不存在这个问题。
另外还有随处可见的各种经典的设计模式的应用,API接口的封装等等就不细讲了,有兴趣可自己看看源码。
参考:
https://segmentfault.com/a/1190000007403937
https://www.javadoop.com/post/netty-part-7
https://www.cnblogs.com/cherish010/p/9044776.html
另外文中有错误的地方,请帮忙指正,多谢!




