Netty源码解析--Echo示例

• 前言
• 源码解析
• • 事件循环组（EventLoopGroup）的初始化
  • 引导类配置
  • 启动服务

想学习netty的话，需要两个知识做支撑，一个是io多路复用，一个是reactor反应堆；
下面两篇文章是我认为非常细致清楚的对上面两个知识点的讲解；
io多路复用
reactor反应堆模型

netty使用的是主从reactor模型，主reactor负责处理连接事件，创建channel，并监听io事件，如果有io事件产生，会将channel注册到从reactor上，从reactor会挑选工作线程来进行处理；

整个服务的启动过程表面看起来还是比较简单的；
首先是初始化boss工作组和worker工作组，其次通过引导类来根据你的服务的需要进行配置，最后绑定端口服务启动；
netty的源码比较晦涩，内部的继承关系十分复杂，希望读着可以打开源码通过debug调试的方式跟着我的脚步一起走；
debug启动的时候可能会报错，需要先用maven compile 编译下common包，里面有几个template模板，编译成功后在启动main方法；

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 {
        
        // 这里都是些ssl的加密设置，可以忽略掉
        final SslContext sslCtx;
        if (SSL) {
            SelfSignedCertificate ssc = new SelfSignedCertificate();
            sslCtx = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey()).build();
        } else {
            sslCtx = null;
        }

		// 这里是第一步 初始化事件循环组，一个是boss 一个是worker，boss就是主从reactor里面的mainReactor，
		// worker的其中一个线程会被当做	subReactor，其他线程充当工作线程，也就是实际执行具体处理的线程，这个我们在后面的解析中会看到；
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        final EchoServerHandler serverHandler = new EchoServerHandler();
        try {
        
        // 这里是第二步，serverBootStrap是个引导类，说白了就是让你通过这个类来进行一些配置的；
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .option(ChannelOption.SO_BACKLOG, 100)
             .handler(new LoggingHandler(LogLevel.INFO))
             .childHandler(new ChannelInitializer() {
                 @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);
                 }
             });

			// 这里是第三步，去绑定端口，但是实际上这里面的逻辑是最重要的，上面设置的配置都会在这里一一初始化并且启动；
            ChannelFuture f = b.bind(PORT).sync();
			
			// 等待链接完成并关闭，这里不是重点；
            f.channel().closeFuture().sync();
        } finally {
        	// 优雅关闭，也不是重点
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

上面的代码看完，相信大家已经有一定理解，我们先不必纠结内部的实现过程，这里我们只需要了解整个启动过程即可，其内部的实现，我会带大家逐步拆解；
首先拆解第一个问题，也就是工作组的初始化，上面代码中boss和worker的初始化过程；
可以注意掉，echoServer中用的是NioEventLoopGroup，这里大家不必纠结，不同服务类型和需求你要使用的工作组可能不同，但是他们的原理都大同小异，我们还是以NioEventLoop为例进行解析；

public NioEventLoopGroup(int nThreads) {
		// 这里 这个executor是null我们需要注意，后面会对这个null做处理
        this(nThreads, (Executor) null);
    }

我们进入到NioEventLoopGroup中，下面是上面第一步中调用的构造器；我只需要沿着这个构造器一直追踪下去即可；

public NioEventLoopGroup(int nThreads, Executor executor) {
        this(nThreads, executor, SelectorProvider.provider());
    }

可以看到，相比上一步多了个selectorProvider，我们可以进去看一下

 public static SelectorProvider provider() {
        synchronized (lock) {
            if (provider != null)
                return provider;
            return AccessController.doPrivileged(
                new PrivilegedAction() {
                    public SelectorProvider run() {
                            if (loadProviderFromProperty())
                                return provider;
                            if (loadProviderAsService())
                                return provider;
                            provider = sun.nio.ch.DefaultSelectorProvider.create();
                            return provider;
                        }
                    });
        }
    }

到了这里，loadProviderFromProperty和loadProviderAsService应该是让我们自己来提供provider，如果都没有的话会给我们一个nio的默认provider，即DefaultSelectorProvider.create()；

public class DefaultSelectorProvider {
  private DefaultSelectorProvider() {
  }

  public static SelectorProvider create() {
    return new KQueueSelectorProvider();
  }
}

可以看到这里的provider是一个KQueueSelectorProvider，我们继续进入；

public class KQueueSelectorProvider extends SelectorProviderImpl {
  public KQueueSelectorProvider() {
  }

  public AbstractSelector openSelector() throws IOException {
    return new KQueueSelectorImpl(this);
  }
}

注意，这里的openSelector我们后面会用到，其含义就是通过provider为们提供一个selector实例，来轮训io事件；
接下来我们回到刚才的内容，这里我在贴一下这部分代码；

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);
    }

可以看到又多了一个DefaultSelectStrategyFactory，大家可以进去看一下，代码不复杂，里面是一个单例模式的一个轮训策略工厂；
继续深入；

 public NioEventLoopGroup(int nThreads, Executor executor, final SelectorProvider selectorProvider,
                             final SelectStrategyFactory selectStrategyFactory) {
        super(nThreads, executor, selectorProvider, selectStrategyFactory, RejectedExecutionHandlers.reject());
    }

线程池的拒绝策略，没什么好说的，我们继续；

  
    protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {
        super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
    }

接下来我们来到了NioEventLoopGroup的父类MultithreadEventExecutorGroup，这对线程数做了一下处理，继续；

protected MultithreadEventExecutorGroup(int nThreads, Executor executor, Object... args) {
        this(nThreads, executor, DefaultEventExecutorChooserFactory.INSTANCE, args);
    }

DefaultEventExecutorChooserFactory是给工作组用来挑选处理线程的，先略过，后面会说；
到这里，我们迎来了初始化NioEventLoopGroup的关键地方；

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) {
                // TODO: Think about if this is a good exception type
                throw new IllegalStateException("failed to create a child event loop", e);
            } finally {
                if (!success) {
                    for (int j = 0; j < i; j ++) {
                        children[j].shutdownGracefully();
                    }

                    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;
                        }
                    }
                }
            }
        }

        chooser = chooserFactory.newChooser(children);
        
		// 后面不是
        final FutureListener