首页 > 学院 > 开发设计 > 正文

Netty学习之旅------图说Netty线程模型

2019-11-08 00:59:12
字体:
来源:转载
供稿:网友
1、图说Netty线程模型提到Netty的线程模型,我们不得不再重复提到主从Reactor线程模型,Netty线程模型基本上基于主从Reactor模型的实现方式,Netty线程模型将从如下两个图进行展开:Reactor 主从线程模型Netty服务端示例代码关于Reactor主从多线程模型的讲解,请重点关注Netty线程模型前置篇,Reactor主从模式源码实现,从Netty服务端示例代码可以看出,Netty的线程模型,必然与EventLoopGroup脱离不了干系。不错,上文中的bossGroup就是相当与Reactor模式中的Main Reactor,而workGroup,则相当于SubReactor。Netty线程模型类层次结构:Netty线程模型类继承图:从如上图可以知道,Netty的线程模型,中有4个基础接口,它们分别是EventLoopGroup、EventLoop、EventExecuteGroup、EventExecute。其中EventExecute扩展自java.util.concurrent.ScheduledExecutorService接口,类似与线程池(执行)的职责,而EventLoop首先继承自EventExecute,并主要扩展了register方法,就是将通道Channel注册到Selector的方法。NioEventLoop,就是基于Nio的实现。在这个类中有一个亮点,就是规避了JDK nio的一个bug,Selector select方法的空轮询,核心思想是,如果连续多少次(默认为512)在没有超时的情况就返回,并且已经准备就绪的键的数量为0,则认为发生了空轮询,如果发生了空轮询,就新建一个新的Selector,并重新将通道,关心的事件注册到新的Selector,并关闭旧的Selector,其源码实现如下:
PRivate void select(boolean oldWakenUp) 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() || hasScheduledTasks()) {                    // - Selected something,                    // - waken up by user, or                    // - the task queue has a pending task.                    // - a scheduled task is ready for processing                    break;                }                if (Thread.interrupted()) {                    // Thread was interrupted so reset selected keys and break so we not run into a busy loop.                    // As this is most likely a bug in the handler of the user or it's client library we will                    // also log it.                    //                    // See https://github.com/netty/netty/issues/2426                    if (logger.isDebugEnabled()) {                        logger.debug("Selector.select() returned prematurely because " +                                "Thread.currentThread().interrupt() was called. Use " +                                "NioEventLoop.shutdownGracefully() to shutdown the NioEventLoop.");                    }                    selectCnt = 1;                    break;                }                long time = System.nanoTime();                if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {                    // timeoutMillis elapsed without anything selected.                    selectCnt = 1;                } else 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 = time;            }            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        }    }/**     * Replaces the current {@link Selector} of this event loop with newly created {@link Selector}s to work     * around the infamous epoll 100% CPU bug.     */    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.isValid() || key.channel().keyFor(newSelector) != null) {                            continue;                        }                        int interestOps = key.interestOps();                        key.cancel();                        SelectionKey newKey = key.channel().register(newSelector, interestOps, a);                        if (a instanceof AbstractNioChannel) {                            // Update SelectionKey                            ((AbstractNioChannel) a).selectionKey = newKey;                        }                        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.");    }
发表评论 共有条评论
用户名: 密码:
验证码: 匿名发表