Netty服务端启动流程

　　首先附上一个简单的服务端启动代码

 1 public void bind(int port) throws Exception {
 2 //        线程组 一个用于接受客户端连接 一个用于IO操作
 3 //        parentGroup用于serverBootstrap的父类AbstractBootstrap使用的线程池
 4 //        AbstractBootstrap是个工厂类，用于接受客户端连接
 5 //        如果只监听一个端口则只需要一个线程即可，因为一个NioServerSocketChannel只能够与一个NioEventLoop绑定，该channel的所有操作均由绑定的NioEventLoop进行
 6         EventLoopGroup parentGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("parent"));
 7 //        childGroup是ServerBootstrap使用的线程池
 8         EventLoopGroup childGroup = new NioEventLoopGroup(0, new DefaultThreadFactory("child"));
 9 //        netty启动辅助类
10         ServerBootstrap bootstrap = new ServerBootstrap();
11         try {
12 //            绑定线程组
13             bootstrap.group(parentGroup, childGroup)
14 //                设置使用的channel
15                 .channel(NioServerSocketChannel.class)
16 //                设置参数
17                 .option(ChannelOption.SO_BACKLOG, 1024)
18 //                绑定事件处理类
19                 .childHandler(new ChildChannelHandler());
20 //            绑定端口并且同步等待操作完成
21             ChannelFuture channelFuture = bootstrap.bind(port);
22 //            等待服务端链路关闭之后main函数才退出
23             channelFuture.channel().closeFuture().sync();
24         } finally {
25 //            优雅关闭线程组,释放资源
26             parentGroup.shutdownGracefully();
27             childGroup.shutdownGracefully();
28         }
29     }

　　其中第21行的bind方法便是启动入口方法。该方法最终会调用AbstrractBootstrap#doBind方法。

 1 private ChannelFuture doBind(final SocketAddress localAddress) {
 2     // 创建、初始化、注册channel
 3     final ChannelFuture regFuture = initAndRegister();
 4     final Channel channel = regFuture.channel();
 5     if (regFuture.cause() != null) {
 6         return regFuture;
 7     }
 8 
 9     final ChannelPromise promise;
10     // 将channel绑定到对应地址上
11     if (regFuture.isDone()) {
12         promise = channel.newPromise();
13         doBind0(regFuture, channel, localAddress, promise);
14     } else {
15         // Registration future is almost always fulfilled already, but just in case it's not.
16         promise = new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE);
17         regFuture.addListener(new ChannelFutureListener() {
18             @Override
19             public void operationComplete(ChannelFuture future) throws Exception {
20                 doBind0(regFuture, channel, localAddress, promise);
21             }
22         });
23     }
24     return promise;
25 }

　　首先创建、初始化、注册netty的channel，由于注册是一个异步的过程，所以initAndRegister()方法返回的是一个ChannelFuture，该类是netty对java的Future的拓展类。

　　在执行绑定操作之前需要先判断是否注册成功，如果成功则执行绑定方法，否则添加一个listener到Future中，listener会在future成功之后会被调用。

　　接下来看下initAndRegister()方法

 1 final ChannelFuture initAndRegister() {
 2     Channel channel;
 3     try {
 4         // 创建channel
 5         channel = createChannel();
 6     } catch (Throwable t) {
 7         return VoidChannel.INSTANCE.newFailedFuture(t);
 8     }
 9 
10     try {
11         // 初始化channel
12         init(channel);
13     } catch (Throwable t) {
14         channel.unsafe().closeForcibly();
15         return channel.newFailedFuture(t);
16     }
17 
18     ChannelPromise regFuture = channel.newPromise();
19     // 注册channel
20     channel.unsafe().register(regFuture);
21     if (regFuture.cause() != null) {
22         if (channel.isRegistered()) {
23             channel.close();
24         } else {
25             channel.unsafe().closeForcibly();
26         }
27     }
28     return regFuture;
29 }

　　首先是创建channel实例，这里的channel是我们启动时设置的，该方法是通过反射的方式创建channel，源码如下

 1 @Override
 2 Channel createChannel() {
 3     EventLoop eventLoop = group().next();
 4     return channelFactory().newChannel(eventLoop, childGroup);
 5 }
 6 
 7 private static final class ServerBootstrapChannelFactory<T extends ServerChannel>
 8         implements ServerChannelFactory<T> {
 9 
10     private final Class<? extends T> clazz;
11     ServerBootstrapChannelFactory(Class<? extends T> clazz) {
12         this.clazz = clazz;
13     }
14 
15     @Override
16     public T newChannel(EventLoop eventLoop, EventLoopGroup childGroup) {
17         try {
18             Constructor<? extends T> constructor = clazz.getConstructor(EventLoop.class, EventLoopGroup.class);
19             return constructor.newInstance(eventLoop, childGroup);
20         } catch (Throwable t) {
21             throw new ChannelException("Unable to create Channel from class " + clazz, t);
22         }
23     }
24 
25     @Override
26     public String toString() {
27         return StringUtil.simpleClassName(clazz) + ".class";
28     }
29 }

　　首先是从parentGroup中获取一个eventLoop，然后将childGroup作为参数传递。

　　通过反射调用channel的构造方法创建实例，在创建channel的过程中做了三件重要的事情

　　1、与eventLoop绑定以及将childGroup作为IO线程池处理IO操作

　　2、创建对应的java socketChannel

　　3、初始化readInterestOp，这个参数后面会提到

　　具体源码读者自行查看NioServerSockerChannel的构造方法，其实现简单，读者可以自行跟踪。

　　然后是初始化channel，将channel、childHandler的参数设置到对应对象上，然后将childHandler添加到channel的pipleline中，源码如下

 1 @Override
 2 void init(Channel channel) throws Exception {
 3     final Map<ChannelOption<?>, Object> options = options();
 4     synchronized (options) {
 5         channel.config().setOptions(options);
 6     }
 7 
 8     final Map<AttributeKey<?>, Object> attrs = attrs();
 9     synchronized (attrs) {
10         for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) {
11             @SuppressWarnings("unchecked")
12             AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();
13             channel.attr(key).set(e.getValue());
14         }
15     }
16 
17     ChannelPipeline p = channel.pipeline();
18     if (handler() != null) {
19         p.addLast(handler());
20     }
21 
22     final ChannelHandler currentChildHandler = childHandler;
23     final Entry<ChannelOption<?>, Object>[] currentChildOptions;
24     final Entry<AttributeKey<?>, Object>[] currentChildAttrs;
25     synchronized (childOptions) {
26         currentChildOptions = childOptions.entrySet().toArray(newOptionArray(childOptions.size()));
27     }
28     synchronized (childAttrs) {
29         currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(childAttrs.size()));
30     }
31 
32     p.addLast(new ChannelInitializer<Channel>() {
33         @Override
34         public void initChannel(Channel ch) throws Exception {
35             ch.pipeline().addLast(new ServerBootstrapAcceptor(currentChildHandler, currentChildOptions,
36                     currentChildAttrs));
37         }
38     });
39 }

　　前面都是在初始化属性，重要的最后一段代码，它注册了一个ChannelInitializer到channelpipline上，该ChannelInitializer的initChannel方法创建了一个ServerBootstrapAcceptor注册到channel上，这个ServerBootstrapAcceptor会在服务器处理连接请求时使用，这个在我的另一篇服务端如何接受并分发请求文章中提到。

　　最后是注册方法，代码如下

 1 public final void register(final ChannelPromise promise) {
 2     if (eventLoop.inEventLoop()) {
 3         register0(promise);
 4     } else {
 5         try {
 6             eventLoop.execute(new Runnable() {
 7                 @Override
 8                 public void run() {
 9                     register0(promise);
10                 }
11             });
12         } catch (Throwable t) {
13             logger.warn(
14                     "Force-closing a channel whose registration task was not accepted by an event loop: {}",
15                     AbstractChannel.this, t);
16             closeForcibly();
17             closeFuture.setClosed();
18             promise.setFailure(t);
19         }
20     }
21 }

　　先判断当前线程是否是线程池的线程，如果是则直接执行注册方法，否则提交任务到线程池。为什么要这样做呢？

　　《Netty权威指南 第二版》中是这样说到——首先判断是否是NioEventLoop自身发起的操作。如果是，则不存在并发操作，直接执行Channel注册；如果由其他线程发起，则封装成一个Task放入消息队列中异步执行。此处，由于是由ServerBootstrap所在线程执行的注册操作，所以会将其封装成Task投递到NioEventLoop中执行。

　　个人觉得并非如此，注册方法最终会调用到AbstractSelectableChannel#register方法（下面会讲到），该方法使用synchronized关键字同步进行注册，是线程安全的。所以这里并不是用来防止并发操作，仅仅是由于注册操作结果用户并不关心且注册是耗时操作，所以这里为了提升性能做成了异步；同时还有一个作用是如果线程池还有未启动的线程，提交任务能够启动一个线程。如有错误，希望指正。

　　我们继续看register0方法

private void register0(ChannelPromise promise) {
    try {
        // check if the channel is still open as it could be closed in the mean time when the register
        // call was outside of the eventLoop
        if (!ensureOpen(promise)) {
            return;
        }
        doRegister();
        registered = true;
        // 将线程执行结果设置为成功
        promise.setSuccess();
        // 通知ChannelRegistered事件
        pipeline.fireChannelRegistered();
        if (isActive()) {
            pipeline.fireChannelActive();
        }
    } catch (Throwable t) {
        // Close the channel directly to avoid FD leak.
        closeForcibly();
        closeFuture.setClosed();
        if (!promise.tryFailure(t)) {
            logger.warn(
                    "Tried to fail the registration promise, but it is complete already. " +
                            "Swallowing the cause of the registration failure:", t);
        }
    }
}

　　doRegister方法是真正执行注册操作的方法，注册操作成功后会将promise的执行结果设置为成功，该方法同时会调用注册的listener。

　　接下来看看AbstractNioChannel#doRegister方法

 1 protected void doRegister() throws Exception {
 2     boolean selected = false;
 3     for (;;) {
 4         try {
 5             // 将channel注册到多路复用器上
 6             selectionKey = javaChannel().register(eventLoop().selector, 0, this);
 7             return;
 8         } catch (CancelledKeyException e) {
 9             if (!selected) {
10                 // Force the Selector to select now as the "canceled" SelectionKey may still be
11                 // cached and not removed because no Select.select(..) operation was called yet.
12                 eventLoop().selectNow();
13                 selected = true;
14             } else {
15                 // We forced a select operation on the selector before but the SelectionKey is still cached
16                 // for whatever reason. JDK bug ?
17                 throw e;
18             }
19         }
20     }
21 }

　　这里将channel注册到执行当前操作的eventLoop的多路复用器上，并且将附件设置channel作为附件，注册的操作为0。

　　在SelectionKey类中定义了4中多路复用器的操作值，如下

public static final int OP_READ = 1 << 0;
public static final int OP_WRITE = 1 << 2;
public static final int OP_CONNECT = 1 << 3;
public static final int OP_ACCEPT = 1 << 4;

　　这里注册的操作是0，也就是不关心任何操作，这是为什么呢？

　　《Netty权威指南 第二版》中是这样说到——注册方法是多台的，它既可以被NioServerSocketChannel用来监听客户端的连接接入，也可以注册socketChannel用来监听网络读或者写操作。

　　那么什么时候会将操作设置为正确的值呢？请往后看。

　　经过以上流程遍完成了channel的创建、初始化、注册，这里只讲了大致的流程，其中比较细节的，例如流程中用到的参数是何时初始化的等问题读者可以自行阅读源码。

　　接下来回头看doBind方法，执行完了initAndRegister方法后，接下来会执行绑定操作，让我们看看doBind0方法源码

 1 private static void doBind0(
 2         final ChannelFuture regFuture, final Channel channel,
 3         final SocketAddress localAddress, final ChannelPromise promise) {
 4 
 5     // This method is invoked before channelRegistered() is triggered.  Give user handlers a chance to set up
 6     // the pipeline in its channelRegistered() implementation.
 7     channel.eventLoop().execute(new Runnable() {
 8         @Override
 9         public void run() {
10             if (regFuture.isSuccess()) {
11                 channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
12             } else {
13                 promise.setFailure(regFuture.cause());
14             }
15         }
16     });
17 }

　　doBind0方法很简单，将绑定操作提交到线程池中，这样做的原因与注册操作是一样的。

　　channel的bind方法最终会执行到AbstractUnsafe#bind方法

@Override
public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
    if (!ensureOpen(promise)) {
        return;
    }
    // 省略部分代码
    // 判断channel是否已经激活
    boolean wasActive = isActive();
    try {
        doBind(localAddress);
    } catch (Throwable t) {
        promise.setFailure(t);
        closeIfClosed();
        return;
    }
    if (!wasActive && isActive()) {
        invokeLater(new Runnable() {
            @Override
            public void run() {
                pipeline.fireChannelActive();
            }
        });
    }
    promise.setSuccess();
}

　　doBind方法是真正执行绑定操作的方法，是调用java的的ServerSocket#bind方法。

　　绑定成功之后判断是否是第一次注册，如果是则通知channelActive事件，代码如下

1 @Override
2 public ChannelPipeline fireChannelActive() {
3     head.fireChannelActive();
4     if (channel.config().isAutoRead()) {
5         channel.read();
6     }
7     return this;
8 }

　　通知完channelActive事件后会进行判断，channel是否是自动读，该值默认为true，所以会默认调用channel.read方法，该方法最终会调用AbstractNioUnsafe#doBeginRead方法，该方法代码如下

 1 protected void doBeginRead() throws Exception {
 2     if (inputShutdown) {
 3         return;
 4     }
 5     final SelectionKey selectionKey = this.selectionKey;
 6     if (!selectionKey.isValid()) {
 7         return;
 8     }
 9     final int interestOps = selectionKey.interestOps();
10     if ((interestOps & readInterestOp) == 0) {
11         selectionKey.interestOps(interestOps | readInterestOp);
12     }
13 }

　　注意第11行代码，将selectionKey的注册操作改为了readInterestOp，该值是一个NioServerSocketChannel的父类AbstractNioChannel的属性，让我们看下NioServerSocketChannel的构造函数

1 public NioServerSocketChannel(EventLoop eventLoop, EventLoopGroup childGroup) {
2     super(null, eventLoop, childGroup, newSocket(), SelectionKey.OP_ACCEPT);
3     config = new DefaultServerSocketChannelConfig(this, javaChannel().socket());
4 }

　　可以看到，NioServerSocketChannel将readInterestOp设置成了OP_ACCEPT。所以当服务端channel完成绑定操作之后会将注册到多路复用器上的操作变为OP_ACCEPT。

　　以上便是服务端的启动流程，客户端的启动流程实际上与服务端类似，读者可以自行阅读源码。

　　在文章开头的启动代码中，还有一个地方没有讲到，那就是最后的优雅停机，这个实现较为简单但是涉及到许多代码这里不赘述，这里仅大概说一下他的原理。优雅停机与java的线程池关闭类似，都是通过一个state变量来表示线程池的状态，当线程池中的线程判断到state变为关闭等状态时，便会执行退出操作，当所有线程都退出之后便完成了优雅停机。

　　大家可以看到在netty服务端启动过程中大量的涉及到了线程池操作，线程池是netty的核心之一，所有的操作都是在NioEventLoop中进行，感兴趣的可以自行了解一下NioEventLoop的run方法，关于netty的线程模型可以看下这篇文章https://www.infoq.cn/article/netty-threading-model。

【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/ouhaitao/p/12876027.html