1
2
3
4
5
6
7
8

NioEventLoopGroup group = new NioEventLoopGroup(1);
       ServerBootstrap bootstrap = new ServerBootstrap();
       bootstrap.group(group)
               .channel(NioServerSocketChannel.class)
               .channel(NioServerSocketChannel.class)
               .option(ChannelOption.TCP_NODELAY, true)
               .option(ChannelOption.SO_BACKLOG, 1024)
               .childHandler(new ServerHandlerInitializer());

1
2
3
4
5
6
7

NioEventLoopGroup eventGroup = new NioEventLoopGroup();
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(eventGroup)
        .channel(NioServerSocketChannel.class)
        .option(ChannelOption.TCP_NODELAY, true)
        .option(ChannelOption.SO_BACKLOG, 1024)
        .childHandler(new ServerHandlerInitializer());

1
2
3
4
5
6
7
8

NioEventLoopGroup bossGroup = new NioEventLoopGroup();
NioEventLoopGroup workerGroup = new NioEventLoopGroup();
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup,workerGroup)
        .channel(NioServerSocketChannel.class)
        .option(ChannelOption.TCP_NODELAY, true)
        .option(ChannelOption.SO_BACKLOG, 1024)
        .childHandler(new ServerHandlerInitializer());

1
2
3
4
5

private Selector selector;
private Selector unwrappedSelector;
private SelectedSelectionKeySet selectedKeys;

private final SelectorProvider provider;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider,
             SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler,
             EventLoopTaskQueueFactory queueFactory) {
    super(parent, executor, false, newTaskQueue(queueFactory), newTaskQueue(queueFactory),
            rejectedExecutionHandler);
    if (selectorProvider == null) {
        throw new NullPointerException("selectorProvider");
    }
    if (strategy == null) {
        throw new NullPointerException("selectStrategy");
    }
    provider = selectorProvider;
    final SelectorTuple selectorTuple = openSelector();
    selector = selectorTuple.selector;
    unwrappedSelector = selectorTuple.unwrappedSelector;
    selectStrategy = strategy;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

protected void run() {
       for (;;) {
           try {
               try {
                   //通过hasTasks方法判断队列中是否还有未处理的方法
                   switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
                   case SelectStrategy.CONTINUE:
                       continue;

                   case SelectStrategy.BUSY_WAIT:
                       // fall-through to SELECT since the busy-wait is not supported with NIO
				//没有任务则执行，select()执行网络IO
                   case SelectStrategy.SELECT:
                       select(wakenUp.getAndSet(false));
                       if (wakenUp.get()) {
                           selector.wakeup();
                       }
                       // fall through
                   default:
                   }
               } catch (IOException e) {
                   //如果本轮Selector的轮询结果为null，那么可能触发了jdk epoll的bug
                   //该bug会导致IO线程处于100%的状态，需要重建Selector来解决
                   rebuildSelector0();
                   handleLoopException(e);
                   continue;
               }

               cancelledKeys = 0;
               needsToSelectAgain = false;
               //处理IO事件所需的事件和花费在处理task的时间的比例，默认为50%
               final int ioRatio = this.ioRatio;
               if (ioRatio == 100) {
                   try {
                       //如果比例为100.则表示每次处理完IO后，才开始处理task
                       processSelectedKeys();
                   } finally {
                       
                       // 执行task任务
                       runAllTasks();
                   }
               } else {
                   //记录处理IO的开始时间
                   final long ioStartTime = System.nanoTime();
                   try {
                       //处理IO请求
                       processSelectedKeys();
                   } finally {
                       //计算IO请求的耗时
                       final long ioTime = System.nanoTime() - ioStartTime;
                       //执行task。判断执行task任务时间是否超过配置的比例，如果超过则停止执行task
                       runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
                   }
               }
           } catch (Throwable t) {
               handleLoopException(t);
           }
           // Always handle shutdown even if the loop processing threw an exception.
           try {
               if (isShuttingDown()) {
                   closeAll();
                   if (confirmShutdown()) {
                       return;
                   }
               }
           } catch (Throwable t) {
               handleLoopException(t);
           }
       }
   }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

private void rebuildSelector0() {
      final Selector oldSelector = selector;
      final SelectorTuple newSelectorTuple;

      if (oldSelector == null) {
          return;
      }

      try {
          //创建一个新的Selector
          newSelectorTuple = openSelector();
      } catch (Exception e) {
          logger.warn("Failed to create a new Selector.", e);
          return;
      }

      // Register all channels to the new Selector.
      int nChannels = 0;
    
      for (SelectionKey key: oldSelector.keys()) {
          //将原Selector上注册的所有SelectionKey转移到新的Selector
          Object a = key.attachment();
          try {
              if (!key.isValid() || key.channel().keyFor(newSelectorTuple.unwrappedSelector) != null) {
                  continue;
              }

              int interestOps = key.interestOps();
              key.cancel();
              SelectionKey newKey = key.channel().register(newSelectorTuple.unwrappedSelector, interestOps, a);
              if (a instanceof AbstractNioChannel) {
                  // Update SelectionKey
                  ((AbstractNioChannel) a).selectionKey = newKey;
              }
              nChannels ++;
          } catch (Exception e) {
              logger.warn("Failed to re-register a Channel to the new Selector.", e);
              if (a instanceof AbstractNioChannel) {
                  AbstractNioChannel ch = (AbstractNioChannel) a;
                  ch.unsafe().close(ch.unsafe().voidPromise());
              } else {
                  @SuppressWarnings("unchecked")
                  NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
                  invokeChannelUnregistered(task, key, e);
              }
          }
      }

    //用新的Selector替换旧的
      selector = newSelectorTuple.selector;
      unwrappedSelector = newSelectorTuple.unwrappedSelector;

      try {
          // time to close the old selector as everything else is registered to the new one
          //关闭旧的Selector
          oldSelector.close();
      } catch (Throwable t) {
          if (logger.isWarnEnabled()) {
              logger.warn("Failed to close the old Selector.", t);
          }
      }

      if (logger.isInfoEnabled()) {
          logger.info("Migrated " + nChannels + " channel(s) to the new Selector.");
      }
  }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
    final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();

    // 省略代码 ......

    try {
        int readyOps = k.readyOps();
        // We first need to call finishConnect() before try to trigger a read(...) or write(...) as otherwise
        // the NIO JDK channel implementation may throw a NotYetConnectedException.
        if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
            // remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
            // See https://github.com/netty/netty/issues/924
            int ops = k.interestOps();
            ops &= ~SelectionKey.OP_CONNECT;
            k.interestOps(ops);

            unsafe.finishConnect();
        }


        if ((readyOps & SelectionKey.OP_WRITE) != 0) {
            // Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to write
            ch.unsafe().forceFlush();
        }

        if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
            unsafe.read();
        }
    } catch (CancelledKeyException ignored) {
        unsafe.close(unsafe.voidPromise());
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

private final class NioMessageUnsafe extends AbstractNioUnsafe {

        @Override
        public void read() {
            // 省略代码 ......
            // 获取 Channel 对应的 ChannelPipeline
            final ChannelPipeline pipeline = pipeline();

            boolean closed = false;
            Throwable exception = null;
            try {
                // 省略代码 ......
                int size = readBuf.size();
                for (int i = 0; i < size; i ++) {
                    readPending = false;
                    // 委托给 pipeline 中的 Handler 进行读取数据
                    pipeline.fireChannelRead(readBuf.get(i));
                }