NioEventLoop

NioEventLoop 创建

new NioEventLoopGroup 线程组,默认 2*cpu new ThreadPerTaskExecutor() 线程创建器,创建 NioEventLoop 底层的线程 循环调用 newChild() 创建 NioEventLoop chooserFactory.newChooser() 线程选择器,为每个新链接分配 NioEventLoop 线程

ThreadPerTaskExecutor

每次只需任务时都会创建一个线程实体

NioEventLoop 线程命名规则 nioEventLoop-1-xx : -1代表第几个 NioEventLoopGroup

DefaultThreadFactory

MultithreadEventExecutorGroup#MultithreadEventExecutorGroup()

protected MultithreadEventExecutorGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory, Object... args) { if (nThreads <= 0) { throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads)); } if (executor == null) { executor = new ThreadPerTaskExecutor(newDefaultThreadFactory()); // 创建 ThreadPerTaskExecutor } children = new EventExecutor[nThreads]; for (int i = 0; i < nThreads; i ++) { boolean success = false; try { children[i] = newChild(executor, args); // 创建线程池中的线程,其实创建的是 NioEventLoop 实例 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) { // 只要有一个 child 没有实例化成功,就会走这里失败处理逻辑 for (int j = 0; j < i; j ++) { // 把已经成功实例化的"线程" shutdown,shutdown 是异步操作 children[j].shutdownGracefully(); } for (int j = 0; j < i; j ++) { // 等待这些线程成功 shutdown 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; } } } } } // 设置 chooserFactory,用来实现从线程池中选择一个线程的选择策略 chooser = chooserFactory.newChooser(children); // 设置一个 Listener 用来监听该线程池的 termination 事件 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) {// 给池中每一个线程都设置这个 listener，当监听到所有线程都 terminate 以后，这个线程池就算真正的 terminate 了。 e.terminationFuture().addListener(terminationListener); } Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length); Collections.addAll(childrenSet, children); readonlyChildren = Collections.unmodifiableSet(childrenSet); }

newChild 逻辑

保存线程执行器 ThreadPerTaskExecutor 创建一个 MpscQueue 创建一个 selector

方法: io.netty.channel.nio.NioEventLoopGroup#newChild

newChild() NioEventLoopGroup.newChild() NioEventLoop构造方法:创建selector,关联NioEventLoop newTaskQueue()创建MPSC队列

创建 chooser

// 设置 chooserFactory,用来实现从线程池中选择一个线程的选择策略 chooser = chooserFactory.newChooser(children);

方法: io.netty.util.concurrent.DefaultEventExecutorChooserFactory#newChooser

public EventExecutorChooser newChooser(EventExecutor[] executors) {// 不同的线程池选择策略,选择效率会高一点 if (isPowerOfTwo(executors.length)) {// 如果线程池的线程数量是 2^n return new PowerOfTwoEventExecutorChooser(executors); } else { return new GenericEventExecutorChooser(executors);// 如果不是，用取模的方式 } }