Netty4.1源码 :NioEventLoopGroup

 NioEventLoopGroup

 

 

  NioEventLoopGroup 是一个线程池（组），聚合的N个线程（NioEventLoop）,每一个NioEventLoop启用

一条线程处理网络IO事件及任务。N个线程:默认为主机CPU个数*2

 

 register 事方法：从线程组中挑选出一个（NioEventLoop），委派其执行注册事件。

 

 

public abstract class MultithreadEventExecutorGroup extends AbstractEventExecutorGroup {

    /**
	* 实例化时，创建线程个数的EventExecutor数组，数组中实例为NioEventLoop
	* 默认为主机CPU个数*2个线程
	**/
    private final EventExecutor[] children; 
	 
	/**
	*以下两个用于停止线程池时使用
	* children中每一条线程停止后，terminatedChildren = terminatedChildren+1，当terminatedChildren==children.length说明
	* 线程都停掉了，才调用terminationFuture.setSuccess();
	**/
    private final AtomicInteger terminatedChildren = new AtomicInteger();
    private final Promise<?> terminationFuture = new DefaultPromise(GlobalEventExecutor.INSTANCE);
	
	/**
	* 默认使用chooser = PowerOfTwoEventExecutorChooser，轮询方式返回EventExecutor
	**/
    private final EventExecutorChooserFactory.EventExecutorChooser chooser; 
	
	/**
	*  通过轮询方式从EventExecutor数组中选择下一个返回
	**/
	public EventExecutor next() {
        return chooser.next();
    }
	//构建器
	protected MultithreadEventExecutorGroup(int nThreads, Executor executor,
                                            EventExecutorChooserFactory chooserFactory, Object... args) {
      

        if (executor == null) {//默认线程池，execute方法启动一个新的线程
            executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
        }

        children = new EventExecutor[nThreads];//实例化

        for (int i = 0; i < nThreads; i ++) {
            boolean success = false;
            try {
                children[i] = newChild(executor, args);//由子类NioEventLoopGroup实现,实例化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);
            }
        }

        chooser = chooserFactory.newChooser(children);//创建选择器，默认轮询方式选择 
	//停线程，调用每一个线程停止，可线程并未立即停止，当前方法返回terminationFuture，当每一个
	//线程停止后， terminationFuture.setSuccess(null);
        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);
    }

//返回下一个NioEventLoop 进行注册
    public ChannelFuture register(Channel channel) {
        return next().register(channel);
    }

   //返回下一个NioEventLoop 进行注册
    public ChannelFuture register(ChannelPromise promise) {
        return next().register(promise);
    }
	//返回下一个NioEventLoop 进行注册
    public ChannelFuture register(Channel channel, ChannelPromise promise) {
        return next().register(channel, promise);
    }