淺談Netty中的NioEventLoop

時間  2019-11-06
標籤 淺談 netty nioeventloop 欄目 Netty 简体版
原文   原文鏈接

NioEventLoop建立

步驟

  • new NioEventLoopGroup() 線程組，默認2*cpu
    • new ThreadPerTaskExecutor() 線程建立器
      • 每次執行任務都會建立一個線程實體
    • for(){newChild()} 構造NioEventLoop
      • 保存線程執行器ThreadPerTaskExecutor
      • 建立一個MpscQueue
      • 建立一個selector
    • chooserFactory.newChooser() 線程選擇器
      • isPowerOfTwo() 是否等於2的冪次
        • PowerOfTwoEventExecutorChooser
          • index++&(length-1)
        • GenericEventExecutorChooser
          • abs(index++%length)

分析

  • new NioEventLoopGroup() 線程組，默認2*cpu
protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {
        super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
    }
    static {
        // 可獲取線程數*2，目前大部分都是超線程，因此NettyRuntime.availableProcessors()可能獲取的核數的2倍
        DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt(
                "io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));
        ...
    }
複製代碼
  • new ThreadPerTaskExecutor() 線程建立器
public final class ThreadPerTaskExecutor implements Executor {

    private final ThreadFactory threadFactory;

    public ThreadPerTaskExecutor(ThreadFactory threadFactory) {
        if (threadFactory == null) {
            throw new NullPointerException("threadFactory");
        }
        this.threadFactory = threadFactory;
    }

    // 它會給每個執行任務建立一個新的netty包裝過的FastThreadLocalThread去運行
    // 使用了命令設計模式
    @Override
    public void execute(Runnable command) {
        threadFactory.newThread(command).start();
    }
}
複製代碼
  • for(){newChild()} 構造NioEventLoop
children = new EventExecutor[nThreads];
    for (int i = 0; i < nThreads; i ++) {
        boolean success = false;
        try {
            children[i] = newChild(executor, args);
            success = true;
        } catch (Exception e) {
           ...
        } finally {
            ...
        }
    }
複製代碼
  • 跟着newChild(..)到NioEventLoop的構造方法，依次進入它的父類構造方法
NioEventLoop(..) {
        super(parent, executor, false, DEFAULT_MAX_PENDING_TASKS, rejectedExecutionHandler);
        ...
        provider = selectorProvider;
        // 建立一個selectorTuple，netty對底層的selectionKeySet進行了優化，由於本來的selectionKeySet底層是Map結構，add的時候時間複雜度是O(n)的，netty中直接使用數組的結構來代替它add編程O(1)
        final SelectorTuple selectorTuple = openSelector();
        // 被包裝過的selector，它提供了一些基於新的keyset的一些遍歷的方法
        selector = selectorTuple.selector; 
        // 沒有被包裝過的selector 也就nio的selector
        unwrappedSelector = selectorTuple.unwrappedSelector;
        ...
    }
    protected SingleThreadEventExecutor(..) {
        // 省略代碼
        ...
        // 保存線程執行器ThreadPerTaskExecutor
        this.executor = ObjectUtil.checkNotNull(executor, "executor");
        ...
    }
    // 它的父類 SingleThreadEventLoop 會初始化個tailTasks
    protected SingleThreadEventLoop(...) {
        super(parent, executor, addTaskWakesUp, maxPendingTasks, rejectedExecutionHandler)
        // 初始化taskQueue
        tailTasks = newTaskQueue(maxPendingTasks);
    }
    // 建立一個MpscQueue
    protected Queue<Runnable> newTaskQueue(int maxPendingTasks) {
        // This event loop never calls takeTask()
        return maxPendingTasks == Integer.MAX_VALUE ? PlatformDependent.<Runnable>newMpscQueue()
                                                    : PlatformDependent.<Runnable>newMpscQueue(maxPendingTasks);
    }
複製代碼
  • chooserFactory.newChooser() 線程選擇器
public final class DefaultEventExecutorChooserFactory implements EventExecutorChooserFactory {

    public static final DefaultEventExecutorChooserFactory INSTANCE = new DefaultEventExecutorChooserFactory();

    private DefaultEventExecutorChooserFactory() { }

    @SuppressWarnings("unchecked")
    @Override
    public EventExecutorChooser newChooser(EventExecutor[] executors) {

        /**
        * 這塊優化的問題能夠簡化爲：
        *   對於能被2整除的數n，爲何x%n = x&(n-1)
        *   舉個例子 int n = 4,8 的二進制存儲
        *    0000 0000 0000 0100  -1後   0000 0000 0000 0011
        *    0000 0000 0000 1000  -1後   0000 0000 0000 0111
        *    根據例子能夠發現知足了一點就是模一個數n以後最大也就爲n-1，當前的數值x我也不用關注大於個人前邊的值爲1仍是0 
        *  只關注按位與上個人後面的結果便可。    
        */
        if (isPowerOfTwo(executors.length)) {
            return new PowerOfTwoEventExecutorChooser(executors);
        } else {
            return new GenericEventExecutorChooser(executors);
        }
    }
    // 判斷是不是2的冪次
    private static boolean isPowerOfTwo(int val) {
        return (val & -val) == val;
    }
    private static final class PowerOfTwoEventExecutorChooser implements EventExecutorChooser {
        private final AtomicInteger idx = new AtomicInteger();
        private final EventExecutor[] executors;

        PowerOfTwoEventExecutorChooser(EventExecutor[] executors) {
            this.executors = executors;
        }
        @Override
        public EventExecutor next() {
            return executors[idx.getAndIncrement() & executors.length - 1];
        }
    }
    private static final class GenericEventExecutorChooser implements EventExecutorChooser {
        private final AtomicInteger idx = new AtomicInteger();
        private final EventExecutor[] executors;

        GenericEventExecutorChooser(EventExecutor[] executors) {
            this.executors = executors;
        }
        @Override
        public EventExecutor next() {
            return executors[Math.abs(idx.getAndIncrement() % executors.length)];
        }
    }
}
複製代碼

NioEventLoop啓動觸發器

步驟

  • 服務端啓動綁定端口
    • bind() -> execute(task) 入口
    • startThread() -> doStartThread() 建立線程
      • ThreadPerTaskExecutor.execute()
        • thread = Thread.currentThread()
        • NioEventLoop.run() 啓動
  • 新連結接入經過chooser選擇個EventLoop

分析

  • bind() -> execute(task) 入口
private static void doBind0(..) {
        // 省略代碼...
        channel.eventLoop().execute(task);
    }
複製代碼
  • startThread() -> doStartThread() 建立線程
public void execute(Runnable task) {
        if (task == null) {
            throw new NullPointerException("task");
        }

        boolean inEventLoop = inEventLoop();
        addTask(task);
        if (!inEventLoop) {
            startThread();
            ...
        }

        if (!addTaskWakesUp && wakesUpForTask(task)) {
            wakeup(inEventLoop);
        }
    }
    private void doStartThread() {
        assert thread == null;
        executor.execute(new Runnable() {
            @Override
            public void run() {
                // thread = Thread.currentThread()
                thread = Thread.currentThread();
                if (interrupted) {
                    thread.interrupt();
                }
                boolean success = false;
                updateLastExecutionTime();
                try {
                    // 這個時候NioEventLoop的run方法才執行
                    SingleThreadEventExecutor.this.run();
                    success = true;
                } catch (Throwable t) {
                    logger.warn("Unexpected exception from an event executor: ", t);
                } finally {
                    ...
                }
            }
        });
    }
複製代碼

NioEventLoop執行邏輯

步驟

  • NioEventLoop.run()
    • run() -> for(;;)
      • select() 檢查是否有io事件
        • deadline以及任務穿插邏輯處理
        • 阻塞式select
        • 避免jdk空輪詢bug
      • processSelectedKeys() 處理io事件
        • selected keySet優化 其實是數組 add事件複雜度爲O(1)
        • proccessSelectedKeysOptimized()
      • runAllTasks() 處理異步任務隊列
        • task的添加
          • 普通taskQueue
          • scheduleQueue建立的添加
        • 任務的聚合
        • 任務的執行

分析

  • run() -> for(;;)
@Override
    protected void run() {
        for (;;) {
            try {
                try {
                    switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
                    case SelectStrategy.CONTINUE:
                        continue;

                    case SelectStrategy.BUSY_WAIT:
                        // ...

                    case SelectStrategy.SELECT:
                        select(wakenUp.getAndSet(false));
                        // 省略代碼...
                        if (wakenUp.get()) {
                            selector.wakeup();
                        }
                        // fall through
                    default:
                    }
                } catch (IOException e) {
                    ...
                    continue;
                }
                cancelledKeys = 0;
                needsToSelectAgain = false;
                final int ioRatio = this.ioRatio;

                if (ioRatio == 100) {
                    try {
                        // 處理select出來的事件
                        processSelectedKeys();
                    } finally {
                        ...
                        runAllTasks();
                    }
                } else {   // 默認是50
                    final long ioStartTime = System.nanoTime();
                    try {
                        // 處理select出來的事件
                        processSelectedKeys();
                    } finally {
                        ...
                        final long ioTime = System.nanoTime() - ioStartTime;
                        runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
                    }
                }
            } catch (Throwable t) {
                handleLoopException(t);
            }
            // ...省略代碼
        }
    }
複製代碼
  • select() 檢查是否有io事件
private void select(boolean oldWakenUp) throws IOException {
        Selector selector = this.selector;
        try {
            int selectCnt = 0;
            long currentTimeNanos = System.nanoTime();
            long selectDeadLineNanos = currentTimeNanos + delayNanos(currentTimeNanos);

            for (;;) {

                // 若是到了第一個定時任務的截止事件尚未select，那麼就進行一次非阻塞的selectNow，這樣會執行堆積的task
                long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L;
                if (timeoutMillis <= 0) {
                    if (selectCnt == 0) {
                        selector.selectNow();
                        selectCnt = 1;
                    }
                    break;
                }

                // 若是咱們的任務隊列中有任務，也進行一次selectNow處理任務
                if (hasTasks() && wakenUp.compareAndSet(false, true)) {
                    selector.selectNow();
                    selectCnt = 1;
                    break;
                }

                // 若是上面狀況都不知足，就進行一次阻塞的select
                int selectedKeys = selector.select(timeoutMillis);
                selectCnt ++;

                // 下面這段代碼是解決jdk空輪詢的bug
                long time = System.nanoTime();
                // 執行到這的時間time - 在selector.select(timeoutMillis)方法以前獲取的時間 currentTimeNanos 都大於等於 輪詢的時間timeoutMillis 就說明輪詢是空輪詢
                if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {
                    // timeoutMillis elapsed without anything selected.
                    selectCnt = 1;
                } else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 &&
                        selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) { // 空輪詢次數大於從新構建selector閾值512
                    // 避免jdk空輪詢bug，解決方式就是從新構建selector
                    selector = selectRebuildSelector(selectCnt);
                    selectCnt = 1;  // 輪詢次數歸1
                    break;
                }
                currentTimeNanos = time;
            }
        } catch (CancelledKeyException e) {
            //省略代碼...
        }
    }
複製代碼
  • processSelectedKeys() 處理io事件
  • selected keySet優化 其實是數組 add事件複雜度爲O(1) 在上面newChild有提到過
// 優化本來的set，底層直接用數組存儲。由於這裏不會輪到重複的key
final class SelectedSelectionKeySet extends AbstractSet<SelectionKey> {

    SelectionKey[] keys;
    int size;

    SelectedSelectionKeySet() {
        keys = new SelectionKey[1024];
    }

    @Override
    public boolean add(SelectionKey o) {
        if (o == null) {
            return false;
        }

        keys[size++] = o;
        if (size == keys.length) {
            increaseCapacity();
        }

        return true;
    }

    @Override
    public boolean remove(Object o) {
        return false;
    }

    @Override
    public boolean contains(Object o) {
        return false;
    }

    @Override
    public int size() {
        return size;
    }

    @Override
    public Iterator<SelectionKey> iterator() {
        return new Iterator<SelectionKey>() {
            private int idx;

            @Override
            public boolean hasNext() {
                return idx < size;
            }

            @Override
            public SelectionKey next() {
                if (!hasNext()) {
                    throw new NoSuchElementException();
                }
                return keys[idx++];
            }

            @Override
            public void remove() {
                throw new UnsupportedOperationException();
            }
        };
    }

    void reset() {
        reset(0);
    }

    void reset(int start) {
        Arrays.fill(keys, start, size, null);
        size = 0;
    }

    private void increaseCapacity() {
        SelectionKey[] newKeys = new SelectionKey[keys.length << 1];
        System.arraycopy(keys, 0, newKeys, 0, size);
        keys = newKeys;
    }
}
複製代碼

// 經過反射的方式對原先的selectedKeysField進行從新賦值java

if (!(maybeSelectorImplClass instanceof Class) ||
            // ensure the current selector implementation is what we can instrument.
            !((Class<?>) maybeSelectorImplClass).isAssignableFrom(unwrappedSelector.getClass())) {
            if (maybeSelectorImplClass instanceof Throwable) {
                Throwable t = (Throwable) maybeSelectorImplClass;
                logger.trace("failed to instrument a special java.util.Set into: {}", unwrappedSelector, t);
            }
            return new SelectorTuple(unwrappedSelector);
        }

        final Class<?> selectorImplClass = (Class<?>) maybeSelectorImplClass;
        final SelectedSelectionKeySet selectedKeySet = new SelectedSelectionKeySet();

        Object maybeException = AccessController.doPrivileged(new PrivilegedAction<Object>() {
            @Override
            public Object run() {
                try {
                    Field selectedKeysField = selectorImplClass.getDeclaredField("selectedKeys");
                    Field publicSelectedKeysField = selectorImplClass.getDeclaredField("publicSelectedKeys");

                    if (PlatformDependent.javaVersion() >= 9 && PlatformDependent.hasUnsafe()) {
                        long selectedKeysFieldOffset = PlatformDependent.objectFieldOffset(selectedKeysField);
                        long publicSelectedKeysFieldOffset =
                                PlatformDependent.objectFieldOffset(publicSelectedKeysField);

                        if (selectedKeysFieldOffset != -1 && publicSelectedKeysFieldOffset != -1) {
                            PlatformDependent.putObject(
                                    unwrappedSelector, selectedKeysFieldOffset, selectedKeySet);
                            PlatformDependent.putObject(
                                    unwrappedSelector, publicSelectedKeysFieldOffset, selectedKeySet);
                            return null;
                        }
                        // We could not retrieve the offset, lets try reflection as last-resort.
                    }

                    Throwable cause = ReflectionUtil.trySetAccessible(selectedKeysField, true);
                    if (cause != null) {
                        return cause;
                    }
                    cause = ReflectionUtil.trySetAccessible(publicSelectedKeysField, true);
                    if (cause != null) {
                        return cause;
                    }

                    selectedKeysField.set(unwrappedSelector, selectedKeySet);
                    publicSelectedKeysField.set(unwrappedSelector, selectedKeySet);
                    return null;
                } catch (NoSuchFieldException e) {
                    return e;
                } catch (IllegalAccessException e) {
                    return e;
                }
            }
        });
複製代碼
  • 處理事件proccessSelectedKeysOptimized()
int readyOps = k.readyOps();
      // 處理鏈接事件
      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();
      }
複製代碼
  • runAllTasks() 處理異步任務隊列
  • 普通任務隊列taskQueue的新增
public void execute(Runnable task) {
        ...
        addTask(task);
        ...
    }
   protected void addTask(Runnable task) {
        if (task == null) {
            throw new NullPointerException("task");
        }
        if (!offerTask(task)) {
            reject(task);
        }
    }

    final boolean offerTask(Runnable task) {
        if (isShutdown()) {
            reject();
        }
        return taskQueue.offer(task);
    }
複製代碼
  • 任務的聚合與執行 在NioEventLoop的run方法中，runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
protected boolean runAllTasks(long timeoutNanos) {
        // ...省略代碼
        // 定時任務聚合到普通任務隊列
        fetchFromScheduledTaskQueue();
        // 運行全部普通任務隊列的任務
        afterRunningAllTasks();
        return true;
    }
    // 把大於當前執行事件的定時任務都加入到普通隊列中
    private boolean fetchFromScheduledTaskQueue() {
        long nanoTime = AbstractScheduledEventExecutor.nanoTime();
        Runnable scheduledTask  = pollScheduledTask(nanoTime);
        while (scheduledTask != null) {
            if (!taskQueue.offer(scheduledTask)) {
                // No space left in the task queue add it back to the scheduledTaskQueue so we pick it up again.
                scheduledTaskQueue().add((ScheduledFutureTask<?>) scheduledTask);
                return false;
            }
            scheduledTask  = pollScheduledTask(nanoTime);
        }
        return true;
    }
    // 全部任務
    protected final boolean runAllTasksFrom(Queue<Runnable> taskQueue) {
        Runnable task = pollTaskFrom(taskQueue);
        if (task == null) {
            return false;
        }
        for (;;) {
            // 執行task的run方法
            safeExecute(task);
            task = pollTaskFrom(taskQueue);
            if (task == null) {
                return true;
            }
        }
    }
複製代碼
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程