Netty源码

NioEventLoopGroup

初始化NioEventLoopGroup时若未传入线程数，在调用超类MultithreadEventLoopGroup构造方法时使用CPU核数的两倍作为线程数。根据线程数创建EventExecutor数组，然后调用子类NioEventLoopGroup的newChild方法从而调用NioEventLoop的构造方法对EventExecutor数组中的每个元素实例化，在NioEventLoop的构造方法中调用openSelector()方法创建Selector，相当于NIO中Selector.open()。且在NioEventLoop超类SingleThreadEventExecutor中实例化了taskQueue任务队列。

对于chooser选择器的生成是调用的DefaultEventExecutorChooserFactory.INSTANCE的newChooser方法，通过isPowerOfTwo方法判断工作组的数量即Selector是否为2的整数次幂，然后创建具体的选择器，其实PowerOfTwoEventExecutorChooser和GenericEventExecutorChooser实现的逻辑其实是一样的都是将工作组当成一个环形依次取区中的元素，前者由于是2的整数次幂使用的是取模的方式，后者使用的是取余的方式。

public class NioEventLoopGroup extends MultithreadEventLoopGroup {
    public NioEventLoopGroup() {
        this(0);
    }
    public NioEventLoopGroup(int nThreads) {
        this(nThreads, (Executor) null);
    }
    public NioEventLoopGroup(int nThreads, Executor executor) {
        this(nThreads, executor, SelectorProvider.provider());
    }
    public NioEventLoopGroup(int nThreads, Executor executor, final SelectorProvider selectorProvider) {
        this(nThreads, executor, selectorProvider, DefaultSelectStrategyFactory.INSTANCE);
    }
    protected EventLoop newChild(Executor executor, Object... args) throws Exception {
        return new NioEventLoop(this, executor, (SelectorProvider) args[0], ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]);
    }
}
public abstract class MultithreadEventLoopGroup extends MultithreadEventExecutorGroup implements EventLoopGroup {
    private static final int DEFAULT_EVENT_LOOP_THREADS;
    static { // DEFAULT_EVENT_LOOP_THREADS默认为CPU核数的两倍
        DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt("io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));
    }
    protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {
        super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
    }
}
public abstract class MultithreadEventExecutorGroup extends AbstractEventExecutorGroup {
    protected MultithreadEventExecutorGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory, Object... args) {
        checkPositive(nThreads, "nThreads");
        if (executor == null) {
            executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
        }
        children = new EventExecutor[nThreads];
        for (int i = 0; i < nThreads; i ++) {
            boolean success = false;
            try {
                children[i] = newChild(executor, args);
                success = true;
            } catch (Exception e) {
                throw new IllegalStateException("failed to create a child event loop", e);
            }
        }
        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);
    }
}
public final class NioEventLoop extends SingleThreadEventLoop {
    NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider, SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler) {
        super(parent, executor, false, DEFAULT_MAX_PENDING_TASKS, rejectedExecutionHandler);
        if (selectorProvider == null) {
            throw new NullPointerException("selectorProvider");
        }
        if (strategy == null) {
            throw new NullPointerException("selectStrategy");
        }
        provider = selectorProvider;
        final SelectorTuple selectorTuple = openSelector(); // 创建一个选择器selector，相当于NIO中的Selector.open()
        selector = selectorTuple.selector;
        unwrappedSelector = selectorTuple.unwrappedSelector;
        selectStrategy = strategy;
    }
}

ServerBootstrap

使用时首先创建两个EventLoopGroup，然后创建一个ServerBootstrap通过channel方法设置NioServerSocketChannel作为服务器的通道实现，然后通过childHandler方法给ChannelPipeline添加Handler，最后调用bind方法绑定端口。

EventLoopGroup bossGroup = new NioEventLoopGroup(3);
EventLoopGroup workerGroup = new NioEventLoopGroup(8);
ServerBootstrap bootstrap = new ServerBootstrap(); // 创建服务器端的启动对象
// 使用链式编程来配置参数
bootstrap.group(bossGroup, workerGroup) //设置两个线程组
    .channel(NioServerSocketChannel.class) // 使用NioServerSocketChannel作为服务器的通道实现
    .option(ChannelOption.SO_BACKLOG, 1024)
    .childHandler(new ChannelInitializer<SocketChannel>() {//创建通道初始化对象，设置初始化参数，在SocketChannel建立起来之前执行
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline().addLast(new NettyServerHandler()); // 对workerGroup的SocketChannel设置处理器
        }
    });
ChannelFuture cf = bootstrap.bind(9000).sync();

调用ServerBootstrap的channel方法最终调用超类AbstractBootstrap的channel，在该方法中将传入的NioServerSocketChannel封装到ReflectiveChannelFactory中，在将该对象设置给channelFactory属性，后续使用时直接调用ReflectiveChannelFactory的newChannel方法从而调用NioServerSocketChannel的构造方法实例化。

NioServerSocketChannel构造方法中首先调用newSocket方法，调用SelectorProvider的openServerSocketChannel方法创建一个在本地端口进行监听的服务Socket通道，相当于NIO中调用ServerSocketChannel.open()方法。然后调用超类的构造方法初始化ChannelPipeline，且将NIO的ServerSocketChannel设置为非阻塞模式，且设置readInterestOp为对客户端accept连接操作感兴趣。

public abstract class AbstractBootstrap<B extends AbstractBootstrap<B, C>, C extends Channel> implements Cloneable {
    public B channel(Class<? extends C> channelClass) {
        if (channelClass == null) {
            throw new NullPointerException("channelClass");
        }
        return channelFactory(new ReflectiveChannelFactory<C>(channelClass));
    }
    public B channelFactory(ChannelFactory<? extends C> channelFactory) {
        if (channelFactory == null) {
            throw new NullPointerException("channelFactory");
        }
        if (this.channelFactory != null) {
            throw new IllegalStateException("channelFactory set already");
        }
        this.channelFactory = channelFactory;
        return self();
    }
}
public class ReflectiveChannelFactory<T extends Channel> implements ChannelFactory<T> {
    private final Constructor<? extends T> constructor;
    public ReflectiveChannelFactory(Class<? extends T> clazz) {
        ObjectUtil.checkNotNull(clazz, "clazz");
        try {
            this.constructor = clazz.getConstructor();
        } catch (NoSuchMethodException e) {
            throw new IllegalArgumentException("Class " + StringUtil.simpleClassName(clazz) + " does not have a public non-arg constructor", e);
        }
    }
    @Override
    public T newChannel() {
        try {
            return constructor.newInstance();
        } catch (Throwable t) {
            throw new ChannelException("Unable to create Channel from class " + constructor.getDeclaringClass(), t);
        }
    }
}
public class NioServerSocketChannel extends AbstractNioMessageChannel implements io.netty.channel.socket.ServerSocketChannel {
    private static final SelectorProvider DEFAULT_SELECTOR_PROVIDER = SelectorProvider.provider();
    public NioServerSocketChannel() {
        this(newSocket(DEFAULT_SELECTOR_PROVIDER));
    }
    private static ServerSocketChannel newSocket(SelectorProvider provider) {
        try { // 调用NIO的ServerSocketChannel.open()方法，创建一个在本地端口进行监听的服务Socket通道
            return provider.openServerSocketChannel();
        } catch (IOException e) {
            throw new ChannelException("Failed to open a server socket.", e);
        }
    }
    public NioServerSocketChannel(ServerSocketChannel channel) {
        super(null, channel, SelectionKey.OP_ACCEPT); // 对客户端accept连接操作感兴趣
        config = new NioServerSocketChannelConfig(this, javaChannel().socket());
    }
}
public abstract class AbstractNioMessageChannel extends AbstractNioChannel {
    protected AbstractNioMessageChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
        super(parent, ch, readInterestOp);
    }
}
public abstract class AbstractNioChannel extends AbstractChannel {
    protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
        super(parent); // 初始化ChannelPipeline
        this.ch = ch;
        this.readInterestOp = readInterestOp;
        try { // 将NIO的ServerSocketChannel设置为非阻塞模式
            ch.configureBlocking(false); 
        } catch (IOException e) {
            try {
                ch.close();
            } catch (IOException e2) {
            }
            throw new ChannelException("Failed to enter non-blocking mode.", e);
        }
    }
}
public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {
    protected AbstractChannel(Channel parent) {
        this.parent = parent;
        id = newId();
        unsafe = newUnsafe();
        pipeline = newChannelPipeline(); // 初始化ChannelPipeline
    }
    protected DefaultChannelPipeline newChannelPipeline() {
        return new DefaultChannelPipeline(this); // 实例化ChannelPipeline
    }
}
public class DefaultChannelPipeline implements ChannelPipeline {
    protected DefaultChannelPipeline(Channel channel) {
        this.channel = ObjectUtil.checkNotNull(channel, "channel");
        succeededFuture = new SucceededChannelFuture(channel, null);
        voidPromise =  new VoidChannelPromise(channel, true);
        tail = new TailContext(this); // 创建ChannelPipeline的尾节点
        head = new HeadContext(this); // 创建ChannelPipeline的头节点
        head.next = tail; // 将head头节点的next指向tail尾节点
        tail.prev = head; // 将tail尾节点的prev指向head头节点
    }
}

调用ServerBootstrap的bind方法最终调用超类AbstractBootstrap的bind，然后调用调用doBind方法，在initAndRegister方法首先调用ChannelFactory的newChannel方式，即调用ReflectiveChannelFactory的newChannel方法实例化NioServerSocketChannel对客户端accept连接操作感兴趣，且初始化ChannelPipeline。然后通过init方法给NioServerSocketChannel设置TCP参数，且将用户自定义的ChannelPipeline封装成ServerBootstrapAcceptor添加到ChannelPipeline。

public abstract class AbstractBootstrap<B extends AbstractBootstrap<B, C>, C extends Channel> implements Cloneable {
    public ChannelFuture bind(int inetPort) {
        return bind(new InetSocketAddress(inetPort));
    }
    public ChannelFuture bind(SocketAddress localAddress) {
        validate();
        if (localAddress == null) {
            throw new NullPointerException("localAddress");
        }
        return doBind(localAddress);
    }
    public ChannelFuture bind(SocketAddress localAddress) {
        validate();
        if (localAddress == null) {
            throw new NullPointerException("localAddress");
        }
        return doBind(localAddress);
    }
    private ChannelFuture doBind(final SocketAddress localAddress) {
        final ChannelFuture regFuture = initAndRegister();
        final Channel channel = regFuture.channel();
        if (regFuture.cause() != null) {
            return regFuture;
        }
        if (regFuture.isDone()) {
            ChannelPromise promise = channel.newPromise();
            doBind0(regFuture, channel, localAddress, promise);
            return promise;
        } else {
            final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
            regFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    Throwable cause = future.cause();
                    if (cause != null) {
                        promise.setFailure(cause);
                    } else {
                        promise.registered();
                        doBind0(regFuture, channel, localAddress, promise);
                    }
                }
            });
            return promise;
        }
    }
    final ChannelFuture initAndRegister() {
        Channel channel = null; // 实际类型为NioServerSocketChannel
        try { // 实例化NioServerSocketChannel并初始化ChannelPipeline
            channel = channelFactory.newChannel(); // 这里就是调用前面ReflectiveChannelFactory的newChannel方法实例化NioServerSocketChannel
            init(channel); // 给NioServerSocketChannel设置options参数，将用户自定义的ChannelPipeline封装成ServerBootstrapAcceptor添加到ChannelPipeline
        } catch (Throwable t) {
            if (channel != null) {
                channel.unsafe().closeForcibly();
                return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t);
            }
            return new DefaultChannelPromise(new FailedChannel(), GlobalEventExecutor.INSTANCE).setFailure(t);
        }
        // 将ServerSocketChannel注册到BossGroup中的一个线程的Selector上
        ChannelFuture regFuture = config().group().register(channel);
        if (regFuture.cause() != null) {
            if (channel.isRegistered()) {
                channel.close();
            } else {
                channel.unsafe().closeForcibly();
            }
        }
        return regFuture;
    }
}

AbstractBootstrap中的init方法会调用子类ServerBootstrap的init方法，首先设置NioServerSocketChannel的参数然后获取其中初始化好的ChannelPipeline，然后获取Worker Group和ServerBootstrap的childHandler方法设置的回调方法将其封装到ServerBootstrapAcceptor中，当有连接事件时调用channelRead方法将用户自定义的childHandler添加到ChannelPipeline中，且将连接过来的SocketChannel注册到WorkGroup中的一个线程的Selector上，和ServerSocketChannel注册到BossGroup走的同一个逻辑。

public class ServerBootstrap extends AbstractBootstrap<ServerBootstrap, ServerChannel> {
    void init(Channel channel) throws Exception {
        final Map<ChannelOption<?>, Object> options = options0();
        synchronized (options) {
            setChannelOptions(channel, options, logger);
        }
        final Map<AttributeKey<?>, Object> attrs = attrs0();
        synchronized (attrs) {
            for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) {
                @SuppressWarnings("unchecked")
                AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();
                channel.attr(key).set(e.getValue());
            }
        }
        ChannelPipeline p = channel.pipeline(); // 获取到在NioServerSocketChannel中初始化的ChannelPipeline
        final EventLoopGroup currentChildGroup = childGroup; // 即Worker Group
        final ChannelHandler currentChildHandler = childHandler; // ServerBootstrap的childHandler方法设置的回调方法
        final Entry<ChannelOption<?>, Object>[] currentChildOptions;
        final Entry<AttributeKey<?>, Object>[] currentChildAttrs;
        synchronized (childOptions) {
            currentChildOptions = childOptions.entrySet().toArray(newOptionArray(0));
        }
        synchronized (childAttrs) {
            currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(0));
        }
        p.addLast(new ChannelInitializer<Channel>() {
            @Override
            public void initChannel(final Channel ch) throws Exception {
                final ChannelPipeline pipeline = ch.pipeline();
                ChannelHandler handler = config.handler();
                if (handler != null) {
                    pipeline.addLast(handler);
                }
                ch.eventLoop().execute(new Runnable() {
                    @Override
                    public void run() {
                        pipeline.addLast(new ServerBootstrapAcceptor(ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                    }
                });
            }
        });
    }
}
private static class ServerBootstrapAcceptor extends ChannelInboundHandlerAdapter {
    private final EventLoopGroup childGroup;
    private final ChannelHandler childHandler;
    private final Entry<ChannelOption<?>, Object>[] childOptions;
    private final Entry<AttributeKey<?>, Object>[] childAttrs;
    private final Runnable enableAutoReadTask;
    ServerBootstrapAcceptor(final Channel channel, EventLoopGroup childGroup, ChannelHandler childHandler,
        Entry<ChannelOption<?>, Object>[] childOptions, Entry<AttributeKey<?>, Object>[] childAttrs) {
        this.childGroup = childGroup;
        this.childHandler = childHandler;
        this.childOptions = childOptions;
        this.childAttrs = childAttrs;
        enableAutoReadTask = new Runnable() {
            @Override
            public void run() {
                channel.config().setAutoRead(true);
            }
        };
    }
    @Override
    @SuppressWarnings("unchecked")
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // 当有连接事件时调用channelRead方法将用户自定义的childHandler添加到ChannelPipeline中
        final Channel child = (Channel) msg;
        child.pipeline().addLast(childHandler);
        setChannelOptions(child, childOptions, logger);
        for (Entry<AttributeKey<?>, Object> e: childAttrs) {
            child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());
        }
        try { // 将连接过来的SocketChannel注册到WorkGroup中的一个线程的Selector上，和ServerSocketChannel注册到BossGroup走的同一个逻辑
            // 这里调用的childGroup的register方法，实际是调用MultithreadEventLoopGroup的register方法
            childGroup.register(child).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (!future.isSuccess()) {
                        forceClose(child, future.cause());
                    }
                }
            });
        } catch (Throwable t) {
            forceClose(child, t);
        }
    }
}

这里的next方法最终调用前面创建的PowerOfTwoEventExecutorChooser或GenericEventExecutorChooser的next方法从工作组中选择一个NioEventLoop来执行register方法。最终异步调用AbstractChannel的register0方法。首先调用AbstractNioChannel的doRegister方法将ServerSocketChannel注册到Selector上，相当于NIO中调用ServerSocketChannel的register方法。然后通过SingleThreadEventExecutor的execute方法将该任务添加到taskQueue异步执行AbstractChannel的register0方法。

public abstract class MultithreadEventLoopGroup extends MultithreadEventExecutorGroup implements EventLoopGroup {
    public ChannelFuture register(Channel channel) {
        // next方法最终调用前面创建的PowerOfTwoEventExecutorChooser或GenericEventExecutorChooser的next方法从工作组中选择一个NioEventLoop来执行register方法
        return next().register(channel);
    }
}
public abstract class SingleThreadEventLoop extends SingleThreadEventExecutor implements EventLoop {
    public ChannelFuture register(Channel channel) {
        return register(new DefaultChannelPromise(channel, this));
    }
}
public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {
    public final void register(EventLoop eventLoop, final ChannelPromise promise) {
        if (eventLoop == null) {
            throw new NullPointerException("eventLoop");
        }
        if (isRegistered()) {
            promise.setFailure(new IllegalStateException("registered to an event loop already"));
            return;
        }
        if (!isCompatible(eventLoop)) {
            promise.setFailure(new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));
            return;
        }
        AbstractChannel.this.eventLoop = eventLoop;
        if (eventLoop.inEventLoop()) {
            register0(promise);
        } else {
            try { // 通过SingleThreadEventExecutor将该任务添加到taskQueue异步执行
                eventLoop.execute(new Runnable() {
                    @Override
                    public void run() {
                        // 将ServerSocketChannel注册到Selector上，相当于NIO中调用ServerSocketChannel的register方法
                        register0(promise);
                    }
                });
            } catch (Throwable t) {
                closeForcibly();
                closeFuture.setClosed();
                safeSetFailure(promise, t);
            }
        }
    }
    private void register0(ChannelPromise promise) {
        try {
            if (!promise.setUncancellable() || !ensureOpen(promise)) {
                return;
            }
            boolean firstRegistration = neverRegistered;
            doRegister(); // 调用AbstractNioChannel把ServerSocketChannel注册到selector上
            neverRegistered = false;
            registered = true;
            // 实际调用DefaultChannelPipeline的invokeHandlerAddedIfNeeded方法，最终调用ServerBootstrap的init方法中添加的ChannelInitializer，真正将ServerBootstrapAcceptor添加到ChannelPipeline
            pipeline.invokeHandlerAddedIfNeeded(); 
            safeSetSuccess(promise);
            pipeline.fireChannelRegistered(); // 调用pipeline中每个handler的channelRegistered方法
            if (isActive()) {
                if (firstRegistration) {
                    pipeline.fireChannelActive();
                } else if (config().isAutoRead()) {
                    beginRead();
                }
            }
        } catch (Throwable t) {
            closeForcibly();
            closeFuture.setClosed();
            safeSetFailure(promise, t);
        }
    }
}
public abstract class AbstractNioChannel extends AbstractChannel {
    protected void doRegister() throws Exception {
        boolean selected = false;
        for (;;) {
            try { // 相当于NIO中调用ServerSocketChannel的register方法，把ServerSocketChannel注册到selector上
                selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
                return;
            } catch (CancelledKeyException e) {
                if (!selected) {
                    eventLoop().selectNow();
                    selected = true;
                } else {
                    throw e;
                }
            }
        }
    }
}

调用DefaultChannelPipeline的invokeHandlerAddedIfNeeded方法，最终调用ServerBootstrap的init方法中添加的ChannelInitializer，真正将ServerBootstrapAcceptor添加到ChannelPipeline。

public class DefaultChannelPipeline implements ChannelPipeline {
    final void invokeHandlerAddedIfNeeded() {
        assert channel.eventLoop().inEventLoop();
        if (firstRegistration) {
            firstRegistration = false;
            callHandlerAddedForAllHandlers();
        }
    }
    private void callHandlerAddedForAllHandlers() {
        final PendingHandlerCallback pendingHandlerCallbackHead;
        synchronized (this) {
            assert !registered;
            registered = true;
            pendingHandlerCallbackHead = this.pendingHandlerCallbackHead;
            this.pendingHandlerCallbackHead = null;
        }
        PendingHandlerCallback task = pendingHandlerCallbackHead;
        while (task != null) {
            task.execute();
            task = task.next;
        }
    }
    private void callHandlerAdded0(final AbstractChannelHandlerContext ctx) {
        try {
            ctx.callHandlerAdded();
        } catch (Throwable t) {
            boolean removed = false;
            try {
                remove0(ctx);
                ctx.callHandlerRemoved();
                removed = true;
            } catch (Throwable t2) {
            }
            if (removed) {
                fireExceptionCaught(new ChannelPipelineException(ctx.handler().getClass().getName() + ".handlerAdded() has thrown an exception; removed.", t));
            } else {
                fireExceptionCaught(new ChannelPipelineException(ctx.handler().getClass().getName() + ".handlerAdded() has thrown an exception; also failed to remove.", t));
            }
        }
    }
}
private final class PendingHandlerAddedTask extends PendingHandlerCallback {
    PendingHandlerAddedTask(AbstractChannelHandlerContext ctx) {
        super(ctx);
    }
    @Override
    public void run() {
        callHandlerAdded0(ctx);
    }
    @Override
    void execute() {
        EventExecutor executor = ctx.executor();
        if (executor.inEventLoop()) {
            callHandlerAdded0(ctx);
        } else {
            try {
                executor.execute(this);
            } catch (RejectedExecutionException e) {
                remove0(ctx);
                ctx.setRemoved();
            }
        }
    }
}
abstract class AbstractChannelHandlerContext implements ChannelHandlerContext, ResourceLeakHint {
    final void callHandlerAdded() throws Exception {
        if (setAddComplete()) {
            handler().handlerAdded(this);
        }
    }
}
public abstract class ChannelInitializer<C extends Channel> extends ChannelInboundHandlerAdapter {
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        if (ctx.channel().isRegistered()) {
            if (initChannel(ctx)) {
                removeState(ctx);
            }
        }
    }
    private boolean initChannel(ChannelHandlerContext ctx) throws Exception {
        if (initMap.add(ctx)) {
            try { // 调用ServerBootstrap的init方法中添加的ChannelInitializer，真正将ServerBootstrapAcceptor添加到ChannelPipeline
                initChannel((C) ctx.channel());
            } catch (Throwable cause) {
                exceptionCaught(ctx, cause);
            } finally {
                ChannelPipeline pipeline = ctx.pipeline();
                if (pipeline.context(this) != null) {
                    pipeline.remove(this);
                }
            }
            return true;
        }
        return false;
    }
}

调用SingleThreadEventExecutor的execute方法首先将任务添加到队列taskQueue队列中，然后调用startThread方法最终调用SingleThreadEventExecutor的子类NioEventLoop的run方法。首先判断taskQueue队列中是否有任务需要处理，若没有则在NioEventLoop的select方法中调用Selector的select方法阻塞监听IO事件，且通过selectCnt变量来解决空轮训CPU100%的问题，若是空轮训selectCnt会一直++，当该数值大于等于默认512时，将重新创建Selector然后替换当前的Selector。

public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {
    public void execute(Runnable task) {
        if (task == null) {
            throw new NullPointerException("task");
        }
        boolean inEventLoop = inEventLoop();
        addTask(task); // 将任务添加到队列taskQueue队列中
        if (!inEventLoop) {
            startThread();
            if (isShutdown()) {
                boolean reject = false;
                try {
                    if (removeTask(task)) {
                        reject = true;
                    }
                } catch (UnsupportedOperationException e) {
                }
                if (reject) {
                    reject();
                }
            }
        }
        if (!addTaskWakesUp && wakesUpForTask(task)) {
            wakeup(inEventLoop);
        }
    }
    private void startThread() {
        if (state == ST_NOT_STARTED) {
            if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {
                try {
                    doStartThread();
                } catch (Throwable cause) {
                    STATE_UPDATER.set(this, ST_NOT_STARTED);
                    PlatformDependent.throwException(cause);
                }
            }
        }
    }
    private void doStartThread() {
        assert thread == null;
        executor.execute(new Runnable() {
            @Override
            public void run() {
                thread = Thread.currentThread();
                if (interrupted) {
                    thread.interrupt();
                }
                boolean success = false;
                updateLastExecutionTime();
                try { // 调用子类NioEventLoop的run方法
                    SingleThreadEventExecutor.this.run();
                    success = true;
                } catch (Throwable t) {
                }
            }
        });
    }
}
public final class NioEventLoop extends SingleThreadEventLoop {
    protected void run() {
        for (;;) {
            try {
                try {
                    switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
                        case SelectStrategy.CONTINUE:
                            continue;
                        case SelectStrategy.BUSY_WAIT:
                        case SelectStrategy.SELECT: // 若队列中没有任务
                            select(wakenUp.getAndSet(false)); // 监听IO事件
                            if (wakenUp.get()) {
                                selector.wakeup();
                            }
                        default:
                    }
                } catch (IOException e) {
                    rebuildSelector0();
                    handleLoopException(e);
                    continue;
                }
                cancelledKeys = 0;
                needsToSelectAgain = false;
                final int ioRatio = this.ioRatio;
                if (ioRatio == 100) {
                    try {
                        processSelectedKeys();
                    } finally { // Ensure we always run tasks.
                        runAllTasks();
                    }
                } else {
                    final long ioStartTime = System.nanoTime();
                    try {
                        processSelectedKeys();
                    } finally { // Ensure we always run tasks.
                        final long ioTime = System.nanoTime() - ioStartTime;
                        runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
                    }
                }
            } catch (Throwable t) {
                handleLoopException(t);
            }
            try { // Always handle shutdown even if the loop processing threw an exception.
                if (isShuttingDown()) {
                    closeAll();
                    if (confirmShutdown()) {
                        return;
                    }
                }
            } catch (Throwable t) {
                handleLoopException(t);
            }
        }
    }
    private void select(boolean oldWakenUp) throws IOException {
        Selector selector = this.selector;
        try {
            int selectCnt = 0;
            long currentTimeNanos = System.nanoTime();
            long selectDeadLineNanos = currentTimeNanos + delayNanos(currentTimeNanos);
            for (;;) {
                long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L;
                if (timeoutMillis <= 0) {
                    if (selectCnt == 0) {
                        selector.selectNow();
                        selectCnt = 1;
                    }
                    break;
                }
                if (hasTasks() && wakenUp.compareAndSet(false, true)) {
                    selector.selectNow();
                    selectCnt = 1;
                    break;
                }
                int selectedKeys = selector.select(timeoutMillis); // 当timeoutMillis超时或有事件发生会break处理
                selectCnt ++; // 用于解决空轮训CPU100%的问题
                if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks() || hasScheduledTasks()) {
                    break;
                }
                if (Thread.interrupted()) {
                    selectCnt = 1;
                    break;
                }
                long time = System.nanoTime();
                if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {
                    selectCnt = 1;
                } else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 && selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) { // 用于解决空轮训CPU100%的问题
                    selector = selectRebuildSelector(selectCnt); // 重新构建替换Selector
                    selectCnt = 1;
                    break;
                }
                currentTimeNanos = time;
            }
        } catch (CancelledKeyException e) {
        }
    }
}
public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {
    protected boolean runAllTasks(long timeoutNanos) {
        fetchFromScheduledTaskQueue();
        Runnable task = pollTask();
        if (task == null) {
            afterRunningAllTasks();
            return false;
        }
        final long deadline = ScheduledFutureTask.nanoTime() + timeoutNanos;
        long runTasks = 0;
        long lastExecutionTime;
        for (;;) {
            safeExecute(task);
            runTasks ++;
            if ((runTasks & 0x3F) == 0) {
                lastExecutionTime = ScheduledFutureTask.nanoTime();
                if (lastExecutionTime >= deadline) {
                    break;
                }
            }

            task = pollTask();
            if (task == null) {
                lastExecutionTime = ScheduledFutureTask.nanoTime();
                break;
            }
        }
        afterRunningAllTasks();
        this.lastExecutionTime = lastExecutionTime;
        return true;
    }
}
public abstract class AbstractEventExecutor extends AbstractExecutorService implements EventExecutor {
    protected static void safeExecute(Runnable task) {
        try {
            task.run();
        } catch (Throwable t) {
        }
    }
}

当有事件需要处理时调用NioEventLoop的processSelectedKeys方法，若是客户端的连接事件最终调用AbstractNioMessageChannel的NioMessageUnsafe的read方法处理客户端发送数据和连接请求事件。然后调用NioServerSocketChannel的doReadMessages方法获取SocketChannel，然后将SocketChannel包装为NioSocketChannel，且初始化ChannelPipeline将阻塞模式设置为非阻塞。然后调用ChannelPipeline的fireChannelRead方法，从而调用到前面封装用户自定义的childHandler的ServerBootstrapAcceptor的channelRead方法，将用户自定义的childHandler从BossGroup添加到WorkGroup，且将当前SocketChannel注册到WorkGroup中的一个Selector中。

值得注意的是将SocketChannel包装为NioSocketChannel时调用的超类AbstractNioByteChannel的构造方法中给SocketChannel绑定的感兴趣的时间为OP_READ事件，而前面NioServerSocketChannel构造方法中绑定的是OP_ACCEPT事件，从而将连接事件交给BossGroup处理，将发送数据的读写事件交给WorkGroup处理。且BossGroup和WorkGroup处理事件的逻辑是同一套逻辑。

public final class NioEventLoop extends SingleThreadEventLoop {
    private void processSelectedKeys() {
        if (selectedKeys != null) {
            processSelectedKeysOptimized();
        } else {
            processSelectedKeysPlain(selector.selectedKeys());
        }
    }
    private void processSelectedKeysOptimized() {
        for (int i = 0; i < selectedKeys.size; ++i) { // 循环处理selectedKeys中所有的key
            final SelectionKey k = selectedKeys.keys[i];
            selectedKeys.keys[i] = null;
            final Object a = k.attachment();
            if (a instanceof AbstractNioChannel) {
                processSelectedKey(k, (AbstractNioChannel) a);
            } else {
                NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
                processSelectedKey(k, task);
            }
            if (needsToSelectAgain) {
                selectedKeys.reset(i + 1);
                selectAgain();
                i = -1;
            }
        }
    }
    private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
        final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
        if (!k.isValid()) {
            final EventLoop eventLoop;
            try {
                eventLoop = ch.eventLoop();
            } catch (Throwable ignored) {
                return;
            }
            if (eventLoop != this || eventLoop == null) {
                return;
            }
            unsafe.close(unsafe.voidPromise());
            return;
        }
        try {
            int readyOps = k.readyOps();
            if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
                int ops = k.interestOps();
                ops &= ~SelectionKey.OP_CONNECT; // 移除OP_CONNECT事件
                k.interestOps(ops);
                unsafe.finishConnect();
            }
            if ((readyOps & SelectionKey.OP_WRITE) != 0) {
                ch.unsafe().forceFlush(); // 处理写事件
            }
            if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
                unsafe.read(); // 处理读事件，客户端发送数据或连接请求时会出发该方法
            }
        } catch (CancelledKeyException ignored) {
            unsafe.close(unsafe.voidPromise());
        }
    }
}
private final class NioMessageUnsafe extends AbstractNioUnsafe {
    private final List<Object> readBuf = new ArrayList<Object>();
    @Override
    public void read() {
        assert eventLoop().inEventLoop();
        final ChannelConfig config = config();
        final ChannelPipeline pipeline = pipeline();
        final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
        allocHandle.reset(config);
        boolean closed = false;
        Throwable exception = null;
        try {
            try {
                do {
                    int localRead = doReadMessages(readBuf);
                    if (localRead == 0) {
                        break;
                    }
                    if (localRead < 0) {
                        closed = true;
                        break;
                    }
                    allocHandle.incMessagesRead(localRead);
                } while (allocHandle.continueReading());
            } catch (Throwable t) {
                exception = t;
            }
            int size = readBuf.size(); // readBuf中存储的是OP_ACCEPT事件连接过来的所有NioSocketChannel
            for (int i = 0; i < size; i ++) {
                readPending = false;
                pipeline.fireChannelRead(readBuf.get(i)); // 调用ServerBootstrapAcceptor的channelRead方法
            }
            readBuf.clear();
            allocHandle.readComplete();
            pipeline.fireChannelReadComplete();
            if (exception != null) {
                closed = closeOnReadError(exception);
                pipeline.fireExceptionCaught(exception);
            }
            if (closed) {
                inputShutdown = true;
                if (isOpen()) {
                    close(voidPromise());
                }
            }
        } finally {
            if (!readPending && !config.isAutoRead()) {
                removeReadOp();
            }
        }
    }
}
public class NioServerSocketChannel extends AbstractNioMessageChannel implements io.netty.channel.socket.ServerSocketChannel {
    protected int doReadMessages(List<Object> buf) throws Exception {
        SocketChannel ch = SocketUtils.accept(javaChannel()); // 获取SocketChannel
        try {
            if (ch != null) { // 将SocketChannel包装为NioSocketChannel，且初始化ChannelPipeline将阻塞模式设置为非阻塞
                buf.add(new NioSocketChannel(this, ch));
                return 1;
            }
        } catch (Throwable t) {
            try {
                ch.close();
            } catch (Throwable t2) {
            }
        }
        return 0;
    }
}
public class NioSocketChannel extends AbstractNioByteChannel implements io.netty.channel.socket.SocketChannel {
    public NioSocketChannel(Channel parent, SocketChannel socket) {
        super(parent, socket);
        config = new NioSocketChannelConfig(this, socket.socket());
    }
}
public abstract class AbstractNioByteChannel extends AbstractNioChannel {
    protected AbstractNioByteChannel(Channel parent, SelectableChannel ch) {
        super(parent, ch, SelectionKey.OP_READ); // 将channel感兴趣的事件设置为OP_READ
    }
}
public abstract class AbstractNioChannel extends AbstractChannel {
    protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
        super(parent); // 调用超类AbstractChannel的构造方法初始化ChannelPipeline
        this.ch = ch;
        this.readInterestOp = readInterestOp;
        try {
            ch.configureBlocking(false); // 设置为非阻塞模式
        } catch (IOException e) {
            try {
                ch.close();
            } catch (IOException e2) {
            }
            throw new ChannelException("Failed to enter non-blocking mode.", e);
        }
    }
}

若是客户端的发送数据事件最终调用AbstractNioByteChannel的NioByteUnsafe的read方法，将接收到的数据调用SocketChannel中pipeline中所有hander的channelRead方法处理。

protected class NioByteUnsafe extends AbstractNioUnsafe {
    public final void read() {
        final ChannelConfig config = config();
        if (shouldBreakReadReady(config)) {
            clearReadPending();
            return;
        }
        final ChannelPipeline pipeline = pipeline();
        final ByteBufAllocator allocator = config.getAllocator();
        final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();
        allocHandle.reset(config);
        ByteBuf byteBuf = null;
        boolean close = false;
        try {
            do {
                byteBuf = allocHandle.allocate(allocator);
                allocHandle.lastBytesRead(doReadBytes(byteBuf));
                if (allocHandle.lastBytesRead() <= 0) {
                    byteBuf.release();
                    byteBuf = null;
                    close = allocHandle.lastBytesRead() < 0;
                    if (close) {
                        readPending = false;
                    }
                    break;
                }
                allocHandle.incMessagesRead(1);
                readPending = false;
                pipeline.fireChannelRead(byteBuf); // 调用SocketChannel中pipeline中所有hander的channelRead方法
                byteBuf = null;
            } while (allocHandle.continueReading());
            allocHandle.readComplete();
            pipeline.fireChannelReadComplete(); // 调用SocketChannel中pipeline中所有hander的channelReadComplete方法
            if (close) {
                closeOnRead(pipeline);
            }
        } catch (Throwable t) {
            handleReadException(pipeline, byteBuf, t, close, allocHandle);
        } finally {
            if (!readPending && !config.isAutoRead()) {
                removeReadOp();
            }
        }
    }
}