netty 客户端接入源码分析

冷不防 2023-06-13 04:44 29阅读 0赞

## netty 客户端接入源码分析 ##

netty 线程选择为`NioEventLoop`其实际运行的线程函数为

@Override
    protected void run() { 
        for (;;) { 
            try { 
                try { 
                    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
    
                    case SelectStrategy.SELECT:
                        select(wakenUp.getAndSet(false));
    
                        // 'wakenUp.compareAndSet(false, true)' is always evaluated
                        // before calling 'selector.wakeup()' to reduce the wake-up
                        // overhead. (Selector.wakeup() is an expensive operation.)
                        //
                        // However, there is a race condition in this approach.
                        // The race condition is triggered when 'wakenUp' is set to
                        // true too early.
                        //
                        // 'wakenUp' is set to true too early if:
                        // 1) Selector is waken up between 'wakenUp.set(false)' and
                        // 'selector.select(...)'. (BAD)
                        // 2) Selector is waken up between 'selector.select(...)' and
                        // 'if (wakenUp.get()) { ... }'. (OK)
                        //
                        // In the first case, 'wakenUp' is set to true and the
                        // following 'selector.select(...)' will wake up immediately.
                        // Until 'wakenUp' is set to false again in the next round,
                        // 'wakenUp.compareAndSet(false, true)' will fail, and therefore
                        // any attempt to wake up the Selector will fail, too, causing
                        // the following 'selector.select(...)' call to block
                        // unnecessarily.
                        //
                        // To fix this problem, we wake up the selector again if wakenUp
                        // is true immediately after selector.select(...).
                        // It is inefficient in that it wakes up the selector for both
                        // the first case (BAD - wake-up required) and the second case
                        // (OK - no wake-up required).
    
                        if (wakenUp.get()) { 
                            selector.wakeup();
                        }
                        // fall through
                    default:
                    }
                } catch (IOException e) { 
                    // If we receive an IOException here its because the Selector is messed up. Let's rebuild
                    // the selector and retry. https://github.com/netty/netty/issues/8566
                    rebuildSelector0();
                    handleLoopException(e);
                    continue;
                }
    
                cancelledKeys = 0;
                needsToSelectAgain = false;
                final int ioRatio = this.ioRatio;
                if (ioRatio == 100) { 
                    try { 
                        processSelectedKeys();
                    } finally { 
                        // Ensure we always run tasks.
                        runAllTasks();
                    }
                } else { 
                    final long ioStartTime = System.nanoTime();
                    try { 
                        processSelectedKeys();
                    } finally { 
                        // Ensure we always run tasks.
                        final long ioTime = System.nanoTime() - ioStartTime;
                        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);
            }
        }
    }

### 执行策略 ###

selectStrategy只有一个实现

final class DefaultSelectStrategy implements SelectStrategy { 
        static final SelectStrategy INSTANCE = new DefaultSelectStrategy();
    
        private DefaultSelectStrategy() {  }
    
        @Override
        public int calculateStrategy(IntSupplier selectSupplier, boolean hasTasks) throws Exception { 
            return hasTasks ? selectSupplier.get() : SelectStrategy.SELECT;
        }
    }

### io 时间占比 ###

若存在 task，则返回selectNow 的值，否则的话返回 SELECT

ioRatio为**循环中io 占用的百分比**

若为 100则使用如下策略执行

> 执行 io，执行一个任务

否则按百分比执行

> 执行任务的时间所占的比例为 ((100-ioRatio)/ioRation)\*io 时间

这也正是`runAllTasks`和带时间的`runAllTasks`的区别，一个仅执行一个任务，一个则执行一段时间

### 执行 io ###

private void processSelectedKeys() { 
        if (selectedKeys != null) { 
            processSelectedKeysOptimized();
        } else { 
            processSelectedKeysPlain(selector.selectedKeys());
        }
    }
    private void processSelectedKeysPlain(Set<SelectionKey> selectedKeys) { 
      // check if the set is empty and if so just return to not create garbage by
      // creating a new Iterator every time even if there is nothing to process.
      // See https://github.com/netty/netty/issues/597
      if (selectedKeys.isEmpty()) { 
        return;
      }
    
      Iterator<SelectionKey> i = selectedKeys.iterator();
      for (;;) { 
        final SelectionKey k = i.next();
        final Object a = k.attachment();
        i.remove();
    
        if (a instanceof AbstractNioChannel) { 
          processSelectedKey(k, (AbstractNioChannel) a);
        } else { 
          @SuppressWarnings("unchecked")
          NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
          processSelectedKey(k, task);
        }
    
        if (!i.hasNext()) { 
          break;
        }
    
        if (needsToSelectAgain) { 
          selectAgain();
          selectedKeys = selector.selectedKeys();
    
          // Create the iterator again to avoid ConcurrentModificationException
          if (selectedKeys.isEmpty()) { 
            break;
          } else { 
            i = selectedKeys.iterator();
          }
        }
      }
    }

因为 selectKeys 非空，因此执行`processSelectedKeysPlain`，又因为是 NioServerSocketChannel，因此执行processSelectedKey(k, (AbstractNioChannel) a);

private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) { 
        final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
        if (!k.isValid()) { 
            final EventLoop eventLoop;
            try { 
                eventLoop = ch.eventLoop();
            } catch (Throwable ignored) { 
                // If the channel implementation throws an exception because there is no event loop, we ignore this
                // because we are only trying to determine if ch is registered to this event loop and thus has authority
                // to close ch.
                return;
            }
            // Only close ch if ch is still registered to this EventLoop. ch could have deregistered from the event loop
            // and thus the SelectionKey could be cancelled as part of the deregistration process, but the channel is
            // still healthy and should not be closed.
            // See https://github.com/netty/netty/issues/5125
            if (eventLoop != this || eventLoop == null) { 
                return;
            }
            // close the channel if the key is not valid anymore
            unsafe.close(unsafe.voidPromise());
            return;
        }
    
        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();
            }
    
            // Process OP_WRITE first as we may be able to write some queued buffers and so free memory.
            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();
            }
    
            // Also check for readOps of 0 to workaround possible JDK bug which may otherwise lead
            // to a spin loop
            if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) { 
                unsafe.read();
            }
        } catch (CancelledKeyException ignored) { 
            unsafe.close(unsafe.voidPromise());
        }
    }

对于 NioServerSocketChannel 来说，其关心的是`OP_ACCEPT`因此会触发 unsafe.read 操作，在服务端分析中，已经分析过其 unsafe 的 read 函数在 NioMessageUnsafe 中

public void read() { 
        assert eventLoop().inEventLoop();
        final ChannelConfig config = config();
        final ChannelPipeline pipeline = pipeline();
        final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
        allocHandle.reset(config);
    
        boolean closed = false;
        Throwable exception = null;
        try { 
            try { 
                do { 
                    int localRead = doReadMessages(readBuf);
                    if (localRead == 0) { 
                        break;
                    }
                    if (localRead < 0) { 
                        closed = true;
                        break;
                    }
    
                    allocHandle.incMessagesRead(localRead);
                } while (allocHandle.continueReading());
            } catch (Throwable t) { 
                exception = t;
            }
    
            int size = readBuf.size();
            for (int i = 0; i < size; i ++) { 
                readPending = false;
                pipeline.fireChannelRead(readBuf.get(i));
            }
            readBuf.clear();
            allocHandle.readComplete();
            pipeline.fireChannelReadComplete();
    
            if (exception != null) { 
                closed = closeOnReadError(exception);
    
                pipeline.fireExceptionCaught(exception);
            }
    
            if (closed) { 
                inputShutdown = true;
                if (isOpen()) { 
                    close(voidPromise());
                }
            }
        } finally { 
            // Check if there is a readPending which was not processed yet.
            // This could be for two reasons:
            // * The user called Channel.read() or ChannelHandlerContext.read() in channelRead(...) method
            // * The user called Channel.read() or ChannelHandlerContext.read() in channelReadComplete(...) method
            //
            // See https://github.com/netty/netty/issues/2254
            if (!readPending && !config.isAutoRead()) { 
                removeReadOp();
            }
        }
    }

分析doReadMessages函数，发现其是创建 SocketChannel

protected int doReadMessages(List<Object> buf) throws Exception { 
        SocketChannel ch = SocketUtils.accept(javaChannel());
    
        try { 
            if (ch != null) { 
                buf.add(new NioSocketChannel(this, ch));
                return 1;
            }
        } catch (Throwable t) { 
            logger.warn("Failed to create a new channel from an accepted socket.", t);
    
            try { 
                ch.close();
            } catch (Throwable t2) { 
                logger.warn("Failed to close a socket.", t2);
            }
        }
    
        return 0;
    }

随后将用每一个 channel 触发channelRead。

for (int i = 0; i < size; i ++) { 
        readPending = false;
        pipeline.fireChannelRead(readBuf.get(i));
    }

因此，channel 将传入到`ServerBootstrapAcceptor`

public void channelRead(ChannelHandlerContext ctx, Object msg) { 
        final Channel child = (Channel) msg;
    
        child.pipeline().addLast(childHandler);
    
        setChannelOptions(child, childOptions, logger);
    
        for (Entry<AttributeKey<?>, Object> e: childAttrs) { 
            child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());
        }
    
        try { 
            childGroup.register(child).addListener(new ChannelFutureListener() { 
                @Override
                public void operationComplete(ChannelFuture future) throws Exception { 
                    if (!future.isSuccess()) { 
                        forceClose(child, future.cause());
                    }
                }
            });
        } catch (Throwable t) { 
            forceClose(child, t);
        }
      
    }

主要包含了四个步骤

1.  将 childHandler 传入 channel 的 Pipeline 中
2.  设置 channel 的参数
3.  将 childAttr 传入 channel
4.  将 channel 注册到 childGroup 中

childGroup 的 register 实际上最后调用到了 AbstractNioChannel 的 doRegister 。（在服务端分析中有详细的路径追踪），随后将触发 active 事件，并最终调用到AbstractNioChannel 的doBeginRead 操作

@Override
    protected void doBeginRead() throws Exception { 
        // Channel.read() or ChannelHandlerContext.read() was called
        final SelectionKey selectionKey = this.selectionKey;
        if (!selectionKey.isValid()) { 
            return;
        }
    
        readPending = true;
    
        final int interestOps = selectionKey.interestOps();
        if ((interestOps & readInterestOp) == 0) { 
            selectionKey.interestOps(interestOps | readInterestOp);
        }
    }

对于NioSocketChannel来说，readInterestOps 为OP\_ACCEPT

protected AbstractNioByteChannel(Channel parent, SelectableChannel ch) { 
        super(parent, ch, SelectionKey.OP_READ);
    }