4-2~4-5 NioEventLoop的创建
Posted programmLover
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了4-2~4-5 NioEventLoop的创建相关的知识，希望对你有一定的参考价值。
一 NioEventLoop的创建大纲
1. new NioEventLoopGroup() //创建线程组，默认2*CPU
2. new ThreadPerTaskExecutor() //创建线程执行器
3. for() { newChild() } //构造NioEventLoop
4. 根据NioEventLoop数组创建Chooser //线程选择器
二 NioEventLoop的创建流程
1. 进入 MultithreadEventExecutorGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory, Object... args)
/** * Create a new instance. * * @param nThreads the number of threads that will be used by this instance. * @param executor the Executor to use, or {@code null} if the default should be used. * @param chooserFactory the {@link EventExecutorChooserFactory} to use. * @param args arguments which will passed to each {@link #newChild(Executor, Object...)} call */ protected MultithreadEventExecutorGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory, Object... args) { if (nThreads <= 0) { throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads)); } //创建 executor， 默认为 ThreadPerTaskExecutor if (executor == null) { executor = new ThreadPerTaskExecutor(newDefaultThreadFactory()); } //children 创建 EventExecutor children = new EventExecutor[nThreads]; for (int i = 0; i < nThreads; i ++) { boolean success = false; try { /** * EventLoopGroup 创建 EventLoop， NioEventLoopGroup 执行的方法为 * {@link io.netty.channel.nio.NioEventLoopGroup#newChild(java.util.concurrent.Executor, java.lang.Object...)} */ children[i] = newChild(executor, args); success = true; } catch (Exception e) { // TODO: Think about if this is a good exception type throw 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; } } } } } //创建 EventExecutorChooser chooser = chooserFactory.newChooser(children); final FutureListener<Object> terminationListener = new FutureListener<Object>() { @Override public 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); }
MultithreadEventExecutorGroup 先创建一个 ThreadPerTaskExecutor 作为 NioEventLoop 的执行器，然后调用 MultithreadEventExecutorGroup#newChild(Executor executor, Object... args) 方法创建 NioEventLoop，最后调用 EventExecutorChooserFactory#newChooser 创建 EventExecutorChooser
查看 ThreadPerTaskExecutor
public final class ThreadPerTaskExecutor implements Executor { private final ThreadFactory threadFactory; public ThreadPerTaskExecutor(ThreadFactory threadFactory) { this.threadFactory = ObjectUtil.checkNotNull(threadFactory, "threadFactory"); } @Override public void execute(Runnable command) { /** * @see DefaultThreadFactory#newThread(java.lang.Runnable, java.lang.String) */ threadFactory.newThread(command).start(); } }
从 execute 方法可以看出，ThreadPerTaskExecutor 实例每次 executor 都会创建一个新的 thread 并执行，并且其默认的 threadFactory 为 io.netty.util.concurrent.DefaultThreadFactory，这个threadFactory 创建的线程为 FastThreadLocalThread 线程，这样每个 NioEventLoop 实例调用到 execute 方法时就会创建一个新的 FastThreadLocalThread 线程并执行
进入 newChild() 方法，具体方法为 io.netty.channel.nio.NioEventLoopGroup#newChild(java.util.concurrent.Executor, java.lang.Object...)
@Override protected EventLoop newChild(Executor executor, Object... args) throws Exception { EventLoopTaskQueueFactory queueFactory = args.length == 4 ? (EventLoopTaskQueueFactory) args[3] : null; //创建并返回 NioEventLoop return new NioEventLoop(this, executor, (SelectorProvider) args[0], ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2], queueFactory); }
从上一步可以看出， 这里的 executor 为 ThreadPerTaskExecutor， args 按顺序应为 SelectorProvider、SelectStrategyFactory、RejectedExecutionHandler、EventLoopTaskQueueFactory，继续调用 NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider, SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler, EventLoopTaskQueueFactory queueFactory) 构造方法创建 NioEventLoop 并返回
进入NioEventLoop 构造器
NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider, SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler, EventLoopTaskQueueFactory queueFactory) { //创建两个非阻塞的 Queue 并继续调用父类的构造方法 super(parent, executor, false, newTaskQueue(queueFactory), newTaskQueue(queueFactory), rejectedExecutionHandler); this.provider = ObjectUtil.checkNotNull(selectorProvider, "selectorProvider"); this.selectStrategy = ObjectUtil.checkNotNull(strategy, "selectStrategy"); //selectorTuple 里保存了一个 未装饰过的 selector 和一个经过装饰器的 selector final SelectorTuple selectorTuple = openSelector(); this.selector = selectorTuple.selector; this.unwrappedSelector = selectorTuple.unwrappedSelector; }
NioEventLoop 构造器先调用 newTaskQueue(EventLoopTaskQueueFactory queueFactory) 方法创建两个非阻塞的 Queue 并继续调用父类构造方法，调用父类构造方法就是保存属性，这里不过多深入，然后保存 provider、selectStrategy、 最后调用 openSelector() 方法创建 selector 并保存
进入 newTaskQueue(EventLoopTaskQueueFactory queueFactory) 方法
private static Queue<Runnable> newTaskQueue( EventLoopTaskQueueFactory queueFactory) { if (queueFactory == null) { return newTaskQueue0(DEFAULT_MAX_PENDING_TASKS); } return queueFactory.newTaskQueue(DEFAULT_MAX_PENDING_TASKS); }
继续调用 Queue<Runnable> newTaskQueue0(int maxPendingTasks) 方法
进入 Queue<Runnable> newTaskQueue0(int maxPendingTasks) 方法
private static Queue<Runnable> newTaskQueue0(int maxPendingTasks) { // This event loop never calls takeTask() //默认创建一个 newMpscQueue return maxPendingTasks == Integer.MAX_VALUE ? PlatformDependent.<Runnable>newMpscQueue() : PlatformDependent.<Runnable>newMpscQueue(maxPendingTasks); }
Queue<Runnable> newTaskQueue0(int maxPendingTasks) 方法创建了一个 mpscQueue 并返回，mpscQueue 全称为 multiple process single consumer queue，是一个 多处理单消费并发安全的 Queue
进入 openSelector() 方法查看 Selector 创建流程
private SelectorTuple openSelector() { final Selector unwrappedSelector; try { // provider 创建一个 Nio 的 原生selector unwrappedSelector = provider.openSelector(); } catch (IOException e) { throw new ChannelException("failed to open a new selector", e); } //如果禁用 KeySet 优化，返回原生selector if (DISABLE_KEY_SET_OPTIMIZATION) { return new SelectorTuple(unwrappedSelector); } //加载并获取 sun.nio.ch.SelectorImpl 类 Object maybeSelectorImplClass = AccessController.doPrivileged(new PrivilegedAction<Object>() { @Override public Object run() { try { return Class.forName( "sun.nio.ch.SelectorImpl", false, PlatformDependent.getSystemClassLoader()); } catch (Throwable cause) { return cause; } } }); if (!(maybeSelectorImplClass instanceof Class) || // ensure the current selector implementation is what we can instrument. !((Class<?>) maybeSelectorImplClass).isAssignableFrom(unwrappedSelector.getClass())) { if (maybeSelectorImplClass instanceof Throwable) { Throwable t = (Throwable) maybeSelectorImplClass; logger.trace("failed to instrument a special java.util.Set into: {}", unwrappedSelector, t); } return new SelectorTuple(unwrappedSelector); } final Class<?> selectorImplClass = (Class<?>) maybeSelectorImplClass; //创建一个 SelectedSelectionKeySet final SelectedSelectionKeySet selectedKeySet = new SelectedSelectionKeySet(); Object maybeException = AccessController.doPrivileged(new PrivilegedAction<Object>() { @Override public Object run() { try { //获取 sun.nio.ch.SelectorImpl 类中的 selectedKeys 和 publicSelectedKeys 两个属性 Field selectedKeysField = selectorImplClass.getDeclaredField("selectedKeys"); Field publicSelectedKeysField = selectorImplClass.getDeclaredField("publicSelectedKeys"); if (PlatformDependent.javaVersion() >= 9 && PlatformDependent.hasUnsafe()) { // Let us try to use sun.misc.Unsafe to replace the SelectionKeySet. // This allows us to also do this in Java9+ without any extra flags. long selectedKeysFieldOffset = PlatformDependent.objectFieldOffset(selectedKeysField); long publicSelectedKeysFieldOffset = PlatformDependent.objectFieldOffset(publicSelectedKeysField); if (selectedKeysFieldOffset != -1 && publicSelectedKeysFieldOffset != -1) { PlatformDependent.putObject( unwrappedSelector, selectedKeysFieldOffset, selectedKeySet); PlatformDependent.putObject( unwrappedSelector, publicSelectedKeysFieldOffset, selectedKeySet); return null; } // We could not retrieve the offset, lets try reflection as last-resort. } Throwable cause = ReflectionUtil.trySetAccessible(selectedKeysField, true); if (cause != null) { return cause; } cause = ReflectionUtil.trySetAccessible(publicSelectedKeysField, true); if (cause != null) { return cause; } //通过反射将 selectedKeys 和 publicSelectedKeys 两个属性值强制设置成 selectedKeySet selectedKeysField.set(unwrappedSelector, selectedKeySet); publicSelectedKeysField.set(unwrappedSelector, selectedKeySet); return null; } catch (NoSuchFieldException e) { return e; } catch (IllegalAccessException e) { return e; } } }); if (maybeException instanceof Exception) { selectedKeys = null; Exception e = (Exception) maybeException; logger.trace("failed to instrument a special java.util.Set into: {}", unwrappedSelector, e); return new SelectorTuple(unwrappedSelector); } //本地保存 selectedKeySet，这样就不需要访问 selector 中的 selectedKeySet，直接在本地获取 selectedKeys = selectedKeySet; logger.trace("instrumented a special java.util.Set into: {}", unwrappedSelector); return new SelectorTuple(unwrappedSelector, new SelectedSelectionKeySetSelector(unwrappedSelector, selectedKeySet)); }
openSelector() 方法主要有以下几步
- 使用 provider 创建一个 Nio 原生的 selector
- 加载并获取 sun.nio.ch.SelectorImpl 类
- 创建一个 SelectedSelectionKeySet 实例 selectedKeySet
- 利用反射将原生 selector 中的 selectedKeySet、publicSelectedKeySet 替换成 selectedKeySet，这是 Netty 对 JDK Nio 的一步很重要的优化，JDK 中 SelectorImpl 的 selectedKeySet 底层是以 Set 来实现的，可能存在Hash冲突，而 Netty 自己实现的 SelectedSelectionKeySet 实例虽然叫 Set，但是底层是以数组来实现的，可以完全避免Hash 冲突
- 本地保存 selectedKeySet，这样就不需要访问 selector 中的 selectedKeySet，直接在本地获取
进入 EventExecutorChooserFactory#newChooser 方法，查看 chooser 创建流程
@Override public EventExecutorChooser newChooser(EventExecutor[] executors) { //判断 executors 的个数是不是 2 的幂，如2,4,8,16，如果是，创建 PowerOfTwoEventExecutorChooser， // 如果不是，创建GenericEventExecutorChooser if (isPowerOfTwo(executors.length)) { return new PowerOfTwoEventExecutorChooser(executors); } else { return new GenericEventExecutorChooser(executors); } }
EventExecutorChooserFactory#newChooser 方法会根据 executors 的个数是不是 2 的幂来创建 PowerOfTwoEventExecutorChooser 或 GenericEventExecutorChooser，PowerOfTwoEventExecutorChooser 会以 &运算 来计算下一个 EventExecutor，GenericEventExecutorChooser就是以常规取模运算来计算下一个 EventExecutor，这也是 Netty 的一步优化
至此 NioEventLoop 创建完毕
三 总结：
NioEventLoopGroup 会创建 nThreads 个NioEventLoop，每个 NioEventLoop 内部的 executor 都为同一个 ThreadPerTaskExecutor 实例，NioEventLoop 调用 Executor#execute 时，都会创建一个新的FastThreadLocalThread 线程实例来运行
Netty 优化点：
- netty 以自己重新实现的 SelectedSelectionKeySet 来代替了 SelectImpl 中的 selectedKeySet，使用数组方式代替 set，从而加快了 add 速度
- 如果 workGroup 指定 nThreads 长度为 2的幂次方，Netty就以与运算来代替取模原生来加快获取下一个 NioEventLoop 的速度
以上是关于4-2~4-5 NioEventLoop的创建的主要内容，如果未能解决你的问题，请参考以下文章