@Sharable
public class EchoServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ctx.write(msg);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // Close the connection when an exception is raised.
        cause.printStackTrace();
        ctx.close();
    }
}


/**
 * Echoes back any received data from a client.
 */
public final class EchoServer {

    static final boolean SSL = System.getProperty("ssl") != null;
    static final int PORT = Integer.parseInt(System.getProperty("port", "8007"));

    public static void main(String[] args) throws Exception {
        // Configure SSL.
        final SslContext sslCtx;
        if (SSL) {
            SelfSignedCertificate ssc = new SelfSignedCertificate();
            sslCtx = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey()).build();
        } else {
            sslCtx = null;
        }

        // Configure the server.
        // 在官方的example中默认是Reactor主从多线程模式
        // 1.配置线程池
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        final EchoServerHandler serverHandler = new EchoServerHandler();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class) // 2.初始化channel类型
             .option(ChannelOption.SO_BACKLOG, 100) // 2.1.设置channel参数
             .handler(new LoggingHandler(LogLevel.INFO))
              // 注册channelhandler，在netty中通过ChannelPipeline 去注册多个 ChannelHandler
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();
                     if (sslCtx != null) {
                         p.addLast(sslCtx.newHandler(ch.alloc()));
                     }
                     //p.addLast(new LoggingHandler(LogLevel.INFO));
                     p.addLast(serverHandler);
                 }
             });

            // Start the server. 
            // 3.启动 通过bind() 方法会真正触发启动，sync() 方法则会阻塞
            ChannelFuture f = b.bind(PORT).sync();
            // Wait until the server socket is closed.
            f.channel().closeFuture().sync();
        } finally {
            // Shut down all event loops to terminate all threads.
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

private ChannelFuture doBind(final SocketAddress localAddress) {
    // 1.调用 initAndRegister()方法初始化Channel,接受返回到ChannelFuture。
    final ChannelFuture regFuture = initAndRegister();
  // 2.调用 ChannelFuture.channel方法，返回Channel。
    final Channel channel = regFuture.channel();
  // 3.调用 ChannelFuture.cause方法判断 initAndRegister()是否存在异常。存在异常则直接返回
    if (regFuture.cause() != null) {
        return regFuture;
    }
    // 4.调用 ChannelFuture.isDone方法，判断initAndRegister()方法是否执行完成。
    // 	 如果执行完成则调用bind0方法，如果没有执行完成，ChannelFuture 添加一个ChannelFutureListener回调监听，当initAndRegister()方法执行完成后回调operationComplete方法，调用调用 ChannelFuture.cause()方法判断initAndRegister()是否存在异常，无异常在执行bind0方法。
    if (regFuture.isDone()) {
        // At this point we know that the registration was complete and successful.
        ChannelPromise promise = channel.newPromise();
        doBind0(regFuture, channel, localAddress, promise);
        return promise;
    } else {
        // Registration future is almost always fulfilled already, but just in case it's not.
        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) {
                    // Registration on the EventLoop failed so fail the ChannelPromise directly to not cause an
                    // IllegalStateException once we try to access the EventLoop of the Channel.
                    promise.setFailure(cause);
                } else {
                    // Registration was successful, so set the correct executor to use.
                    // See https://github.com/netty/netty/issues/2586
                    promise.registered();

                    doBind0(regFuture, channel, localAddress, promise);
                }
            }
        });
        return promise;
    }
}

@Override
    protected void run() {
        for (;;) {
            try {
                try {
                    switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
                    case SelectStrategy.CONTINUE:
                        continue;

                    case SelectStrategy.BUSY_WAIT:
                        // fall-through to SELECT since the busy-wait is not supported with NIO

                    case SelectStrategy.SELECT: // 处理select事件
                        // 轮训io事件
                        select(wakenUp.getAndSet(false));

                        // 'wakenUp.compareAndSet(false, true)' is always evaluated
                        // before calling 'selector.wakeup()' to reduce the wake-up
                        // overhead. (Selector.wakeup() is an expensive operation.)
                        //
                        // However, there is a race condition in this approach.
                        // The race condition is triggered when 'wakenUp' is set to
                        // true too early.
                        //
                        // 'wakenUp' is set to true too early if:
                        // 1) Selector is waken up between 'wakenUp.set(false)' and
                        //    'selector.select(...)'. (BAD)
                        // 2) Selector is waken up between 'selector.select(...)' and
                        //    'if (wakenUp.get()) { ... }'. (OK)
                        //
                        // In the first case, 'wakenUp' is set to true and the
                        // following 'selector.select(...)' will wake up immediately.
                        // Until 'wakenUp' is set to false again in the next round,
                        // 'wakenUp.compareAndSet(false, true)' will fail, and therefore
                        // any attempt to wake up the Selector will fail, too, causing
                        // the following 'selector.select(...)' call to block
                        // unnecessarily.
                        //
                        // To fix this problem, we wake up the selector again if wakenUp
                        // is true immediately after selector.select(...).
                        // It is inefficient in that it wakes up the selector for both
                        // the first case (BAD - wake-up required) and the second case
                        // (OK - no wake-up required).

                        if (wakenUp.get()) {
                            selector.wakeup();
                        }
                        // fall through
                    default:
                    }
                } catch (IOException e) {
                    // If we receive an IOException here its because the Selector is messed up. Let's rebuild
                    // the selector and retry. https://github.com/netty/netty/issues/8566
                    rebuildSelector0();
                    handleLoopException(e);
                    continue;
                }

                cancelledKeys = 0;
                needsToSelectAgain = false;
                final int ioRatio = this.ioRatio;
                if (ioRatio == 100) {
                    try {
                        processSelectedKeys(); // 处理io事件
                    } finally {
                        // Ensure we always run tasks.
                        runAllTasks(); // 处理所有任务
                    }
                } else {
                    final long ioStartTime = System.nanoTime();
                    try {
                        processSelectedKeys(); // 处理io事件
                    } finally {
                        // Ensure we always run tasks.
                        final long ioTime = System.nanoTime() - ioStartTime;
                        runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
                    }
                }
            } catch (Throwable t) {
                handleLoopException(t);
            }
            // Always handle shutdown even if the loop processing threw an exception.
            try {
                if (isShuttingDown()) {
                    closeAll();
                    if (confirmShutdown()) {
                        return;
                    }
                }
            } catch (Throwable t) {
                handleLoopException(t);
            }
        }
    }

   serverBootstrap.childHandler(new ChannelInitializer<NioSocketChannel>() {
            @Override
            protected void initChannel(NioSocketChannel ch){
                ChannelPipeline channelPipeline = ch.pipeline();
                channelPipeline.addLast("idleHandler", new ServerIdleCheckHandler());
                channelPipeline.addLast("frameDecoder",new FrameDecoder());
                channelPipeline.addLast("frameEncoder",new FrameEncoder());
                channelPipeline.addLast("protocolDecoder",new ProtocolDecoder());
                channelPipeline.addLast("protocolEncoder",new ProtocolEncoder());
                channelPipeline.addLast("loggingHandler",new LoggingHandler(LogLevel.INFO));
                channelPipeline.addLast("protocolProcessHandler",new ServerProtocolProcessHandler());
            }
        });

EventLoopGroup bossGroup = new NioEventLoopGroup(0,new DefaultThreadFactory("boss"));
EventLoopGroup workerGroup = new NioEventLoopGroup(,new DefaultThreadFactory("work"));

 channelPipeline.addLast("idleHandler", new ServerIdleCheckHandler());
channelPipeline.addLast("frameDecoder",new FrameDecoder());
 channelPipeline.addLast("frameEncoder",new FrameEncoder());

private static InternalLoggerFactory newDefaultFactory(String name) {
        InternalLoggerFactory f;
        try {
            f = new Slf4JLoggerFactory(true);
            f.newInstance(name).debug("Using SLF4J as the default logging framework");
        } catch (Throwable ignore1) {
            try {
                f = Log4JLoggerFactory.INSTANCE;
                f.newInstance(name).debug("Using Log4J as the default logging framework");
            } catch (Throwable ignore2) {
                try {
                    f = Log4J2LoggerFactory.INSTANCE;
                    f.newInstance(name).debug("Using Log4J2 as the default logging framework");
                } catch (Throwable ignore3) {
                    f = JdkLoggerFactory.INSTANCE;
                    f.newInstance(name).debug("Using java.util.logging as the default logging framework");
                }
            }
        }
        return f;
    }

// mac 
EventLoopGroup bossGroup = new KQueueEventLoopGroup(0,new DefaultThreadFactory("boss"));
EventLoopGroup workerGroup = new KQueueEventLoopGroup(0,new DefaultThreadFactory("work"));
// linxu 
EventLoopGroup bossGroup = new EpollEventLoopGroup(0,new DefaultThreadFactory("boss"));
EventLoopGroup workerGroup = new EpollEventLoopGroup(0,new DefaultThreadFactory("work"));
// 注意开发环境与生产环境,

EventLoopGroup group = new NioEventLoopGroup(1);
ServerBootstrap b = new ServerBootstrap();
b.group(group)

EventLoopGroup group = new NioEventLoopGroup();
ServerBootstrap b = new ServerBootstrap();
b.group(group)

// 该方法在不同的平台上返回不同的类型。
sun.nio.ch.DefaultSelectorProvider.create()
  
  
package sun.nio.ch;
import java.nio.channels.spi.SelectorProvider;

public class DefaultSelectorProvider {
    private DefaultSelectorProvider() {
    }

    public static SelectorProvider create() {
        // 在mac/bsd的环境下返回KQueue
        return new KQueueSelectorProvider();
    }
}

     // 每次执行 Select 操作之前记录当前时间 currentTimeNanos。
                long time = System.nanoTime();
                // time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos，
                // 如果事件轮询的持续时间大于等于 timeoutMillis，那么说明是正常的，否则表明阻塞时间并未达到预期，可能触发了空轮询的 Bug。
                if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {
                    // timeoutMillis elapsed without anything selected.
                    selectCnt = 1;
                    // 计数变量 selectCnt。在正常情况下，selectCnt 会重置，
                    // 否则会对 selectCnt 自增计数。当 selectCnt 达到 SELECTOR_AUTO_REBUILD_THRESHOLD（默认512） 阈值时，
                    // 会触发重建 Selector 对象。
                } else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 &&
                        selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {
                    // The code exists in an extra method to ensure the method is not too big to inline as this
                    // branch is not very likely to get hit very frequently.
                    selector = selectRebuildSelector(selectCnt);
                    selectCnt = 1;
                    break;
                }
                currentTimeNanos = time;

ch.pipeline().addLast(new ProtobufVarint32FrameDecoder());
ch.pipeline().addLast(new ProtobufDecoder(PersonOuterClass.Person.getDefaultInstance()));
ch.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());
ch.pipeline().addLast(new ProtobufEncoder());