netty5笔记-总体流程分析1-ServerBootstrap启动

Love The Way You Lie 2022-05-27 12:39 206阅读 0赞

前面我们分析了netty内存池、线程模型，比较难的两个点已经被一一消化，接下来我们开始进入大家最关心的环节，总体流程分析。 这里我选了io.netty.example.http.snoop来作为分析的入口，分别从server端、client端的维度来看看netty是如果设计的。这里你将了解比较详细的netty处理流程，让你在今后的应用中不再感到疑惑 。 如果还有不清楚的地方，可以直接交流，通过交流发现问题，并不断完善这系列文章。 本文假设你对netty已有大致了解，需要更深入的了解它的运作流程。如果是初学者，可能会发现头晕、眼花、脑抽经等症状。

配置篇

首先看看snoop包的server端启动代码。 在netty中，不管是server还是client，都是由引导类进行启动。在启动之前需要先做好各种参数的配置。可以配置的参数如下：

<table style="width:780px;color:rgb(69,69,69);font-size:16px;background-color:rgb(255,255,255);"> 
 <tbody> 
  <tr> 
   <td>字段</td> 
   <td>类型</td> 
   <td>说明</td> 
   <td>server模式</td> 
   <td>client模式</td> 
  </tr> 
  <tr> 
   <td>options</td> 
   <td>Map</td> 
   <td>channel的配置项</td> 
   <td>作用于ServerChannel</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>childOptions</td> 
   <td>Map</td> 
   <td>channel的配置项</td> 
   <td>作用于Channel</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>attrs</td> 
   <td>Map</td> 
   <td>自定义的channel属性</td> 
   <td>作用于ServerChannel</td> 
   <td>作用于Channel</td> 
  </tr> 
  <tr> 
   <td>childAttrs</td> 
   <td>Map</td> 
   <td>自定义的channel属性</td> 
   <td>作用于Channel</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>handler</td> 
   <td>ChannelHandler</td> 
   <td>连接处理器</td> 
   <td>作用于ServerChannel</td> 
   <td>作用于Channel</td> 
  </tr> 
  <tr> 
   <td>childHandler</td> 
   <td>ChannelHandler</td> 
   <td>连接处理器</td> 
   <td>作用于Channel</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>group</td> 
   <td>EventLoopGroup</td> 
   <td>注册并处理连接</td> 
   <td>作用于ServerChannel</td> 
   <td>作用于Channel</td> 
  </tr> 
  <tr> 
   <td>childGroup</td> 
   <td>EventLoopGroup</td> 
   <td>注册并处理连接</td> 
   <td>作用于Channel</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>channelFactory</td> 
   <td>ChannelFactory</td> 
   <td>生成连接对象的工厂类</td> 
   <td>生成ServerChannel</td> 
   <td>生成Channel</td> 
  </tr> 
 </tbody> 
</table>

除了channelFactory所有的字段都分成了xxx和childxxx两个相对应的字段，名称上能很容易的分出来字段的作用范围。 如我们希望设置SO\_REUSEADDR参数，该参数作用于ServerSocket，则设置时调用option(ChannelOption.SO\_REUSEADDR, true)。对于Server端来说，比较常见的几个设置：SO\_KEEPALIVE、SO\_REUSEADDR、TCP\_NODELAY、SO\_BACKLOG。

我们知道netty采用了reactor的设计模式，其中mainReactor主要负责连接的建立，连接建立后交由subReactor处理，而subReactor则主要负责处理读写等具体的事件。这里mainReactor的实际执行者是bossGroup，而subReactor的实际执行者则是workerGroup。 下面是HttpSnoopServer类中main方法的主要代码（去掉了一部分）

**\[java\]**  [view plain][]  [copy][view plain]

1.  EventLoopGroup bossGroup = new NioEventLoopGroup(1);  
2.  EventLoopGroup workerGroup = new NioEventLoopGroup();  
3.  try \{  
4.      ServerBootstrap b = new ServerBootstrap();  
5.      b.group(bossGroup, workerGroup)  
6.       .channel(NioServerSocketChannel.class)  
7.       .handler(new LoggingHandler(LogLevel.INFO))  
8.       .childHandler(new HttpSnoopServerInitializer(sslCtx));  
9.    
10.     Channel ch = b.bind(PORT).sync().channel();  
11.     ch.closeFuture().sync();  
12. \} finally \{  
13.     bossGroup.shutdownGracefully();  
14.     workerGroup.shutdownGracefully();  
15. \}

这里bossGroup只启用了一个线程，因为一个端口只能创建一个ServerChannel，该ServerChannel的整个生命周期都在bossGroup中。如果你想用同一个ServerBootstrap启动多个端口，则bossGroup的大小需要根据启动的端口数调整。handler设置为LogginHandler，表示在ServerChannel的处理链中加入了日志记录（这个与客户端连接无关，即它只记录ServerChannel的注册、注销、关闭等，而不会记录客户端连接的相应事件。之前有同学加了LoggingHandler而没看到客户端的相应日志，就是这样了。需要的话要在childHandler的Initializer中加入LoggingHandler）。 childHandler设置为HttpSnoopServerInitializer，即用户连接使用HttpSnoopServerInitializer进行处理。

初始化完成开始调用bind(port)方法，bind首先会对各个参数进行验证，如channelFactory是否设置,group、childGroup是否设置，端口是否设置等，验证通过后，最终调用doBind方法（AbstractBootstrap中)。

**\[java\]**  [view plain][]  [copy][view plain]

1.  private ChannelFuture doBind(final SocketAddress localAddress) \{  
2.      // 初始化并注册Channel(此时是ServerChannel)  
3.      final ChannelFuture regFuture = initAndRegister();  
4.      final Channel channel = regFuture.channel();  
5.      // 如果注册出错则直接返回  
6.      if (regFuture.cause() != null) \{  
7.          return regFuture;  
8.      \}  
9.    
10.     // 注册完成调用doBind0，否则添加一个注册事件的监听器，该监听器在监听到注册完成后也会触发doBind0操作  
11.     if (regFuture.isDone()) \{  
12.         // At this point we know that the registration was complete and successful.  
13.         ChannelPromise promise = channel.newPromise();  
14.         doBind0(regFuture, channel, localAddress, promise);  
15.         return promise;  
16.     \} else \{  
17.         // 一般来说都会是进入isDone，这里是以防万一  
18.         final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);  
19.         regFuture.addListener(new ChannelFutureListener() \{  
20.             @Override  
21.             public void operationComplete(ChannelFuture future) throws Exception \{  
22.                 Throwable cause = future.cause();  
23.                 if (cause != null) \{  
24.                     promise.setFailure(cause);  
25.                 \} else \{  
26.                     promise.executor = channel.eventLoop();  
27.                 \}  
28.                 doBind0(regFuture, channel, localAddress, promise);  
29.             \}  
30.         \});  
31.         return promise;  
32.     \}  
33. \}

doBind首先会调用initAndRegister方法，来看看这个方法做了什么：

**\[java\]**  [view plain][]  [copy][view plain]

1.  final ChannelFuture initAndRegister() \{  
2.      final Channel channel = channelFactory().newChannel();  
3.      try \{  
4.          init(channel);  
5.      \} catch (Throwable t) \{  
6.          channel.unsafe().closeForcibly();  
7.          // 此时连接还未注册到EventLoopGroup，因此使用GlobalEventExecutor  
8.          return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t);  
9.      \}  
10.   
11.     // 将连接注册到group中  
12.     ChannelFuture regFuture = group().register(channel);  
13.     if (regFuture.cause() != null) \{  
14.         if (channel.isRegistered()) \{  
15.             channel.close();  
16.         \} else \{  
17.             channel.unsafe().closeForcibly();  
18.         \}  
19.     \}  
20.   
21.     return regFuture;  
22. \}

channelFactory().newChannel()方法创建了一个  NioServerSocketChannel实例，该实例初始化时由SelectorProvider.provider().openServerSocketChannel()来打开一个ServerSocketChannel，同时会调用configureBlocking(false)将其IO模式设置为非阻塞。

**\[java\]**  [view plain][]  [copy][view plain]

1.      private static final SelectorProvider DEFAULT\_SELECTOR\_PROVIDER = SelectorProvider.provider();  
2.      private static ServerSocketChannel newSocket(SelectorProvider provider) \{  
3.          try \{  
4.              // 打开一个ServerSocketChannel  
5.              return provider.openServerSocketChannel();  
6.          \} catch (IOException e) \{  
7.              throw new ChannelException(  
8.                      "Failed to open a server socket.", e);  
9.          \}  
10.     \}  
11.   
12.     public NioServerSocketChannel() \{  
13.         this(newSocket(DEFAULT\_SELECTOR\_PROVIDER));  
14.     \}  
15.   
16.     public NioServerSocketChannel(ServerSocketChannel channel) \{  
17.         // 只对OP\_ACCEPT事件感兴趣  
18.         super(null, channel, SelectionKey.OP\_ACCEPT);  
19.         // 初始化连接对应的配置  
20.         config = new NioServerSocketChannelConfig(this, javaChannel().socket());  
21.     \}  
22.      
23.     protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) \{  
24.         super(parent);  
25.         this.ch = ch;  
26.         this.readInterestOp = readInterestOp;  
27.         try \{  
28.             // 将ServerSocketChannel设置为非阻塞模式  
29.             ch.configureBlocking(false);  
30.         \} catch (IOException e) \{  
31.             try \{  
32.                 ch.close();  
33.             \} catch (IOException e2) \{  
34.                 if (logger.isWarnEnabled()) \{  
35.                     logger.warn(  
36.                             "Failed to close a partially initialized socket.", e2);  
37.                 \}  
38.             \}  
39.   
40.             throw new ChannelException("Failed to enter non-blocking mode.", e);  
41.         \}  
42.     \}  
43.   
44.     protected AbstractChannel(Channel parent) \{  
45.         this.parent = parent;  
46.         // 非配id,该id全局唯一  
47.         id = DefaultChannelId.newInstance();  
48.         // 初始化Unsafe, Server生成的Unsafe类为NioMessageUnsafe，Unsafe属于较底层的操作，不对应用开放  
49.         // 它处理的各种操作：register、bind、connect、disconnect、close、deregister，beginRead、write、flush  
50.         unsafe = newUnsafe();  
51.         // 创建pipeline  
52.         pipeline = new DefaultChannelPipeline(this);  
53.     \}

完成后调用init进行对该ServerSocketChannel进行其他部分的初始化，init方法主要是：1、设置option；2、设置attr；3、如果设置了handler，将handler加入到处理链中（本例中加入LoggingHandler）。最后会加入一个ChannelInitializer，该ChannelInitializer主要功能是获取客户端连接后对连接进行初始化（具体如何初始化稍后再讲）。从下面代码可以看到，所有option/childOption之类的字段最终都会生成一份copy的数据，也就是该引导类可以继续使用（但是不能多个线程同时调用），用于引导其他端口的启动。

**\[java\]**  [view plain][]  [copy][view plain]

1.  void init(Channel channel) throws Exception \{  
2.      final Map<ChannelOption<?>, Object> options = options();  
3.      synchronized (options) \{  
4.          channel.config().setOptions(options);  
5.      \}  
6.    
7.      final Map<AttributeKey<?>, Object> attrs = attrs();  
8.      synchronized (attrs) \{  
9.          for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) \{  
10.             @SuppressWarnings("unchecked")  
11.             AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();  
12.             channel.attr(key).set(e.getValue());  
13.         \}  
14.     \}  
15.   
16.     ChannelPipeline p = channel.pipeline();  
17.     if (handler() != null) \{  
18.         p.addLast(handler());  
19.     \}  
20.   
21.     final EventLoopGroup currentChildGroup = childGroup;  
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.             // 这里的ServerBootstrapAcceptor比较重要先记住  
36.             ch.pipeline().addLast(new ServerBootstrapAcceptor(  
37.                     currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));  
38.         \}  
39.     \});  
40. \}

初始化完成后立即将ServerChannel注册到bossGroup中，注册的时候会进行哪些操作呢？如果你还记得之前的EventLoop源码分析，就是这一句了：channel.unsafe().register(this, promise);  这行代码最终会调用AbstractChannel.AbstractUnsafe.register(EventLoop eventLoop, final ChannelPromise promise)  
方法：

**\[java\]**  [view plain][]  [copy][view plain]

1.  public final void register(EventLoop eventLoop, final ChannelPromise promise) \{  
2.      ...去掉非主要代码...  
3.    
4.      // channel的eventLoop被PausableChannelEventLoop包装，这样设置isAcceptingNewTasks=false时，新任务将被拒绝。这在关闭channel的时候非常有用  
5.      if (AbstractChannel.this.eventLoop == null) \{  
6.          AbstractChannel.this.eventLoop = new PausableChannelEventLoop(eventLoop);  
7.      \} else \{  
8.          AbstractChannel.this.eventLoop.unwrapped = eventLoop;  
9.      \}  
10.   
11.     if (eventLoop.inEventLoop()) \{  
12.         register0(promise);  
13.     \} else \{  
14.         try \{  
15.             eventLoop.execute(new OneTimeTask() \{  
16.                 @Override  
17.                 public void run() \{  
18.                     register0(promise);  
19.                 \}  
20.             \});  
21.         \} catch (Throwable t) \{  
22.             logger.warn(  
23.                     "Force-closing a channel whose registration task was not accepted by an event loop: \{\}",  
24.                     AbstractChannel.this, t);  
25.             closeForcibly();  
26.             closeFuture.setClosed();  
27.             safeSetFailure(promise, t);  
28.         \}  
29.     \}  
30. \}

上面的代码最重要的部分就是PausableChannelEventLoop的封装，接下来调用register0。

**\[java\]**  [view plain][]  [copy][view plain]

1.  private void register0(ChannelPromise promise) \{  
2.      try \{  
3.          // check if the channel is still open as it could be closed in the mean time when the register  
4.          // call was outside of the eventLoop  
5.          if (!promise.setUncancellable() || !ensureOpen(promise)) \{  
6.              return;  
7.          \}  
8.          boolean firstRegistration = neverRegistered;  
9.          // 真正的注册方法  
10.         doRegister();  
11.         neverRegistered = false;  
12.         registered = true;  
13.         // 注册完成以后开启接受任务的开关  
14.         eventLoop.acceptNewTasks();  
15.         safeSetSuccess(promise);  
16.         // 触发channelRegistered事件  
17.         pipeline.fireChannelRegistered();  
18.         // 只有从未注册的channel才会触发channelActive，避免连接注销并重新注册时多次触发channelActive。  
19.         // 注意后面还会出现fireChannelActive方法的调用，正常的第一次启动应该是触发后面那个fireChannelActive而不是这个  
20.         if (firstRegistration && isActive()) \{  
21.             pipeline.fireChannelActive();  
22.         \}  
23.     \} catch (Throwable t) \{  
24.         // Close the channel directly to avoid FD leak.  
25.         closeForcibly();  
26.         closeFuture.setClosed();  
27.         safeSetFailure(promise, t);  
28.     \}  
29. \}

doRegister方法将调用sun.nio.ch.ServerSocketChannelImpl.register方法，该方法将ServerSocketChannel注册到Selector上，因为传入的ops=0,此时并不会有连接进来（到目前为止都还没有与实际的端口进行绑定）。

**\[java\]**  [view plain][]  [copy][view plain]

1.  protected void doRegister() throws Exception \{  
2.      boolean selected = false;  
3.      for (;;) \{  
4.          try \{  
5.              selectionKey = javaChannel().register(((NioEventLoop) eventLoop().unwrap()).selector, 0, this);  
6.              return;  
7.          \} catch (CancelledKeyException e) \{  
8.              if (!selected) \{  
9.                  // 如果发送异常则强制执行Selector.selectNow()方法使 "canceled"的SelectionKey从Selector中移除  
10.                 ((NioEventLoop) eventLoop().unwrap()).selectNow();  
11.                 selected = true;  
12.             \} else \{  
13.                 //JDK bug ?  
14.                 throw e;  
15.             \}  
16.         \}  
17.     \}  
18. \}

注册完成后调用pipeline.fireChannelRegistered(); 该方法最终会是pipeline的处理链进行链式处理，在本例中他会触发两个操作：1、LogginHandler中的channelRegistered；2、在ServerBootstrap.init(Channel)方法中的代码：

**\[java\]**  [view plain][]  [copy][view plain]

1.  p.addLast(new ChannelInitializer<Channel>() \{  
2.         @Override  
3.         public void initChannel(Channel ch) throws Exception \{  
4.             // 触发这里执行  
5.             ch.pipeline().addLast(new ServerBootstrapAcceptor(  
6.                     currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));  
7.         \}  
8.     \});

ServerBootstrapAcceptor类主要作用是接收到客户端连接后，使用childOptions和childAttrs对连接初始化，然后将连接注册到childGroup中。ServerBootstrapAcceptor的channelRead方法如下：

**\[java\]**  [view plain][]  [copy][view plain]

1.  public void channelRead(ChannelHandlerContext ctx, Object msg) \{  
2.          final Channel child = (Channel) msg;  
3.    
4.          child.pipeline().addLast(childHandler);  
5.    
6.          for (Entry<ChannelOption<?>, Object> e: childOptions) \{  
7.              try \{  
8.                  if (!child.config().setOption((ChannelOption<Object>) e.getKey(), e.getValue())) \{  
9.                      logger.warn("Unknown channel option: " + e);  
10.                 \}  
11.             \} catch (Throwable t) \{  
12.                 logger.warn("Failed to set a channel option: " + child, t);  
13.             \}  
14.         \}  
15.   
16.         for (Entry<AttributeKey<?>, Object> e: childAttrs) \{  
17.             child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());  
18.         \}  
19.   
20.         try \{  
21.             childGroup.register(child).addListener(new ChannelFutureListener() \{  
22.                 @Override  
23.                 public void operationComplete(ChannelFuture future) throws Exception \{  
24.                     if (!future.isSuccess()) \{  
25.                         forceClose(child, future.cause());  
26.                     \}  
27.                 \}  
28.             \});  
29.         \} catch (Throwable t) \{  
30.             forceClose(child, t);  
31.         \}  
32.     \}

回到主流程，如果是第一次启动触发channelActive方法，本例中主要触发LoggerHandler.channelActive。调用完成后回到AbstractBootstrap.doBind0()方法：

**\[java\]**  [view plain][]  [copy][view plain]

1.  private static void doBind0(  
2.          final ChannelFuture regFuture, final Channel channel,  
3.          final SocketAddress localAddress, final ChannelPromise promise) \{  
4.      channel.eventLoop().execute(new Runnable() \{  
5.          @Override  
6.          public void run() \{  
7.              if (regFuture.isSuccess()) \{  
8.                  channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE\_ON\_FAILURE);  
9.              \} else \{  
10.                 promise.setFailure(regFuture.cause());  
11.             \}  
12.         \}  
13.     \});  
14. \}

doBind0最终调用channel.bind方法对执行端口进行监听。需要注意的是，为了保证线程安全，channel的所有方法都需要到EventLoop中执行。channel.bind最终调用AbstractChannel.AbstractUnsafe.bind(final SocketAddress localAddress, final ChannelPromise promise):

**\[java\]**  [view plain][]  [copy][view plain]

1.          public final void bind(final SocketAddress localAddress, final ChannelPromise promise) \{  
2.              // ---这里去掉了部分代码---  
3.    
4.              boolean wasActive = isActive();  
5.              try \{  
6.                  doBind(localAddress);  
7.              \} catch (Throwable t) \{  
8.                  safeSetFailure(promise, t);  
9.                  closeIfClosed();  
10.                 return;  
11.             \}  
12.   
13.             if (!wasActive && isActive()) \{  
14.                 // 增加一个任务，该任务触发pipeline.fireChannelActive方法, 该方法将最终触发channel.read()  
15.                 invokeLater(new OneTimeTask() \{  
16.                     @Override  
17.                     public void run() \{  
18.                         pipeline.fireChannelActive();  
19.                     \}  
20.                 \});  
21.             \}  
22.   
23.             safeSetSuccess(promise);  
24.         \}  
25.    
26.         // 最终调用socket的bind方式进行绑定，注意backlog在windows下默认为200，其他系统默认128  
27.         protected void doBind(SocketAddress localAddress) throws Exception \{  
28.             javaChannel().socket().bind(localAddress, config.getBacklog());  
29.         \}  
30.   
31.         // 上面的channel.read()最终会触发AbstractNioChannel.doBeginRead()方法  
32.         protected void doBeginRead() throws Exception \{  
33.             // Channel.read() or ChannelHandlerContext.read() was called   
34.             if (inputShutdown) \{              
35.                 return;  
36.             \}  
37.   
38.             final SelectionKey selectionKey = this.selectionKey;  
39.   
40.             if (!selectionKey.isValid()) \{  
41.                 return;  
42.             \}  
43.   
44.             readPending = true;  
45.   
46.             // 注册readInterestOp,ServerSocket关注的op为OP\_ACCEPT  
47.             final int interestOps = selectionKey.interestOps();  
48.             if ((interestOps & readInterestOp) == 0) \{  
49.                 selectionKey.interestOps(interestOps | readInterestOp);  
50.             \}  
51.   
52.         \}

到这里启动的步骤已经完成，我们再来回顾一下整个启动过程：

1、应用设置启动所需的各个参数

2、应用调用bind(port)启动监听，bind过程如下

3、验证启动参数设置是否正确，调用doBind

4、doBind创建NioServerSocketChannel,并对其进行初始化，包括创建一个实际的ServerSocket，设置其为非阻塞模式，创建底层处理实例NioMessageUnsafe，创建pipeline

5、pipeline中加入一个ChannelInitializer，该ChannelInitializer往pipleline中加入ServerBootstrapAcceptor用于接收客户连接后设置其初始化参数，然后注册到childGroup处理

6、将NioServerSocketChannel注册到bossGroup，此时bossGroup被激活开始接收任务及IO事件。

7、往EventLoop中添加一个任务，该任务的内容为将之前创建的ServerSocket绑定到指定端口。

8、绑定端口后增加一个任务，该任务内容为注册NioServerSocketChannel关注的事件OP\_ACCEPT到SelectKey中。到此，服务端可以接收到来自客户端的请求。

到此，ServerBootstrap的启动过程结束，服务端可以接收到客户端的连接请求。这里还有很多概念比较模糊，pipeline.addLast进行了什么操作，pipeline.channelXXX(如channelActive)是如何最终调用到channel的对应方法的。解开了这个问题，才能往下分析NioServerSocketChannel的请求接收、分发流程。ok, 下一篇文章就是对ChannelPipeline进行分析！

![Center][]

来一张图解解馋，netty本身还是很复杂的，该图进行了简化。

[view plain]: https://blog.csdn.net/youaremoon/article/details/50371221#
[Center]: /images/20220527/adcfa0b00c2944fb925dfbfa639ceda1.png