1. 程式人生 > >Netty NioEventLoop 建立過程原始碼分析

Netty NioEventLoop 建立過程原始碼分析

原文:https://wangwei.one/posts/netty-nioeventloop-analyse-for-create.html

前面 ,我們分析了Netty中的Channel元件,本篇我們來介紹一下與Channel關聯的另一個核心的元件 —— EventLoop

Netty版本:4.1.30

概述

EventLoop定義了Netty的核心抽象,用於處理網路連線生命週期中所有發生的事件。

我們先來從一個比較高的視角來了解一下Channels、Thread、EventLoops、EventLoopGroups之間的關係。

Netty-EventLoop-Channel

上圖是表示了擁有4個EventLoop的EventLoopGroup處理IO的流程圖。它們之間的關係如下:

  • 一個 EventLoopGroup包含一個或多個EventLoop
  • 一個 EventLoop在它的生命週期內只和一個Thread繫結
  • 所有由EventLoop處理的I/O事件都將在它專有的Thread上被處理
  • 一個Channel在它的生命週期內只註冊於一個EventLoop
  • 一個EventLoop可能會被分配給一個或多個Channel

EventLoop 原理

下圖是Netty EventLoop相關類的UML圖。從中我們可以看到EventLoop相關的類都是實現了 java.util.concurrent 包中的 ExecutorService 介面。我們可以直接將任務(Runable 或 Callable) 提交給EventLoop去立即執行或定時執行。

Netty EventLoop UML

例如,使用EventLoop去執行定時任務,樣例程式碼:

public static void scheduleViaEventLoop() {
    Channel ch = new NioSocketChannel();
    ScheduledFuture<?> future = ch.eventLoop().schedule(
            () -> System.out.println("60 seconds later"), 60, TimeUnit.SECONDS);
}

Thread 管理

Netty執行緒模型的高效能主要取決於當前所執行執行緒的身份的確定。一個執行緒提交到EventLoop執行的流程如下:

  • 將Task任務提交給EventLoop執行
  • 在Task傳遞到execute方法之後,檢查當前要執行的Task的執行緒是否是分配給EventLoop的那個執行緒
  • 如果是,則該執行緒會立即執行
  • 如果不是,則將執行緒放入任務佇列中,等待下一次執行

其中,Netty中的每一個EventLoop都有它自己的任務佇列,並且和其他的EventLoop的任務佇列獨立開來。

Nettu EventLoop Thread management

Thread 分配

服務於Channel的I/O和事件的EventLoop包含在EventLoopGroup中。根據不同的傳輸實現,EventLoop的建立和分配方式也不同。

NIO傳輸

Netty EventLoop Thread allocation NIO

在NIO傳輸方式中,使用盡可能少的EventLoop就可以服務多個Channel。如圖所示,EventLoopGroup採用順序迴圈的方式負責為每一個新建立的Channel分配EventLoop,每一個EventLoop會被分配給多個Channels。

一旦一個Channel被分配給了一個EventLoop,則這個Channel的生命週期內,只會繫結這個EventLoop。這就讓我們在ChannelHandler的實現省去了對執行緒安全和同步問題的擔心。

OIO傳輸

Netty EventLoop Thread allocation OIO

與NIO方式的不同在於,一個EventLoop只會服務於一個Channel。

NioEventLoop & NioEventLoopGroup 建立

初步瞭解了 EventLoop 以及 EventLoopGroup 的工作機制,接下來我們以 NioEventLoopGroup 為例,來深入分析 NioEventLoopGroup 是如何建立的,又是如何啟動的,它的內部執行邏輯又是怎樣的等等問題。

MultithreadEventExecutorGroup 構造器

我們從 NioEventLoopGroup 的建構函式開始分析:

EventLoopGroup acceptorEventLoopGroup = new NioEventLoopGroup(1);

NioEventLoopGroup建構函式會呼叫到父類 MultithreadEventLoopGroup 的建構函式,預設情況下,EventLoop的數量 = 處理器數量 x 2:

public abstract class MultithreadEventLoopGroup extends MultithreadEventExecutorGroup implements EventLoopGroup {

    private static final InternalLogger logger = InternalLoggerFactory.getInstance(MultithreadEventLoopGroup.class);

    private static final int DEFAULT_EVENT_LOOP_THREADS;

    // 預設情況下,EventLoop的數量 = 處理器數量 x 2
    static {
        DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt(
                "io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));

        if (logger.isDebugEnabled()) {
            logger.debug("-Dio.netty.eventLoopThreads: {}", DEFAULT_EVENT_LOOP_THREADS);
        }
    }

    protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args)    {
        super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
    }

    ...
}

繼續呼叫父類,會呼叫到 MultithreadEventExecutorGroup 的構造器,主要做三件事情:

  • 建立執行緒任務執行器 ThreadPerTaskExecutor
  • 通過for迴圈建立數量為 nThreads 個的 EventLoop
  • 建立 EventLoop 選擇器 EventExecutorChooser
protected MultithreadEventExecutorGroup(int nThreads, Executor executor,
                                        EventExecutorChooserFactory chooserFactory, Object... args) {
    if (nThreads <= 0) {
        throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));
    }

    // 建立任務執行器 ThreadPerTaskExecutor
    if (executor == null) {
        executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
    }

    // 建立 EventExecutor 陣列
    children = new EventExecutor[nThreads];

    // 通過for迴圈建立數量為 nThreads 個的 EventLoop
    for (int i = 0; i < nThreads; i ++) {
        boolean success = false;
        try {
            // 呼叫 newChild 介面
            children[i] = newChild(executor, args);
            success = true;
        } catch (Exception e) {
            // TODO: Think about if this is a good exception type
            throw new IllegalStateException("failed to create a child event loop", e);
        } finally {
            if (!success) {
                for (int j = 0; j < i; j ++) {
                    children[j].shutdownGracefully();
                }

                for (int j = 0; j < i; j ++) {
                    EventExecutor e = children[j];
                    try {
                        while (!e.isTerminated()) {
                            e.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS);
                        }
                    } catch (InterruptedException interrupted) {
                        // Let the caller handle the interruption.
                        Thread.currentThread().interrupt();
                        break;
                    }
                }
            }
        }
    }
	
    // 建立選擇器
    chooser = chooserFactory.newChooser(children);

    final FutureListener<Object> terminationListener = new FutureListener<Object>() {
        @Override
        public void operationComplete(Future<Object> future) throws Exception {
            if (terminatedChildren.incrementAndGet() == children.length) {
                terminationFuture.setSuccess(null);
            }
        }
    };

    for (EventExecutor e: children) {
        e.terminationFuture().addListener(terminationListener);
    }

    Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);
    Collections.addAll(childrenSet, children);
    readonlyChildren = Collections.unmodifiableSet(childrenSet);
}

建立執行緒任務執行器 ThreadPerTaskExecutor

if (executor == null) {
    executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
}

執行緒任務執行器 ThreadPerTaskExecutor 原始碼如下,具體的任務都由 ThreadFactory 去執行:

public final class ThreadPerTaskExecutor implements Executor {
    private final ThreadFactory threadFactory;

    public ThreadPerTaskExecutor(ThreadFactory threadFactory) {
        if (threadFactory == null) {
            throw new NullPointerException("threadFactory");
        }
        this.threadFactory = threadFactory;
    }
	
    // 使用 threadFactory 執行任務
    @Override
    public void execute(Runnable command) {
        threadFactory.newThread(command).start();
    }
}

來看看 newDefaultThreadFactory 方法:

protected ThreadFactory newDefaultThreadFactory() {
    return new DefaultThreadFactory(getClass());
}

DefaultThreadFactory

接下來看看 DefaultThreadFactory 這個類,實現了 ThreadFactory 介面,我們可以瞭解到:

  • EventLoopGroup的命名規則
  • 具體的執行緒為 FastThreadLocalThread
public class DefaultThreadFactory implements ThreadFactory {
    
	// 執行緒池ID編號自增器
    private static final AtomicInteger poolId = new AtomicInteger();
	// 執行緒ID自增器
    private final AtomicInteger nextId = new AtomicInteger();
    // 執行緒名稱字首
    private final String prefix;
    // 是否為守護程序
    private final boolean daemon;
    // 執行緒優先順序
    private final int priority;
    // 執行緒組
    protected final ThreadGroup threadGroup;

    public DefaultThreadFactory(Class<?> poolType) {
        this(poolType, false, Thread.NORM_PRIORITY);
    }

    ...

    // 獲取執行緒名,返回結果:nioEventLoopGroup
    public static String toPoolName(Class<?> poolType) {
        if (poolType == null) {
            throw new NullPointerException("poolType");
        }

        String poolName = StringUtil.simpleClassName(poolType);
        switch (poolName.length()) {
            case 0:
                return "unknown";
            case 1:
                return poolName.toLowerCase(Locale.US);
            default:
                if (Character.isUpperCase(poolName.charAt(0)) && Character.isLowerCase(poolName.charAt(1))) {
                    return Character.toLowerCase(poolName.charAt(0)) + poolName.substring(1);
                } else {
                    return poolName;
                }
        }
    }

    public DefaultThreadFactory(String poolName, boolean daemon, int priority, ThreadGroup threadGroup) {
        if (poolName == null) {
            throw new NullPointerException("poolName");
        }
        if (priority < Thread.MIN_PRIORITY || priority > Thread.MAX_PRIORITY) {
            throw new IllegalArgumentException(
                    "priority: " + priority + " (expected: Thread.MIN_PRIORITY <= priority <= Thread.MAX_PRIORITY)");
        }
		
        // nioEventLoopGroup-2-
        prefix = poolName + '-' + poolId.incrementAndGet() + '-';
        this.daemon = daemon;
        this.priority = priority;
        this.threadGroup = threadGroup;
    }
	
    public DefaultThreadFactory(String poolName, boolean daemon, int priority) {
        this(poolName, daemon, priority, System.getSecurityManager() == null ?
                Thread.currentThread().getThreadGroup() : System.getSecurityManager().getThreadGroup());
    }
	
    @Override
    public Thread newThread(Runnable r) {
        // 建立新執行緒 nioEventLoopGroup-2-1
        Thread t = newThread(FastThreadLocalRunnable.wrap(r), prefix + nextId.incrementAndGet());
        try {
            if (t.isDaemon() != daemon) {
                t.setDaemon(daemon);
            }
            if (t.getPriority() != priority) {
                t.setPriority(priority);
            }
        } catch (Exception ignored) {
            // Doesn't matter even if failed to set.
        }
        return t;
    }
	
    // 建立新執行緒 FastThreadLocalThread
    protected Thread newThread(Runnable r, String name) {
        return new FastThreadLocalThread(threadGroup, r, name);
    }
    
}

建立NioEventLoop

繼續從 MultithreadEventExecutorGroup 構造器開始,建立完任務執行器 ThreadPerTaskExecutor 之後,進入for迴圈,開始建立 NioEventLoop:

for (int i = 0; i < nThreads; i ++) {
    boolean success = false;
    try {
        // 建立 nioEventLoop
        children[i] = newChild(executor, args);
        success = true;
    } catch (Exception e) {
        // TODO: Think about if this is a good exception type
        throw new IllegalStateException("failed to create a child event loop", e);
    }
    
    ...
 	
}    

NioEventLoopGroup類中的 newChild() 方法:

@Override
protected EventLoop newChild(Executor executor, Object... args) throws Exception {
    return new NioEventLoop(this, executor, (SelectorProvider) args[0],
        ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]);
}

NioEventLoop 構造器:

public final class NioEventLoop extends SingleThreadEventLoop{
    
    ...
    
    NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider, SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler) {
        // 呼叫父類 SingleThreadEventLoop 構造器
        super(parent, executor, false, DEFAULT_MAX_PENDING_TASKS, rejectedExecutionHandler);
        if (selectorProvider == null) {
            throw new NullPointerException("selectorProvider");
        }
        if (strategy == null) {
            throw new NullPointerException("selectStrategy");
        }
        // 設定 selectorProvider
        provider = selectorProvider;
        // 獲取 SelectorTuple 物件,裡面封裝了原生的selector和優化過的selector
        final SelectorTuple selectorTuple = openSelector();
        // 設定優化過的selector
        selector = selectorTuple.selector;
        // 設定原生的selector
        unwrappedSelector = selectorTuple.unwrappedSelector;
        // 設定select策略
        selectStrategy = strategy;
    }
	
	...
    
}

接下來我們看看 獲取多路複用選擇器 方法—— openSelector() ,

// selectKey 優化選項flag
private static final boolean DISABLE_KEYSET_OPTIMIZATION =
    SystemPropertyUtil.getBoolean("io.netty.noKeySetOptimization", false);

private SelectorTuple openSelector() {
    // JDK原生的selector
    final Selector unwrappedSelector;
    try {
        // 通過 SelectorProvider 建立獲得selector
        unwrappedSelector = provider.openSelector();
    } catch (IOException e) {
        throw new ChannelException("failed to open a new selector", e);
    }

    // 如果不優化,則直接返回
    if (DISABLE_KEYSET_OPTIMIZATION) {
        return new SelectorTuple(unwrappedSelector);
    }

    // 通過反射建立 sun.nio.ch.SelectorImpl 物件
    Object maybeSelectorImplClass = AccessController.doPrivileged(new PrivilegedAction<Object>() {
        @Override
        public Object run() {
            try {
                return Class.forName(
                        "sun.nio.ch.SelectorImpl",
                        false,
                        PlatformDependent.getSystemClassLoader());
            } catch (Throwable cause) {
                return cause;
            }
        }
    });
    
    // 如果 maybeSelectorImplClass 不是 selector 的一個實現,則直接返回原生的Selector 
    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);
    }
    // maybeSelectorImplClass 是selector的實現,則轉化為 selector 實現類
    final Class<?> selectorImplClass = (Class<?>) maybeSelectorImplClass;
    // 建立新的 SelectionKey 集合 SelectedSelectionKeySet,內部採用的是 SelectionKey 陣列的形
    // 式,而非 set 集合
    final SelectedSelectionKeySet selectedKeySet = new SelectedSelectionKeySet();

    Object maybeException = AccessController.doPrivileged(new PrivilegedAction<Object>() {
       @Override
       public Object run() {
           try {
            // 通過反射的方式獲取 sun.nio.ch.SelectorImpl 的成員變數 selectedKeys
            Field selectedKeysField = selectorImplClass.getDeclaredField("selectedKeys");
            // 通過反射的方式獲取 sun.nio.ch.SelectorImpl 的成員變數 publicSelectedKeys
            Field publicSelectedKeysField = selectorImplClass.getDeclaredField