• 4、Netty 线程模型

    netty提供2种IO模型，nio线程模型、oio线程模型。这两个模型到底有什么区别？ nio线程模型：有专门接受请求的线程，他只接受并不处理。 oio线程模型：线程接受任务后并且处理掉任务。nio线程模型、oio线程模型netty有提供线程池的概念。在oio模型处理接受请求的线程组可以并行处理任务，在nio线程模型有专门的boss线程组、worker线程组。所以，netty的线程模型是2类线程模型，4种线程模型可能性。使用oio线程模型的netty并发要比使用nio线程模型的差很多。

    IO类型

    IO形式

    通讯特点

    oio线程模型

    oio单线程模型

    同时只容许1个客户端与服务端通讯

    oio线程池模型

    同时容许<=线程池数量的客户端数与服务端通讯

    nio线程模型

    nio单线程模型

    同时只容许1个客户端与服务端通讯,是否是netty BUG

    nio线程池模型

    可以同时容许多个客户端通讯

    一、OIO单线程模型

    1.1、如何实现netty oio单线程模型？

    boss线程组、worker线程组都设置为单线程模式。

    1. /**
    2.  * boss线程组
    3. */
    4. EventLoopGroup bossGroup = new OioEventLoopGroup(0,Executors.newSingleThreadExecutor());
    5.  
    6. /**
    7.  * worker线程组
    8.  */
    9. EventLoopGroup workGroup = new OioEventLoopGroup(0,Executors.newSingleThreadExecutor());
    10.  
    11.  
    12. //设置nio类型的channel
    13. b.channel(OioServerSocketChannel.class);

    1.2、OIO单线程模型效果

    从截图可以看出，oio单线程模型只容许一个客户端和服务端通讯。当有新服务端加入通讯，会被阻塞。

    1.3、OIO线程模型如何实现？

    1. //OioServerSocketChannel.java 
    2. @Override
    3. protected int doReadMessages(List buf) throws Exception {
    4.         if (socket.isClosed()) {
    5.             return -1;
    6.         }
    7.  
    8.         try {
    9.             //netty oio模型使用bio通讯的证据
    10.             Socket s = socket.accept();
    11.             try {
    12.                 //这块吧socket给线程池去处理，线程池线程数，可以自定义处理
    13.                 buf.add(new OioSocketChannel(this, s));
    14.                 return 1;
    15.             } catch (Throwable t) {
    16.                 logger.warn("Failed to create a new channel from an accepted socket.", t);
    17.                 try {
    18.                     s.close();
    19.                 } catch (Throwable t2) {
    20.                     logger.warn("Failed to close a socket.", t2);
    21.                 }
    22.             }
    23.         } catch (SocketTimeoutException e) {
    24.             // Expected
    25.         }
    26.         return 0;
    27.     }
    28. }
      29.  

       

      二、oio线程池模型
       

      这块就不演示了oio通过OioSocketChannel包装，所有的socket处理请求是OioSocketChannel处理，其处理委托给一个线程池处理。
       

      2.2、oio多线程模型效果
       

      从截图可以看出，oio多线程模型（具体2个线程）只容许2个客户端和服务端通讯。当有新服务端加入通讯，会被阻塞。
       

       

      三、nio线程模型
       

      nio线程模型，其是基于jdk nio io模型的。但是nio模型EventLoopGroup work线程组设置不能是单线程，否则不支持同时处理多个客户端的能力。这在nio是支持的，这个可能是netty的一个bug？还搞不清楚。
       

      1. //AbstractNioChannel.java 
      2. protected void doRegister() throws Exception {
      3.         boolean selected = false;
      4.         for (;;) {
      5.             try {
      6.                 selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 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. }
       

      四、netty线程模型总结
       

      netty其实就是一个单线程、多线程的外壳，包裹这jdk的 bio、nio实现。bio虽然是同步阻塞io，但是如果有多线程支持一样的可以做到同时支持多个客户端。nio天生就支持多个客户端，netty将接受请求线程、工作线程分开处理。
       

      五、boss线程调用链
       

      5.1、boss线程池初始化
       

      EventLoopGroup[boss、worke线程组] -->NioEventLoopGroup[nio线程组]-->NioEventLoop.openSelector()方法打开select监听.
       

      1. public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {
      2.    //boss线程继续执行
      3.    super.group(parentGroup);
      4.    if (this.childGroup != null) {
      5.             throw new IllegalStateException("childGroup set already");
      6.    }
      7.    this.childGroup = ObjectUtil.checkNotNull(childGroup, "childGroup");
      8.    return this;
      9. }
      10.  
      11. //获取selector，并对其进行优化,不同操作系统jdk实现不同
      12. //NioEventLoop.java
      13. private SelectorTuple openSelector() {
      14.         final Selector unwrappedSelector;
      15.         try {
      16.             //创建java原生selector
      17.             unwrappedSelector = provider.openSelector();
      18.         } catch (IOException e) {
      19.             throw new ChannelException("failed to open a new selector", e);
      20.         }
      21.  
      22.         //是否需要优化，默认需要DISABLE_KEYSET_OPTIMIZATION=false
      23.         if (DISABLE_KEYSET_OPTIMIZATION) {
      24.             return new SelectorTuple(unwrappedSelector);
      25.         }
      26.  
      27.         //尝试获取sun.nio.ch.SelectorImpl的class对象
      28.         Object maybeSelectorImplClass = AccessController.doPrivileged(new PrivilegedAction() {
      29.             @Override
      30.             public Object run() {
      31.                 try {
      32.                     return Class.forName(
      33.                             "sun.nio.ch.SelectorImpl",
      34.                             false,
      35.                             PlatformDependent.getSystemClassLoader());
      36.                 } catch (Throwable cause) {
      37.                     return cause;
      38.                 }
      39.             }
      40.         });
      41.  
      42.         //如果返回maybeSelectorImplClass不是一个class对象，或者maybeSelectorImplClass不是(unwrappedSelector.getClass()他的子类。
      43.         if (!(maybeSelectorImplClass instanceof Class) ||
      44.                 // ensure the current selector implementation is what we can instrument.
      45.                 !((Class) maybeSelectorImplClass).isAssignableFrom(unwrappedSelector.getClass())) {
      46.  
      47.             //如果是异常说明上面方法抛出异常
      48.             if (maybeSelectorImplClass instanceof Throwable) {
      49.                 Throwable t = (Throwable) maybeSelectorImplClass;
      50.                 logger.trace("failed to instrument a special java.util.Set into: {}", unwrappedSelector, t);
      51.             }
      52.             //创建一个SelectorTuple，内部unwrappedSelector并没有被优化
      53.             return new SelectorTuple(unwrappedSelector);
      54.         }
      55.  
      56.         //selector的class对象
      57.         final Class selectorImplClass = (Class) maybeSelectorImplClass;
      58.         //内部素组结构实现的set接口
      59.         final SelectedSelectionKeySet selectedKeySet = new SelectedSelectionKeySet();
      60.  
      61.         Object maybeException = AccessController.doPrivileged(new PrivilegedAction() {
      62.             @Override
      63.             public Object run() {
      64.                 try {
      65.                     //反射获取属性
      66.                     Field selectedKeysField = selectorImplClass.getDeclaredField("selectedKeys");
      67.                     Field publicSelectedKeysField = selectorImplClass.getDeclaredField("publicSelectedKeys");
      68.  
      69.                     //设置属性可访问
      70.                     Throwable cause = ReflectionUtil.trySetAccessible(selectedKeysField, true);
      71.                     if (cause != null) {
      72.                         return cause;
      73.                     }
      74.  
      75.                     //设置属性可访问
      76.                     cause = ReflectionUtil.trySetAccessible(publicSelectedKeysField, true);
      77.                     if (cause != null) {
      78.                         return cause;
      79.                     }
      80.  
      81.                     //把原来属性设置为selectedKeySet（它是素组实现)，原来是hashmap实现
      82.                     selectedKeysField.set(unwrappedSelector, selectedKeySet);
      83.                     publicSelectedKeysField.set(unwrappedSelector, selectedKeySet);
      84.                     return null;
      85.                 } catch (NoSuchFieldException e) {
      86.                     return e;
      87.                 } catch (IllegalAccessException e) {
      88.                     return e;
      89.                 }
      90.             }
      91.         });
      92.  
      93.         //如果返回结果是异常信息
      94.         if (maybeException instanceof Exception) {
      95.             selectedKeys = null;
      96.             Exception e = (Exception) maybeException;
      97.             logger.trace("failed to instrument a special java.util.Set into: {}", unwrappedSelector, e);
      98.             //返回SelectorTuple，内部unwrappedSelector并没有被优化
      99.             return new SelectorTuple(unwrappedSelector);
      100.         }
      101.  
      102.         //赋值
      103.         selectedKeys = selectedKeySet;
      104.         logger.trace("instrumented a special java.util.Set into: {}", unwrappedSelector);
      105.  
      106.         //unwrappedSelector内部的selectedKeys和publicSelectedKeys俩个数据结构已用素组优化
      107.         //SelectedSelectionKeySetSelector包装了unwrappedSelector和selectedKeySet
      108.         return new SelectorTuple(unwrappedSelector,
      109.                                  new SelectedSelectionKeySetSelector(unwrappedSelector, selectedKeySet));
      110. }
        111.  

        5.2、boss select监听
         

        1. //boss select监听NioEventLoop.run()
        2. //NioEventLoop.java
        3. protected void run() {
        4.     // loop，循环处理IO事件或者处理线程池中的task任务
        5.     for (;;) {
        6.         try {
        7.             // 判断接下来是是执行select还是直接处理IO事件和执行队列中的task
        8.             // hasTask判断当前线程的queue中是否还有待执行的任务
        9.             switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
        10.                 case SelectStrategy.CONTINUE:
        11.                     // NioEventLoop默认不会有这种状态
        12.                     continue;
        13.                 case SelectStrategy.SELECT:
        14.                     // 说明当前queue中没有task待执行
        15.                     select(wakenUp.getAndSet(false));
        16. 					// 唤醒epoll_wait
        17.                     if (wakenUp.get()) {
        18.                         selector.wakeup();
        19.                     }
        20.                     // fall through
        21.                 default:
        22.             }
        23.  
        24.             cancelledKeys = 0;
        25.             needsToSelectAgain = false;
        26.             // 这个比例是处理IO事件所需的时间和花费在处理task时间的比例
        27.             final int ioRatio = this.ioRatio;
        28.             if (ioRatio == 100) {
        29.                 // 如果比例是100，表示每次都处理完IO事件后，执行所有的task
        30.                 try {
        31.                     processSelectedKeys();
        32.                 } finally {
        33.                     // Ensure we always run tasks.
        34.                     // 保证能执行所有的task
        35.                     runAllTasks();
        36.                 }
        37.             } else {
        38.                 // 记录处理IO事件开始的时间
        39.                 final long ioStartTime = System.nanoTime();
        40.                 try {
        41.                     // 处理IO事件
        42.                     processSelectedKeys();
        43.                 } finally {
        44.                     // Ensure we always run tasks.
        45.                     // 当前时间减去处理IO事件开始的时间就是处理IO事件花费的时间
        46.                     final long ioTime = System.nanoTime() - ioStartTime;
        47.                     // 执行task的时间taskTime就是ioTime * (100 - ioRatio) / ioRatio
        48.                     // 如果taskTime时间到了还有未执行的task，runAllTasks也会返回
        49.                     runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
        50.                 }
        51.             }
        52.         } catch (Throwable t) {
        53.             handleLoopException(t);
        54.         }
        55.         // Always handle shutdown even if the loop processing threw an exception.
        56.         try {
        57.             // 如果已经shutdown则关闭所有资源
        58.             if (isShuttingDown()) {
        59.                 closeAll();
        60.                 if (confirmShutdown()) {
        61.                     return;
        62.                 }
        63.             }
        64.         } catch (Throwable t) {
        65.             handleLoopException(t);
        66.         }
        67.     }
        68. }
        69.  
        70. // io.netty.channel.DefaultSelectStrategy#calculateStrategy
        71. public int calculateStrategy(IntSupplier selectSupplier, boolean hasTasks) throws Exception {
        72.     // 如果还有task待执行则先执行selectNow，selectNow是立即返回的，不是阻塞等待
        73.     // 如果没有待执行的task则执行select，有可能是阻塞等待IO事件
        74.     return hasTasks ? selectSupplier.get() : SelectStrategy.SELECT;
        75. }
        76.  
        77. // io.netty.channel.nio.NioEventLoop#selectNowSupplier
        78. private final IntSupplier selectNowSupplier = new IntSupplier() {
        79.     @Override
        80.     public int get() throws Exception {
        81.         // epoll_wait的参数timeout可以指定超时时间，selectNow传入的参数是0，也就是不超时等待立即返回
        82.         return selectNow();
        83.     }
        84. };
        85.  
        86. //AbstractNioChannel.java
        87. protected void doRegister() throws Exception {
        88.         boolean selected = false;
        89.         for (;;) {
        90.             try {
        91.                 //循环监听selectionKey状态
        92.                 selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
        93.                 return;
        94.             } catch (CancelledKeyException e) {
        95.                 if (!selected) {
        96.                     // Force the Selector to select now as the "canceled" SelectionKey may still be
        97.                     // cached and not removed because no Select.select(..) operation was called yet.
        98.                     eventLoop().selectNow();
        99.                     selected = true;
        100.                 } else {
        101.                     // We forced a select operation on the selector before but the SelectionKey is still cached
        102.                     // for whatever reason. JDK bug ?
        103.                     throw e;
        104.                 }
        105.             }
        106.         }
        107. }
        108.  
        109. //NioEventLoop.java
        110. private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
        111.         final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
        112.         if (!k.isValid()) {
        113.             final EventLoop eventLoop;
        114.             try {
        115.                 eventLoop = ch.eventLoop();
        116.             } catch (Throwable ignored) {
        117.                 // If the channel implementation throws an exception because there is no event loop, we ignore this
        118.                 // because we are only trying to determine if ch is registered to this event loop and thus has authority
        119.                 // to close ch.
        120.                 return;
        121.             }
        122.             // Only close ch if ch is still registered to this EventLoop. ch could have deregistered from the event loop
        123.             // and thus the SelectionKey could be cancelled as part of the deregistration process, but the channel is
        124.             // still healthy and should not be closed.
        125.             // See https://github.com/netty/netty/issues/5125
        126.             if (eventLoop == this) {
        127.                 // close the channel if the key is not valid anymore
        128.                 unsafe.close(unsafe.voidPromise());
        129.             }
        130.             return;
        131.         }
        132.  
        133.         try {
        134.             int readyOps = k.readyOps();
        135.             // We first need to call finishConnect() before try to trigger a read(...) or write(...) as otherwise
        136.             // the NIO JDK channel implementation may throw a NotYetConnectedException.
        137.             if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
        138.                 // remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
        139.                 // See https://github.com/netty/netty/issues/924
        140.                 int ops = k.interestOps();
        141.                 ops &= ~SelectionKey.OP_CONNECT;
        142.                 k.interestOps(ops);
        143.  
        144.                 unsafe.finishConnect();
        145.             }
        146.  
        147.             // Process OP_WRITE first as we may be able to write some queued buffers and so free memory.
        148.             if ((readyOps & SelectionKey.OP_WRITE) != 0) {
        149.                 // Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to write
        150.                 ch.unsafe().forceFlush();
        151.             }
        152.  
        153.             // Also check for readOps of 0 to workaround possible JDK bug which may otherwise lead
        154.             // to a spin loop
        155.             if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
        156.                 //这点事重点 io就绪调read()
        157.                 unsafe.read();
        158.             }
        159.         } catch (CancelledKeyException ignored) {
        160.             unsafe.close(unsafe.voidPromise());
        161.         }
        162. }
        163.  
        164. //AbstractNioByteChannel.java
        165. @Override
        166. public final void read() {
        167.     final ChannelConfig config = config();
        168.     final ChannelPipeline pipeline = pipeline();
        169.     // 用来处理内存的分配:池化或者非池化 UnpooledByteBufAllocator
        170.     final ByteBufAllocator allocator = config.getAllocator();
        171.     // 用来计算此次读循环应该分配多少内存 AdaptiveRecvByteBufAllocator 自适应计算缓冲分配
        172.     final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();
        173.     allocHandle.reset(config);// 重置为0
        174.  
        175.     ByteBuf byteBuf = null;
        176.     boolean close = false;
        177.     try {
        178.         do {
        179.             byteBuf = allocHandle.allocate(allocator);//这里会根据上一次的读取情况进行自适应的调整大小
        180.             allocHandle.lastBytesRead(doReadBytes(byteBuf));
        181.             if (allocHandle.lastBytesRead() <= 0) {// 如果上一次读到的字节数小于等于0，清理引用和跳出循环
        182.                 // nothing was read. release the buffer.
        183.                 byteBuf.release();// 引用 -1
        184.                 byteBuf = null;
        185.                 close = allocHandle.lastBytesRead() < 0;// 如果远程已经关闭连接
        186.                 if (close) {
        187.                     // There is nothing left to read as we received an EOF.
        188.                     readPending = false;
        189.                 }
        190.                 break;
        191.             }
        192.  
        193.             allocHandle.incMessagesRead(1);//  totalMessages += amt;
        194.             readPending = false;
        195.             //
        196.             pipeline.fireChannelRead(byteBuf);
        197.             byteBuf = null;
        198.         } while (allocHandle.continueReading());
        199.  
        200.         allocHandle.readComplete();
        201.         pipeline.fireChannelReadComplete();
        202.  
        203.         if (close) {
        204.             closeOnRead(pipeline);
        205.         }
        206.     } catch (Throwable t) {
        207.         handleReadException(pipeline, byteBuf, t, close, allocHandle);
        208.     } finally {
        209.         if (!readPending && !config.isAutoRead()) {
        210.             removeReadOp();
        211.         }
        212. }
         

         

        5.3、boss线程池总结
         

        boss线程池创建打开select监听, NioEventLoop继承多线程ScheduledExecutorService,在run方法实现了获取nio selectKey状态的方法processSelectedKeys.后续调read()封装ByteBuf在保存在堆外内存.
         



         六、worker线程池调用链
         


         6.1、worker线程初始化
         


         将线程池注册到Channel通道上面,然后通过Channel通道获取线程池执行, TODO
         

        1. //ServerBootstrapAcceptor.java
        2. //childGroup 线程池监听通道
        3. public void channelRead(ChannelHandlerContext ctx, Object msg) {
        4.             //channel对象构造
        5.             final Channel child = (Channel) msg;
        6.  
        7.             child.pipeline().addLast(childHandler);
        8.  
        9.             setChannelOptions(child, childOptions, logger);
        10.             setAttributes(child, childAttrs);
        11.  
        12.             try {
        13.                 //将child通道对象,分配给childGroup处理并监听结果
        14.                 childGroup.register(child).addListener(new ChannelFutureListener() {
        15.                     @Override
        16.                     public void operationComplete(ChannelFuture future) throws Exception {
        17.                         if (!future.isSuccess()) {
        18.                             forceClose(child, future.cause());
        19.                         }
        20.                     }
        21.                 });
        22.             } catch (Throwable t) {
        23.                 forceClose(child, t);
        24.             }
        25. }
        26.  
        27.  
        28.  
        29. //Channel对象获取worke线程池
        30. //AbstractChannelHandlerContext.java
        31. public EventExecutor executor() {
        32.         if (executor == null) {
        33.             return channel().eventLoop();
        34.         } else {
        35.             return executor;
        36.         }
        37. }
         


          
        
                
                    
                

                
      111. 

                    相关阅读:
        
                    
Golang一日一库之bcrypt                            
        
【 与百度搜索相同的bootstrap4与5自动补全功能（autocomplete）】                            
        
【论文阅读】Q8BERT: Quantized 8Bit BERT                            
        
【扩散模型】5、Diffusion models beat GAN | 使用类别引导图像生成                            
        
Android frameworks学习                            
        
Linux 日志系统、auditd用户审计、kdump故障定位                            
        
2022英特尔® FPGA中国技术周                            
        
U盘里文件损坏无法打开怎么恢复？                            
        
ITextRenderer将PDF转换为HTML详细教程                            
        
LeetCode 452. 用最少数量的箭引爆气球                            
        
                            
                
        112. 
                
      112. 
                    原文地址：https://blog.csdn.net/god8816/article/details/127774067
                
        113.