A constructor

/**
 * 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(Executor, Object...) } call */
protected MultithreadEventExecutorGroup(int nThreads, ThreadFactory threadFactory, Object... args) {
    this(nThreads, threadFactory == null ? null : new ThreadPerTaskExecutor(threadFactory), 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 args              arguments which will passed to each {@link#newChild(Executor, Object...) } call */
protected MultithreadEventExecutorGroup(int nThreads, Executor executor, Object... args) {
    this(nThreads, executor, DefaultEventExecutorChooserFactory.INSTANCE, 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) {
    checkPositive(nThreads, "nThreads");

    /** * If the executor passed is empty, the default executor ThreadPerTaskExecutor */ is used
    if (executor == null) {
        executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
    }

    Create an EventExecutor array
    children = new EventExecutor[nThreads];

    for (int i = 0; i < nThreads; i ++) {
        boolean success = false;
        try {
            // Create an Executor 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) {// If the creation fails, all created EventExecutor is closed
                for (int j = 0; j < i; j ++) {
                    children[j].shutdownGracefully();
                }
                // Make sure all EventExecutor is closed
                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;
                    }
                }
            }
        }
    }

    // Create EventExecutor selector
    chooser = chooserFactory.newChooser(children);

    // Create a listener to listen for EventExecutor termination,
    final FutureListener<Object> terminationListener = new FutureListener<Object>() {
        @Override
        public void operationComplete(Future<Object> future) throws Exception {
            // Callback logic to notify listeners when all EventExecutor terminates by calling the Promise#setSuccess method
            if (terminatedChildren.incrementAndGet() == children.length) {
                terminationFuture.setSuccess(null); }}};for (EventExecutor e: children) {
        // Set listeners to each EventExecutor
        e.terminationFuture().addListener(terminationListener);
    }

    // Create an immutable EventExecutor array
    Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);
    Collections.addAll(childrenSet, children);
    readonlyChildren = Collections.unmodifiableSet(childrenSet);
}
Copy the code

Note (here we focus on the final constructor) :

NioEventLoopGroup

Executor implementation based on Java NIO Selector, NioEventLoop instances are created here, Selector is bound to EventLoop here, and JDK bugs using epoll are addressed.

/** * We'll just focus on the arguments to its constructor, and here we see the Selector we're looking for@param nThreads the number of threads that will be used by this instance.
 * @param executor the Executor to use, or {@code null} if default one should be used.
 * @param chooserFactory the {@link EventExecutorChooserFactory} to use.
 * @param selectorProvider the {@link SelectorProvider} to use.
 * @param selectStrategyFactory the {@link SelectStrategyFactory} to use.
 * @param rejectedExecutionHandler the {@link RejectedExecutionHandler} to use.
 * @param taskQueueFactory the {@link EventLoopTaskQueueFactory} to use for
 *                         {@link SingleThreadEventLoop#execute(Runnable)},
 *                         or {@code null} if default one should be used.
 * @param tailTaskQueueFactory the {@link EventLoopTaskQueueFactory} to use for
 *                             {@link SingleThreadEventLoop#executeAfterEventLoopIteration(Runnable)},
 *                             or {@code null} if default one should be used.
 */
public NioEventLoopGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory,
                         SelectorProvider selectorProvider,
                         SelectStrategyFactory selectStrategyFactory,
                         RejectedExecutionHandler rejectedExecutionHandler,
                         EventLoopTaskQueueFactory taskQueueFactory,
                         EventLoopTaskQueueFactory tailTaskQueueFactory) {
    super(nThreads, executor, chooserFactory, selectorProvider, selectStrategyFactory,
            rejectedExecutionHandler, taskQueueFactory, tailTaskQueueFactory);
}


/** * Create a NioEventLoop to execute the event, where the args is passed from the constructor to the parent class, which in turn calls the subclass implementation
@Override
protected EventLoop newChild(Executor executor, Object... args) throws Exception {
    SelectorProvider selectorProvider = (SelectorProvider) args[0];
    SelectStrategyFactory selectStrategyFactory = (SelectStrategyFactory) args[1];
    RejectedExecutionHandler rejectedExecutionHandler = (RejectedExecutionHandler) args[2];
    EventLoopTaskQueueFactory taskQueueFactory = null;
    EventLoopTaskQueueFactory tailTaskQueueFactory = null;

    int argsLength = args.length;
    if (argsLength > 3) {
        taskQueueFactory = (EventLoopTaskQueueFactory) args[3];
    }
    if (argsLength > 4) {
        tailTaskQueueFactory = (EventLoopTaskQueueFactory) args[4];
    }
    return new NioEventLoop(this, executor, selectorProvider,
            selectStrategyFactory.newSelectStrategy(),
            rejectedExecutionHandler, taskQueueFactory, tailTaskQueueFactory);
}


/**
 * Replaces the current {@link Selector}s of the child event loops with newly created {@linkSelector}s to work * around the infamous epoll 100%CPU bug. NioEventLoop automatically calls this method to rebuild the Selector object to fix the problem */
public void rebuildSelectors(a) {
    for (EventExecutor e: this) { ((NioEventLoop) e).rebuildSelector(); }}/**
 * Replaces the current {@link Selector} of this event loop with newly created {@linkSelector}s to work * around the infamous poll 100% CPU bug
public void rebuildSelector(a) {
    if(! inEventLoop()) { execute(new Runnable() {
            @Override
            public void run(a) { rebuildSelector0(); }});return;
    }
    rebuildSelector0();
}

Copy the code