Netty4源码分析-NioEventLoop实现的线程运行逻辑
在netty服务端启动源码分析-线程创建一文中已分析SingleThreadEventExecutor所持有的线程的运行逻辑由NioEventLoop实现,那么本文就着手分析NioEventLoop实现的线程运行逻辑:
// NioEventLoop protected void run() { for (;;) { oldWakenUp = wakenUp.getAndSet(false); try { if (hasTasks()) { selectNow(); } else { select(); if (wakenUp.get()) { selector.wakeup(); } } cancelledKeys = 0; final long ioStartTime = System.nanoTime(); needsToSelectAgain = false; if (selectedKeys != null) { processSelectedKeysOptimized(selectedKeys.flip()); } else { processSelectedKeysPlain(selector.selectedKeys()); } final long ioTime = System.nanoTime() - ioStartTime; final int ioRatio = this.ioRatio; runAllTasks(ioTime * (100 - ioRatio) / ioRatio); if (isShuttingDown()) { closeAll(); if (confirmShutdown()) { break; } } } catch (Throwable t) { logger.warn("Unexpected exception in the selector loop.", t); // Prevent possible consecutive immediate failures that lead to // excessive CPU consumption. try { Thread.sleep(1000); } catch (InterruptedException e) { // Ignore. } } } }
分析如下:
- ioEventLoop执行的任务分为两大类:IO任务和非IO任务。IO任务即selectionKey中ready的事件,譬如accept、connect、read、write等;非IO任务则为添加到taskQueue中的任务,譬如之前文章中分析到的register0、bind、channelActive等任务
- 两类任务的执行先后顺序为:IO任务->非IO任务。IO任务由processSelectedKeysOptimized(selectedKeys.flip())或processSelectedKeysPlain(selector.selectedKeys())触发;非IO任务由runAllTasks(ioTime * (100 - ioRatio) / ioRatio)触发
- 两类任务的执行时间比由变量ioRatio控制,譬如:ioRatio=50(该值为默认值),则表示允许非IO任务执行的时间与IO任务的执行时间相等
- 执行IO任务前,需要先进行select,以判断之前注册过的channel是否已经有感兴趣的事件ready
-
如果任务队列中存在非IO任务,则执行非阻塞的selectNow()方法
// NioEventLoop void selectNow() throws IOException { try { selector.selectNow(); } finally { // restore wakup state if needed if (wakenUp.get()) { selector.wakeup(); } } }
否则,执行阻塞的select()方法// NioEventLoop private void select() throws IOException { Selector selector = this.selector; try { int selectCnt = 0; long currentTimeNanos = System.nanoTime(); long selectDeadLineNanos = currentTimeNanos + delayNanos(currentTimeNanos); for (;;) { long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L; if (timeoutMillis <= 0) { if (selectCnt == 0) { selector.selectNow(); selectCnt = 1; } break; } int selectedKeys = selector.select(timeoutMillis); selectCnt ++; if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks()) { // Selected something, // waken up by user, or // the task queue has a pending task. break; } if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 && selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) { // The selector returned prematurely many times in a row. // Rebuild the selector to work around the problem. logger.warn( "Selector.select() returned prematurely {} times in a row; rebuilding selector.", selectCnt); rebuildSelector(); selector = this.selector; // Select again to populate selectedKeys. selector.selectNow(); selectCnt = 1; break; } currentTimeNanos = System.nanoTime(); } if (selectCnt > MIN_PREMATURE_SELECTOR_RETURNS) { if (logger.isDebugEnabled()) { logger.debug("Selector.select() returned prematurely {} times in a row.", selectCnt - 1); } } } catch (CancelledKeyException e) { if (logger.isDebugEnabled()) { logger.debug(CancelledKeyException.class.getSimpleName() + " raised by a Selector - JDK bug?", e); } // Harmless exception - log anyway } }
下面分析阻塞的select方法:
- 首先执行delayNanos(currentTimeNanos):计算延迟任务队列中第一个任务的到期执行时间(即最晚还能延迟执行的时间).注意:(每个SingleThreadEventExecutor都持有一个延迟执行任务的优先队列:final Queue<ScheduledFutureTask<?>> delayedTaskQueue = new PriorityQueue<ScheduledFutureTask<?>>()),在启动线程的时候会往队列中加入一个任务)。最终的结果近似为:1秒钟-(当前时间-delayedTask创建的时间)。如果队列中没有任何任务,则默认返回1秒钟。
//SingleThreadEventExecutor protected long delayNanos(long currentTimeNanos) { ScheduledFutureTask<?> delayedTask = delayedTaskQueue.peek(); if (delayedTask == null) { return SCHEDULE_PURGE_INTERVAL; } return delayedTask.delayNanos(currentTimeNanos); } //ScheduledFutureTask public long delayNanos(long currentTimeNanos) { return Math.max(0, deadlineNanos() - (currentTimeNanos - START_TIME)); } public long deadlineNanos() { return deadlineNanos; }
- 如果当前时间已经超过到期执行时间后的500000纳秒(这个数字是如何定的?),则说明被延迟执行的任务不能再延迟了:如果在进入这个方法后还没有执行过selectNow方法(由标记selectCnt是否为0来判断),则先执行非阻塞的selectNow方法,然后立即返回;否则,立即返回
- 如果当前时间没有超过到期执行时间后的500000L纳秒,则说明被延迟执行的任务还可以再延迟,所以可以让select的阻塞时间长一点(说不定多出的这点时间就能select到一个ready的IO任务),故执行阻塞的selector.select(timeoutMillis)方法
- 如果已经存在ready的selectionKey,或者该selector被唤醒,或者此时非IO任务队列加入了新的任务,则立即返回
- 否则,本次执行selector.select(timeoutMillis)方法后的结果selectedKeys肯定为0,如果连续返回0的select次数还没有超过SELECTOR_AUTO_REBUILD_THRESHOLD(默认值为512),则继续下一次for循环。注意,根据以下算法:long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L。随着currentTimeNanos的增大,在进入第二次for循环时,正常情况下(即:在没有selectionKey已ready的情况下,selector.select(timeoutMillis)确实阻塞了timeoutMillis毫秒才返回0)计算出的timeoutMillis肯定小于0,计算如下:
timeoutMillis1- timeoutMillis1=0
即:timeoutMillis2 < 0。因此第二次不会再进行select,直接跳出循环并返回
- 否则,如果连续多次返回0,说明每次调用selector.select(timeoutMillis)后根本就没有阻塞timeoutMillis时间,而是立即就返回了,且结果为0. 这说明触发了epool cpu100%的bug(https://github.com/netty/netty/issues/327)。解决方案就是对selector重新rebuild
public void rebuildSelector() { if (!inEventLoop()) { execute(new Runnable() { @Override public void run() { rebuildSelector(); } }); return; } final Selector oldSelector = selector; final Selector newSelector; if (oldSelector == null) { return; } try { newSelector = openSelector(); } catch (Exception e) { logger.warn("Failed to create a new Selector.", e); return; } // Register all channels to the new Selector. int nChannels = 0; for (;;) { try { for (SelectionKey key: oldSelector.keys()) { Object a = key.attachment(); try { if (key.channel().keyFor(newSelector) != null) { continue; } int interestOps = key.interestOps(); key.cancel(); key.channel().register(newSelector, interestOps, a); nChannels ++; } catch (Exception e) { logger.warn("Failed to re-register a Channel to the new Selector.", e); if (a instanceof AbstractNioChannel) { AbstractNioChannel ch = (AbstractNioChannel) a; ch.unsafe().close(ch.unsafe().voidPromise()); } else { @SuppressWarnings("unchecked") NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a; invokeChannelUnregistered(task, key, e); } } } } catch (ConcurrentModificationException e) { // Probably due to concurrent modification of the key set. continue; } break; } selector = newSelector; try { // time to close the old selector as everything else is registered to the new one oldSelector.close(); } catch (Throwable t) { if (logger.isWarnEnabled()) { logger.warn("Failed to close the old Selector.", t); } } logger.info("Migrated " + nChannels + " channel(s) to the new Selector."); }Rebuild的本质:其实就是重新创建一个selector,然后将原来的那个selector中已注册的所有channel重新注册到新的selector中,并将老的selectionKey全部cancel掉,最后将的selector关闭。对selector进行rebuild之后,还需要重新调用selectNow方法,检查是否有已ready的selectionKey.
相关推荐
赠送jar包:transport-netty4-client-5.5.1.jar; 赠送原API文档:transport-netty4-client-5.5.1-javadoc.jar; 赠送源代码:transport-netty4-client-5.5.1-sources.jar; 赠送Maven依赖信息文件:transport-netty...
赠送jar包:transport-netty4-client-6.3.0.jar; 赠送原API文档:transport-netty4-client-6.3.0-javadoc.jar; 赠送源代码:transport-netty4-client-6.3.0-sources.jar; 赠送Maven依赖信息文件:transport-netty...
赠送jar包:transport-netty4-client-5.5.1.jar; 赠送原API文档:transport-netty4-client-5.5.1-javadoc.jar; 赠送源代码:transport-netty4-client-5.5.1-sources.jar; 赠送Maven依赖信息文件:transport-netty...
赠送jar包:reactor-netty-core-1.0.15.jar; 赠送原API文档:reactor-netty-core-1.0.15-javadoc.jar; 赠送源代码:reactor-netty-core-1.0.15-sources.jar; 赠送Maven依赖信息文件:reactor-netty-core-1.0.15....
赠送jar包:netty-transport-native-unix-common-4.1.73.Final.jar; 赠送原API文档:netty-transport-native-unix-common-4.1.73.Final-javadoc.jar; 赠送源代码:netty-transport-native-unix-common-4.1.73....
赠送jar包:netty-transport-native-unix-common-4.1.68.Final.jar; 赠送原API文档:netty-transport-native-unix-common-4.1.68.Final-javadoc.jar; 赠送源代码:netty-transport-native-unix-common-4.1.68....
赠送jar包:netty-transport-classes-epoll-4.1.73.Final.jar; 赠送原API文档:netty-transport-classes-epoll-4.1.73.Final-javadoc.jar; 赠送源代码:netty-transport-classes-epoll-4.1.73.Final-sources.jar;...
赠送jar包:netty-transport-native-unix-common-4.1.74.Final.jar; 赠送原API文档:netty-transport-native-unix-common-4.1.74.Final-javadoc.jar; 赠送源代码:netty-transport-native-unix-common-4.1.74....
netty源码netty-netty-4.1.36.Final.rarnetty源码netty-netty-4.1.36.Final.rarnetty源码netty-netty-4.1.36.Final.rarnetty源码netty-netty-4.1.36.Final.rarnetty源码netty-netty-4.1.36.Final.rarnetty源码netty-...
赠送jar包:transport-netty4-client-6.3.0.jar; 赠送原API文档:transport-netty4-client-6.3.0-javadoc.jar; 赠送源代码:transport-netty4-client-6.3.0-sources.jar; 赠送Maven依赖信息文件:transport-netty...
赠送jar包:netty-codec-mqtt-4.1.73.Final.jar; 赠送原API文档:netty-codec-mqtt-4.1.73.Final-javadoc.jar; 赠送源代码:netty-codec-mqtt-4.1.73.Final-sources.jar; 赠送Maven依赖信息文件:netty-codec-...
赠送jar包:netty-transport-classes-epoll-4.1.74.Final.jar; 赠送原API文档:netty-transport-classes-epoll-4.1.74.Final-javadoc.jar; 赠送源代码:netty-transport-classes-epoll-4.1.74.Final-sources.jar;...
赠送jar包:netty-transport-classes-epoll-4.1.73.Final.jar; 赠送原API文档:netty-transport-classes-epoll-4.1.73.Final-javadoc.jar; 赠送源代码:netty-transport-classes-epoll-4.1.73.Final-sources.jar;...
赠送jar包:reactor-netty-http-1.0.11.jar; 赠送原API文档:reactor-netty-http-1.0.11-javadoc.jar; 赠送源代码:reactor-netty-http-1.0.11-sources.jar; 赠送Maven依赖信息文件:reactor-netty-...
赠送jar包:netty-resolver-dns-4.1.65.Final.jar; 赠送原API文档:netty-resolver-dns-4.1.65.Final-javadoc.jar; 赠送源代码:netty-resolver-dns-4.1.65.Final-sources.jar; 赠送Maven依赖信息文件:netty-...
netty-socketio-netty-socketio-2.0.6 ,Socket.IO 是一个库,可以在客户端和服务器之间实现低延迟, 双向和基于事件的通信:netty-socketio-netty-socketio-2.0.6.tar.gznetty-socketio-netty-socketio-2.0.6.zip
赠送jar包:transport-netty4-client-6.2.3.jar; 赠送原API文档:transport-netty4-client-6.2.3-javadoc.jar; 赠送源代码:transport-netty4-client-6.2.3-sources.jar; 赠送Maven依赖信息文件:transport-netty...