netty5笔记-总体流程分析3-NioServerSocketChannel

阳光穿透心脏的1/2处 2022-05-27 12:40 170阅读 0赞

前面我们讲了server的启动过程，同时介绍了其中非常重要的几个概念,ChannelPipeline,ChannelHandlerContext等。接下来我们看看server启动起来以后是如何运转的。

先回忆下之前的几个重要的点，如果要监听某个端口，首先用ServerBootstrap引导启动，启动时创建一个ServerSocketChannel，并注册到bossGroup的EventLoop中，关注的事件为OP\_ACCEPT。boss EventLoop开始运行，并不停的尝试从Selector中查看是否有准备好的连接。由于ServerSocketChannel关注的是OP\_ACCEPT，因此每当有客户端连接到服务端的时候，boss EventLoop都可以select到一个SelectionKey，然后进入以下方法：

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

1.  //NioEventLoop  
2.  private static void processSelectedKey(SelectionKey k, AbstractNioChannel ch) \{  
3.      final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();  
4.      if (!k.isValid()) \{  
5.          // 如果key无效了则关闭连接  
6.          unsafe.close(unsafe.voidPromise());  
7.          return;  
8.      \}  
9.    
10.     try \{  
11.         int readyOps = k.readyOps();  
12.         // 检查readOps，避免由jdk bug导致readOps为0而产生自旋(cpu 100%)  
13.         if ((readyOps & (SelectionKey.OP\_READ | SelectionKey.OP\_ACCEPT)) != 0 || readyOps == 0) \{  
14.             // 调用unsafe.read方法，这个也是我们今天分析的入口  
15.             unsafe.read();  
16.             if (!ch.isOpen()) \{  
17.                 // 连接已经关闭则直接返回，就不用处理后面的write事件了  
18.                 return;  
19.             \}  
20.         \}  
21.         。。。省略不会触发的几句代码。。。  
22.         if ((readyOps & SelectionKey.OP\_CONNECT) != 0) \{  
23.             // 取消此key对 OP\_CONNECT的关注，否则Selector.select(..)不会阻塞而直接返回，导致cpu 100%  
24.             // 此bug见 https://github.com/netty/netty/issues/924  
25.             int ops = k.interestOps();  
26.             ops &= ~SelectionKey.OP\_CONNECT;  
27.             k.interestOps(ops);  
28.   
29.             unsafe.finishConnect();  
30.         \}  
31.     \} catch (CancelledKeyException ignored) \{  
32.         unsafe.close(unsafe.voidPromise());  
33.     \}  
34. \}

注意还有一种情况是processSelectedKey(SelectionKey k, NioTask<SelectableChannel> task)，这是扩展的情况，暂时无人实现，先不分析。

这里引出了本次分析的第一个入口点，ch.unsafe().read()。NioServerSocketChannel对应的unsafe实现为NioMessageUnsafe，我们来看看它的read方法做了哪些事情（去掉部分无关代码）：

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

1.          // NioMessageUnsafe  
2.          private final List<Object> readBuf = new ArrayList<Object>();  
3.    
4.          public void read() \{  
5.              final ChannelConfig config = config();  
6.              // 下面有个循环操作，这里的值代表循环的最大次数，对于NioServerSocketChannel来说也就是单次最大接受的连接数，默认为16，  
7.              // 可以在启动的时候通过初始化时调用引导类的setOption(ChannelOption.MAX\_MESSAGES\_PER\_READ, 32)修改这个值。  
8.              final int maxMessagesPerRead = config.getMaxMessagesPerRead();  
9.              final ChannelPipeline pipeline = pipeline();  
10.             boolean closed = false;  
11.             Throwable exception = null;  
12.             try \{  
13.                 try \{  
14.                     for (;;) \{  
15.                         // 此处会调用到NioServerSocketChannel中的doReadMessages方法  
16.                         int localRead = doReadMessages(readBuf);  
17.                         // 读到的值为0没获取到连接（可能是已经关闭了），注意NioServerSocketChannel中的doReadMessages只会返回0,1，     
18.                         // -1在其他场景中出现，后面再分析                      
19.                         if (localRead == 0) \{  
20.                             break;  
21.                         \}  
22.                           
23.                         // 每次读取成功readBuf中就会多一个连接，达到阈值则先跳出循环，剩下的数据下次循环时再取  
24.                         if (readBuf.size() >= maxMessagesPerRead) \{  
25.                             break;  
26.                         \}  
27.                     \}  
28.                 \} catch (Throwable t) \{  
29.                     exception = t;  
30.                 \}  
31.                 setReadPending(false);  
32.                 int size = readBuf.size();  
33.                 for (int i = 0; i < size; i ++) \{  
34.                     // 对每个连接调用pipeline的fireChannelRead  
35.                     pipeline.fireChannelRead(readBuf.get(i));  
36.                 \}  
37.                 // 清理获取到的数据，下次继续使用该buf  
38.                 readBuf.clear();  
39.                 pipeline.fireChannelReadComplete();  
40.                 // 如果在接收连接时发生了错误，触发fireExceptionCaught事件  
41.                 if (exception != null) \{  
42.                     。。。。。。  
43.   
44.                     pipeline.fireExceptionCaught(exception);  
45.                 \}  
46.                 。。。。。。  
47.             \} finally \{  
48.                 // 如果非autoRead则移除关注的事件  
49.                 if (!config.isAutoRead() && !isReadPending()) \{  
50.                     removeReadOp();  
51.                 \}  
52.             \}  
53.         \}

NioMessageUnsafe还是其他Channel的unsafe实现，这里只把NioServerSocketChannel相关的部分列了出来，逻辑很简单，循环的通过doReadMessages方法读取指定个数的连接到list中（如果其中一次没读取到也会终止循环），完成后对list中的连接依次调用pipeline.fireChannelRead方法。NioServerSocketChannel的doReadMessages比较简单：

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

1.  protected int doReadMessages(List<Object> buf) throws Exception \{  
2.      SocketChannel ch = javaChannel().accept();  
3.    
4.      try \{  
5.          if (ch != null) \{  
6.              // accept到连接则新建一个NioSocketChannel的封装类，并返回读取到的消息数1，否则返回0   
7.             buf.add(new NioSocketChannel(this, ch));  
8.              return 1;  
9.          \}  
10.     \} catch (Throwable t) \{  
11.         logger.warn("Failed to create a new channel from an accepted socket.", t);  
12.   
13.         try \{  
14.             // 创建失败则关闭改连接，这个连接指客户端连接              
15.             ch.close();  
16.         \} catch (Throwable t2) \{  
17.             logger.warn("Failed to close a socket.", t2);  
18.         \}  
19.     \}  
20.   
21.     return 0;  
22. \}

NioSocketChannel的创建过程我们后面再讲，先往后看，pipeline.fireChannelRead触发了什么方法。我们知道channelRead属于inbound时间，因此是从HeadContext->TailContext处理。HeadContext的fireChannelRead方法只是简单的往后传递事件，而TailContext则是对需要释放资源的对象进行释放，因此主要逻辑不用关注这两块。回到ServerBootstrap，在启动时默认会往pipeline中加入一个ServerBootstrapAcceptor（其他用户自定义加入的ChannelHandler如LoggingHandler我们暂不分析）。

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

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

fireChannelRead主要的实现就是它了：

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

1.      // ServerBootstrapAcceptor  
2.      public void channelRead(ChannelHandlerContext ctx, Object msg) \{  
3.          // child是对socket的一个包装、增强  
4.          final Channel child = (Channel) msg;  
5.          // 引导类启动时设置的childHandler加到child的pipeline中  
6.          child.pipeline().addLast(childHandler);  
7.          // 将childOptions中的配置设置到child的option中  
8.          for (Entry<ChannelOption<?>, Object> e: childOptions) \{  
9.              try \{  
10.                 if (!child.config().setOption((ChannelOption<Object>) e.getKey(), e.getValue())) \{  
11.                     logger.warn("Unknown channel option: " + e);  
12.                 \}  
13.             \} catch (Throwable t) \{  
14.                 logger.warn("Failed to set a channel option: " + child, t);  
15.             \}  
16.         \}  
17.         // 将childAttrs中的属性设置到child的属性中  
18.         for (Entry<AttributeKey<?>, Object> e: childAttrs) \{  
19.             child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());  
20.         \}  
21.         // 将连接注册到childGroup中（也就是我们常说的workGroup)，注册完成如果发现注册失败则关闭此链接  
22.         try \{  
23.             childGroup.register(child).addListener(new ChannelFutureListener() \{  
24.                 @Override  
25.                 public void operationComplete(ChannelFuture future) throws Exception \{  
26.                     if (!future.isSuccess()) \{  
27.                         forceClose(child, future.cause());  
28.                     \}  
29.                 \}  
30.             \});  
31.         \} catch (Throwable t) \{  
32.             forceClose(child, t);  
33.         \}  
34.     \}  
35.   
36.     private static void forceClose(Channel child, Throwable t) \{  
37.         child.unsafe().closeForcibly();  
38.         logger.warn("Failed to register an accepted channel: " + child, t);  
39.     \}  
40.   
41.     @Override  
42.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception \{  
43.         final ChannelConfig config = ctx.channel().config();  
44.         if (config.isAutoRead()) \{  
45.             // 先设置为false，不再接收连接  
46.             config.setAutoRead(false);  
47.             // 1秒后设置回来  
48.             ctx.channel().eventLoop().schedule(new Runnable() \{  
49.                 public void run() \{  
50.                     config.setAutoRead(true);  
51.                 \}  
52.             \}, 1, TimeUnit.SECONDS);  
53.         \}  
54.           
55.         // 传递异常事件，这样用户可以自定义处理方法  
56.         ctx.fireExceptionCaught(cause);  
57.     \}  
58. \}

需要注意上面的exceptionCaught方法，从前面的NioMessageUnsafe.read()可以看到，其中的一个调用javaChannel.accept()并没有catch相应的异常，它可能会触发很多种异常，我们需要关注的比如IOException: Too many open files。对于这种情况，如果server继续接收连接则会触发更多的Too many open files，因此此处将autoRead设置为false，在设置为false后，NioMessageUnsafe的read方法会在finally中移除channel关注的事件：OP\_CONNECT，这样server就不会再接收连接了。 同时也会添加一个计划任务，该任务1秒后执行，内容为重新将autoRead设为true。设置autoRead为true时会触发channel.read()方法，read方法接着触发pipeline.read()，最终触发HeadContext的read()，此方法调用了unsafe.beginRead()， beginRead将会再次将OP\_CONNECT设置为关注的事件，此时可以继续接收客户端连接。

上面的channelRead方法通过childGroup.register(child)将客户端连接注册到了workGroup上，跟踪该方法进入到了SingleThreadEventLoop：

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

1.  public ChannelFuture register(final Channel channel, final ChannelPromise promise) \{  
2.      。。。  
3.    
4.      channel.unsafe().register(this, promise);  
5.      return promise;  
6.  \}

即调用了NioSocketChannel的unsafe.register, 此处的unsafe实现类为NioByteUnsafe，对应的register方法在其父类AbstractUnsafe中实现:

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

1.          public final void register(EventLoop eventLoop, final ChannelPromise promise) \{  
2.              if (eventLoop == null) \{  
3.                  throw new NullPointerException("eventLoop");  
4.              \}  
5.              if (promise == null) \{  
6.                  throw new NullPointerException("promise");  
7.              \}  
8.              // 一个连接不能多次注册，防止注册到其他EventLoop出现并发问题  
9.              if (isRegistered()) \{  
10.                 promise.setFailure(new IllegalStateException("registered to an event loop already"));  
11.                 return;  
12.             \}  
13.             // 这里会判断是否时NioEventLoop，NioSocketChannel只能注册到NioEventLoop或者其子类  
14.             if (!isCompatible(eventLoop)) \{  
15.                 promise.setFailure(  
16.                         new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));  
17.                 return;  
18.             \}  
19.   
20.             // It's necessary to reuse the wrapped eventloop object. Otherwise the user will end up with multiple  
21.             // objects that do not share a common state.  
22.             // PausableChannelEventLoop提供了一个开关isAcceptingNewTasks，isAcceptingNewTasks=false时不接收新任务  
23.             if (AbstractChannel.this.eventLoop == null) \{  
24.                 AbstractChannel.this.eventLoop = new PausableChannelEventLoop(eventLoop);  
25.             \} else \{  
26.                 AbstractChannel.this.eventLoop.unwrapped = eventLoop;  
27.             \}  
28.   
29.             // 线程安全的调用register0  
30.             if (eventLoop.inEventLoop()) \{  
31.                 register0(promise);  
32.             \} else \{  
33.                 try \{  
34.                     eventLoop.execute(new OneTimeTask() \{  
35.                         @Override  
36.                         public void run() \{  
37.                             register0(promise);  
38.                         \}  
39.                     \});  
40.                 \} catch (Throwable t) \{  
41.                     logger.warn(  
42.                             "Force-closing a channel whose registration task was not accepted by an event loop: \{\}",  
43.                             AbstractChannel.this, t);  
44.                     closeForcibly();  
45.                     closeFuture.setClosed();  
46.                     safeSetFailure(promise, t);  
47.                 \}  
48.             \}  
49.         \}  
50.   
51.         private void register0(ChannelPromise promise) \{  
52.             try \{  
53.                 // 将对应的promise设为不可取消  
54.                 if (!promise.setUncancellable() || !ensureOpen(promise)) \{  
55.                     return;  
56.                 \}  
57.                 // 调用doRegister进行注册  
58.                 boolean firstRegistration = neverRegistered;  
59.                 doRegister();  
60.                 neverRegistered = false;  
61.                 // register状态设置为true，表示已经注册到EventLoop中  
62.                 registered = true;  
63.                 // eventLoop的isAcceptingNewTasks开关打开  
64.                 eventLoop.acceptNewTasks();  
65.                 safeSetSuccess(promise);  
66.                 // 触发channelRegistered事件  
67.                 pipeline.fireChannelRegistered();  
68.                 // 第一次注册时触发fireChannelActive事件，防止deregister后再次register触发多次fireChannelActive调用  
69.                 if (firstRegistration && isActive()) \{  
70.                     // 这里和前面的ServerSocketChannel分析一样,最终会触发unsafe.beginRead()  
71.                     pipeline.fireChannelActive();  
72.                 \}  
73.             \} catch (Throwable t) \{  
74.                 // Close the channel directly to avoid FD leak.  
75.                 closeForcibly();  
76.                 closeFuture.setClosed();  
77.                 safeSetFailure(promise, t);  
78.             \}  
79.         \}  
80.   
81.     // AbstractNioChannel  
82.     protected void doRegister() throws Exception \{  
83.         boolean selected = false;  
84.         for (;;) \{  
85.             try \{  
86.                 // 将连接注册到selector，此时虽然有注册，但ops为0，即没有关注的事件  
87.                 selectionKey = javaChannel().register(((NioEventLoop) eventLoop().unwrap()).selector, 0, this);  
88.                 return;  
89.             \} catch (CancelledKeyException e) \{  
90.                 if (!selected) \{  
91.                     // 强制Selector调用select now，防止取消的SelectionKey未真正取消（因为还没有调用到Select.select(..))  
92.                     ((NioEventLoop) eventLoop().unwrap()).selectNow();  
93.                     selected = true;  
94.                 \} else \{  
95.                     // We forced a select operation on the selector before but the SelectionKey is still cached  
96.                     // for whatever reason. JDK bug ?  
97.                     throw e;  
98.                 \}  
99.             \}  
100.         \}  
101.     \}

以HttpSnoopServer为例，childHandler为HttpSnoopServerInitializer，初始化是NioSocketChannel的pipeline中只有tail,head及HttpSnoopServerInitializer的实例，tail和head中的channelRegisterd除了传递或者停止外没有其他用处，因此此处的channelRegisterd主要逻辑在HttpSnoopServerInitializer的channelRegistered方法：

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

1.      // 此方法在父类ChannelInitializer中  
2.      public final void channelRegistered(ChannelHandlerContext ctx) throws Exception \{  
3.          ChannelPipeline pipeline = ctx.pipeline();  
4.          boolean success = false;  
5.          try \{  
6.              // initChannel主要是初始化channel对应的pipeline  
7.              initChannel((C) ctx.channel());  
8.              // 初始化完成后将自身移除  
9.              pipeline.remove(this);  
10.             // channelRegistered属于inbound事件，注册后调用一次该方法，这样所有用户添加的handler可以感知到此事件  
11.             ctx.fireChannelRegistered();  
12.             success = true;  
13.         \} catch (Throwable t) \{  
14.             logger.warn("Failed to initialize a channel. Closing: " + ctx.channel(), t);  
15.         \} finally \{  
16.             if (pipeline.context(this) != null) \{  
17.                 pipeline.remove(this);  
18.             \}  
19.             if (!success) \{  
20.                 ctx.close();  
21.             \}  
22.         \}  
23.     \}  
24.       
25.     // 此方法在HttpSnoopServerInitializer中  
26.     public void initChannel(SocketChannel ch) \{  
27.         ChannelPipeline p = ch.pipeline();  
28.         if (sslCtx != null) \{  
29.             p.addLast(sslCtx.newHandler(ch.alloc()));  
30.         \}  
31.         p.addLast(new HttpRequestDecoder());  
32.         // Uncomment the following line if you don't want to handle HttpChunks.  
33.         //p.addLast(new HttpObjectAggregator(1048576));  
34.         p.addLast(new HttpResponseEncoder());  
35.         // Remove the following line if you don't want automatic content compression.  
36.         //p.addLast(new HttpContentCompressor());  
37.         p.addLast(new HttpSnoopServerHandler());  
38.     \}

而pipeline.fireChannelActive()最终会调用unsafe.beginRead()，我们看看NioSocketChannel的unsafe.beginRead()：

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

1.          public final void beginRead() \{  
2.              if (!isActive()) \{  
3.                  return;  
4.              \}  
5.    
6.              try \{  
7.                  doBeginRead();  
8.              \} catch (final Exception e) \{  
9.                  invokeLater(new OneTimeTask() \{  
10.                     public void run() \{  
11.                         pipeline.fireExceptionCaught(e);  
12.                     \}  
13.                 \});  
14.                 close(voidPromise());  
15.             \}  
16.         \}  
17.   
18.     // AbstractNioChannel  
19.     protected void doBeginRead() throws Exception \{  
20.         if (inputShutdown) \{  
21.             return;  
22.         \}  
23.   
24.         final SelectionKey selectionKey = this.selectionKey;  
25.         if (!selectionKey.isValid()) \{  
26.             return;  
27.         \}  
28.   
29.         readPending = true;  
30.         // 注册NioSocketChannel关注的事件（NioSocketChannel初始关注的事件为OP\_READ)  
31.         final int interestOps = selectionKey.interestOps();  
32.         if ((interestOps & readInterestOp) == 0) \{  
33.             selectionKey.interestOps(interestOps | readInterestOp);  
34.         \}  
35.     \}

到这里，一个NioSocketChannel已经可以读取数据了。 我们总结下这个过程：

1、bossGroup通过不停的selector.select或者selectNow获取到客户端连接（每个循环获取多个连接，批量处理），将socket包装为NioSocketChannel；

2、将新增的NioSocketChannel注册到workGroup中，该注册会触发以下操作；

2.1、首先将对应socket注册到workGroup对应的Selector中，此时关注的事件为空；

2.2、触发fireChannelRegistered事件， 该方法用用户自定义的handler及顺序来初始化channel的pipeline；

2.3、触发fireChannelActive事件，最终调用unsafe.beginRead()，beginRead方法设置NioSocketChannel关注的事件(OP\_READ)。

过程比较简单，而且第2步的操作都是在workGroup中进行的，因此bossGroup的工作非常简单，所以效率很高。

[view plain]: https://blog.csdn.net/youaremoon/article/details/50488092#