netty服务端启动源码分析-线程的创建
?构造方法执行过程如下：
// NioEventLoopGrouppublic NioEventLoopGroup() { this(0); }public NioEventLoopGroup(int nThreads) { this(nThreads, null); }public NioEventLoopGroup(int nThreads, ThreadFactory threadFactory) { this(nThreads, threadFactory, SelectorProvider.provider()); }public NioEventLoopGroup( int nThreads, ThreadFactory threadFactory, final SelectorProvider selectorProvider) { super(nThreads, threadFactory, selectorProvider); }
?看下父类MultithreadEventLoopGroup的构造方法
// MultithreadEventLoopGroupprotected MultithreadEventLoopGroup(int nThreads, ThreadFactory threadFactory, Object... args) { super(nThreads == 0? DEFAULT_EVENT_LOOP_THREADS : nThreads, threadFactory, args); }?注：如果没有指定创建的线程数量，则默认创建的线程个数为DEFAULT_EVENT_LOOP_THREADS，该数值为：处理器数量x2
?
再来看MultithreadEventLoopGroup的父类MultithreadEventExecutorGroup的构造方法
?
/** * Create a new instance. * * @param nThreads the number of threads that will be used by this instance. * @param threadFactory the ThreadFactory to use, or {@code null} if the default should be used. * @param args arguments which will passed to each {@link #newChild(ThreadFactory, Object...)} call */ protected MultithreadEventExecutorGroup(int nThreads, ThreadFactory threadFactory, Object... args) { if (nThreads <= 0) { throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads)); } if (threadFactory == null) { threadFactory = newDefaultThreadFactory(); } children = new SingleThreadEventExecutor[nThreads]; for (int i = 0; i < nThreads; i ++) { boolean success = false; try { children[i] = newChild(threadFactory, 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) { Thread.currentThread().interrupt(); break; } } } } } 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); } }?变量children就是用来存放创建的线程的数组，里面每一个元素都通过children[i] = newChild(threadFactory, args)创建。而newChild方法则由子类NioEventLoopGroup实现
?
// NioEventLoopGroup protected EventExecutor newChild( ThreadFactory threadFactory, Object... args) throws Exception { return new NioEventLoop(this, threadFactory, (SelectorProvider) args[0]); }?看到了吧，每个元素的真实类型为NioEventLoop，而NioEventLoop的类关系图如下
?
?（注：其实有点变态的，譬如EventLoop继承EventLoopGroup，不知道是啥原因，仅仅是因为EventLoop里有个方法parent()，返回EventLoopGroup这个功能吗？后续待确认）
?
// NioEventLoop NioEven NioEventLoop tLoop(NioEventLoopGroup parent, ThreadFactory threadFactory, SelectorProvider selectorProvider) { super(parent, threadFactory, false); if (selectorProvider == null) { throw new NullPointerException("selectorProvider"); } provider = selectorProvider; selector = openSelector(); }
?首先分析一下selector = openSelector()
?
// NioEventLoop private Selector openSelector() { final Selector selector; try { selector = provider.openSelector(); } catch (IOException e) { throw new ChannelException("failed to open a new selector", e); } if (DISABLE_KEYSET_OPTIMIZATION) { return selector; } try { SelectedSelectionKeySet selectedKeySet = new SelectedSelectionKeySet(); Class<?> selectorImplClass = Class.forName("sun.nio.ch.SelectorImpl", false, ClassLoader.getSystemClassLoader()); selectorImplClass.isAssignableFrom(selector.getClass()); Field selectedKeysField = selectorImplClass.getDeclaredField("selectedKeys"); Field publicSelectedKeysField = selectorImplClass.getDeclaredField("publicSelectedKeys"); selectedKeysField.setAccessible(true); publicSelectedKeysField.setAccessible(true); selectedKeysField.set(selector, selectedKeySet); publicSelectedKeysField.set(selector, selectedKeySet); selectedKeys = selectedKeySet; logger.trace("Instrumented an optimized java.util.Set into: {}", selector); } catch (Throwable t) { selectedKeys = null; logger.trace("Failed to instrument an optimized java.util.Set into: {}", selector, t); } return selector; }
? ? ? ?这里对sun.nio.ch.SelectorImpl中的selectedKeys和publicSelectedKeys做了优化，NioEventLoop中的变量selectedKeys的类型是SelectedSelectionKeySet，有哪些优化呢？（内部用两个数组存储？初始分配数组大小置为1024避免频繁扩容？当大小超过1024时，对数组进行双倍扩容？）。
? ? ? ?利用反射，当注册到selector中的selectionKey已准备就绪时，selectedKeys中的元素就不会为空（不再需要显示调用selector.selectedKeys()方法），后面会根据selectedKeys进行分发。
?
/** * Create a new instance * * @param parent the {@link EventExecutorGroup} which is the parent of this instance and belongs to it * @param threadFactory the {@link ThreadFactory} which will be used for the used {@link Thread} * @param addTaskWakesUp {@code true} if and only if invocation of {@link #addTask(Runnable)} will wake up the * executor thread */ protected SingleThreadEventExecutor( EventExecutorGroup parent, ThreadFactory threadFactory, boolean addTaskWakesUp) { if (threadFactory == null) { throw new NullPointerException("threadFactory"); } this.parent = parent; this.addTaskWakesUp = addTaskWakesUp; thread = threadFactory.newThread(new Runnable() { @Override public void run() { boolean success = false; updateLastExecutionTime(); try { SingleThreadEventExecutor.this.run(); success = true; } catch (Throwable t) { logger.warn("Unexpected exception from an event executor: ", t); } finally { if (state < ST_SHUTTING_DOWN) { state = ST_SHUTTING_DOWN; } // Check if confirmShutdown() was called at the end of the loop. if (success && gracefulShutdownStartTime == 0) { logger.error( "Buggy " + EventExecutor.class.getSimpleName() + " implementation; " + SingleThreadEventExecutor.class.getSimpleName() + ".confirmShutdown() must be called " + "before run() implementation terminates."); } try { // Run all remaining tasks and shutdown hooks. for (;;) { if (confirmShutdown()) { break; } } } finally { try { cleanup(); } finally { synchronized (stateLock) { state = ST_TERMINATED; } threadLock.release(); if (!taskQueue.isEmpty()) { logger.warn( "An event executor terminated with " + "non-empty task queue (" + taskQueue.size() + ')'); } terminationFuture.setSuccess(null); } } } } }); taskQueue = newTaskQueue(); } /** * Create a new {@link Queue} which will holds the tasks to execute. This default implementation will return a * {@link LinkedBlockingQueue} but if your sub-class of {@link SingleThreadEventExecutor} will not do any blocking * calls on the this {@link Queue} it may make sense to {@code @Override} this and return some more performant * implementation that does not support blocking operations at all. */ protected Queue<Runnable> newTaskQueue() { return new LinkedBlockingQueue<Runnable>(); }
EventLoopGroup bossGroup = new NioEventLoopGroup()发生了以下事情：
? ? ??1、?为NioEventLoopGroup创建数量为：处理器个数 x 2的，类型为NioEventLoop的实例。每个NioEventLoop实例 都持有一个线程，以及一个类型为LinkedBlockingQueue的任务队列
? ? ??2、线程的执行逻辑由NioEventLoop实现
? ? ? 3、每个NioEventLoop实例都持有一个selector，并对selector进行优化。
1 楼 lxzh504 2013-08-16 呵呵，这么多人浏览过就没有人评论？这么好的文章。顶一个。 2 楼 海浪儿 前天 thanks