2021-05-05 16:51  阅读(105)
文章分类:Netty 学习之旅 文章标签:NettyNetty 学习
©  原文作者:中间件兴趣圈 原文地址:https://blog.csdn.net/prestigeding/article/details/53977445

1、图说Netty线程模型

提到Netty的线程模型,我们不得不再重复提到主从Reactor线程模型,Netty线程模型基本上基于主从Reactor模型的实现方式,Netty线程模型将从如下两个图进行展开:

202105051651321051.png

Reactor 主从线程模型

202105051651326922.png

Netty服务端示例代码。

关于Reactor主从多线程模型的讲解,请重点关注Netty 线程模型前置篇,Reactor主从模式源码实现,从Netty服务端示例代码可以看出,Netty的线程模型,必然与EventLoopGroup脱离不了干系。不错,上文中的bossGroup就是相当与Reactor模式中的Main Reactor,而workGroup,则相当于SubReactor。

Netty线程模型类层次结构:

202105051651328923.png202105051651331784.png

Netty线程模型类继承图:

202105051651332525.png

从如上图可以知道,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.");
            }

来源:https://blog.csdn.net/prestigeding/article/details/53977445

点赞(0)
版权归原创作者所有,任何形式转载请联系作者; Java 技术驿站 >> 图说Netty线程模型
上一篇
线程模型前置篇Reactor反应堆设计模式实现(基于java.nio)
下一篇
源码分析ChannelPipeline实现原理