netty源码阅读之客户端新连接之检测新连接

雨点打透心脏的1/2处 2022-05-17 12:18 221阅读 0赞

boss线程的NioEventLoop在用户代码调用bond的时候启动。

检测新连接我们从《netty源码阅读之NioEventLoop之NioEventLoop执行——-processSelectedKey()执行》这一篇文章的这个方法入口：processSelectedKeysOptimized(SelectionKey[] selectedKeys)

看到这一步：

  1. final Object a = k.attachment();
  3.             if (a instanceof AbstractNioChannel) {
  4.                 processSelectedKey(k, (AbstractNioChannel) a);
  5.             } else {
  6.                 @SuppressWarnings("unchecked")
  7.                 NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
  8.                 processSelectedKey(k, task);
  9.             }

之前我们的NioSocket以attachment的方式传入select key set里面，所以我们进入

  1. processSelectedKey(k, (AbstractNioChannel) a);

这个方法里面。

  1. private void processSelectedKeysOptimized(SelectionKey[] selectedKeys) {
  2.         for (int i = 0;; i ++) {
  3.             final SelectionKey k = selectedKeys[i];
  4.             if (k == null) {
  5.                 break;
  6.             }
  7.             // null out entry in the array to allow to have it GC'ed once the Channel close
  8.             // See https://github.com/netty/netty/issues/2363
  9.             selectedKeys[i] = null;
  11.             final Object a = k.attachment();
  13.             if (a instanceof AbstractNioChannel) {
  14.                 processSelectedKey(k, (AbstractNioChannel) a);
  15.             } else {
  16.                 @SuppressWarnings("unchecked")
  17.                 NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
  18.                 processSelectedKey(k, task);
  19.             }
  21.             if (needsToSelectAgain) {
  22.                 // null out entries in the array to allow to have it GC'ed once the Channel close
  23.                 // See https://github.com/netty/netty/issues/2363
  24.                 for (;;) {
  25.                     i++;
  26.                     if (selectedKeys[i] == null) {
  27.                         break;
  28.                     }
  29.                     selectedKeys[i] = null;
  30.                 }
  32.                 selectAgain();
  33.                 // Need to flip the optimized selectedKeys to get the right reference to the array
  34.                 // and reset the index to -1 which will then set to 0 on the for loop
  35.                 // to start over again.
  36.                 //
  37.                 // See https://github.com/netty/netty/issues/1523
  38.                 selectedKeys = this.selectedKeys.flip();
  39.                 i = -1;
  40.             }
  41.         }
  42.     }
  44.     private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
  45.         final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
  46.         ..
  47.         try {
  48.             int readyOps = k.readyOps();
  49.             // We first need to call finishConnect() before try to trigger a read(...) or write(...) as otherwise
  50.             // the NIO JDK channel implementation may throw a NotYetConnectedException.
  51.             if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
  52.                 ...
  53.             }
  55.             // Process OP_WRITE first as we may be able to write some queued buffers and so free memory.
  56.             if ((readyOps & SelectionKey.OP_WRITE) != 0) {
  57.                 ...
  58.             }
  60.             // Also check for readOps of 0 to workaround possible JDK bug which may otherwise lead
  61.             // to a spin loop
  62.             if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
  63.                 unsafe.read();
  64.                 if (!ch.isOpen()) {
  65.                     // Connection already closed - no need to handle write.
  66.                     return;
  67.                 }
  68.             }
  69.         } catch (CancelledKeyException ignored) {
  70.             unsafe.close(unsafe.voidPromise());
  71.         }
  72.     }

由于服务端的socketChannel注册的是accept事件，所以我们会进入这个：

  1. // Also check for readOps of 0 to workaround possible JDK bug which may otherwise lead
  2.             // to a spin loop
  3.             if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
  4.                 unsafe.read();
  5.                 if (!ch.isOpen()) {
  6.                     // Connection already closed - no need to handle write.
  7.                     return;
  8.                 }
  9.             }

NioMessageUnsafe.read()

服务端NioServerSocketChannel的unsafe的实现是NioMessageUnsafe（后续会有文章介绍unsafe），所以我们到达了这个方法:

  1. private final class NioMessageUnsafe extends AbstractNioUnsafe {
  3.         private final List<Object> readBuf = new ArrayList<Object>();
  5.         @Override
  6.         public void read() {
  7.             assert eventLoop().inEventLoop();
  8.             final ChannelConfig config = config();
  9.             final ChannelPipeline pipeline = pipeline();
  10.             final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
  11.             allocHandle.reset(config);
  13.             boolean closed = false;
  14.             Throwable exception = null;
  15.             try {
  16.                 try {
  17.                     do {
  18.                         int localRead = doReadMessages(readBuf);
  19.                         if (localRead == 0) {
  20.                             break;
  21.                         }
  22.                         if (localRead < 0) {
  23.                             closed = true;
  24.                             break;
  25.                         }
  27.                         allocHandle.incMessagesRead(localRead);
  28.                     } while (allocHandle.continueReading());
  29.                 } catch (Throwable t) {
  30.                     exception = t;
  31.                 }
  33.                 int size = readBuf.size();
  34.                 for (int i = 0; i < size; i ++) {
  35.                     readPending = false;
  36.                     pipeline.fireChannelRead(readBuf.get(i));
  37.                 }
  38.                 ...
  40.                 if (closed) {
  41.                     inputShutdown = true;
  42.                     if (isOpen()) {
  43.                         close(voidPromise());
  44.                     }
  45.                 }
  46.             } finally {
  47.                 ...
  48.             }
  49.         }
  50.     }

这个方法里面有个doMessageRead()，是主要的读方法，通过while循环不停地读数据。

这个final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();代码里面有个handle，是读的策略（控制连接接入的速度，每次有一个新连接就加1，默认最大只读16个连接）。在这个

allocHandle.continueReading()里面，可以知道它什么时候停下来：

  1. @Override
  2.         public boolean continueReading() {
  3.             return config.isAutoRead() &&
  4.                    attemptedBytesRead == lastBytesRead &&
  5.                    totalMessages < maxMessagePerRead &&
  6.                    totalBytesRead < Integer.MAX_VALUE;
  7.         }

totalMessages<maxMessagePerRead是判断读到的连接是否大于最大连接，maxMessagePerRead默认是16。

我们继续doMessageRead()吧。

doMessageRead()

  1. @Override
  2.     protected int doReadMessages(List<Object> buf) throws Exception {
  3.         SocketChannel ch = javaChannel().accept();
  5.         try {
  6.             if (ch != null) {
  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);
  13.             try {
  14.                 ch.close();
  15.             } catch (Throwable t2) {
  16.                 logger.warn("Failed to close a socket.", t2);
  17.             }
  18.         }
  20.         return 0;
  21.     }

在这里面，SocketChannel是通过jdk实现的。然后这个buf.add(new NioSocketChannel(this, ch));新建一个客户端的NioSocketChannel，把jdk实现得SocketChannel绑定进去(把jdk底层的channel封装进去)，并且把this，也就是服务端NioServerSocketChannel也作为参数传递进去。

buf是之前传入进来的，其实他就是之前的readBuf，一个list，保存的是客户端的NioSocketChannel：

  1. private final List<Object> readBuf = new ArrayList<Object>();

下一篇文章，我们分析new NioSocketChannel(this, ch)这一步做了什么事情。

发表评论

表情：
评论列表 （有 0 条评论，221人围观）

还没有评论，来说两句吧...

相关阅读

    相关 netty源码阅读之ByteBuf

    今天我们开启新的篇章，netty很重要的内存概念将在这一章介绍。ByteBuf主要介绍以下几点： 1、内存与内存管理器的抽象 2、不同规格大小和不同类别的内存的分配策略

    野性酷女野性酷女/ 0 赞/ 313 阅读