【Netty之旅四】你必定看得懂的Netty客戶端啓動源碼分析！

前言

前面小飛已經講解了NIO和Netty服務端啓動，這一講是Client的啓動過程。java

源碼系列的文章依舊仍是遵循大白話+畫圖的風格來說解，本文Netty源碼及之後的文章版本都基於：4.1.22.Final算法

本篇是以NettyClient啓動爲切入點，帶你們一步步進入Netty源碼的世界。編程

Client啓動流程揭祕

一、探祕的入口：netty-client demo

這裏用netty-exmaple中的EchoClient來做爲例子：bootstrap

public final class EchoClient {
    public static void main(String[] args) throws Exception {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
             .channel(NioSocketChannel.class)
             .option(ChannelOption.TCP_NODELAY, true)
             .handler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();
                     p.addLast(new EchoClientHandler());
                 }
             });

            ChannelFuture f = b.connect(HOST, PORT).sync();

            f.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }
}

代碼沒有什麼獨特的地方，咱們上一篇文章時也梳理過Netty網絡編程的一些套路，這裏就再也不贅述了。 (忘記的小朋友能夠查看Netty系列文章中查找~)數組

上面的客戶端代碼雖然簡單, 可是卻展現了Netty 客戶端初始化時所需的全部內容:服務器

EventLoopGroup：Netty服務端或者客戶端，都必須指定EventLoopGroup，客戶端指定的是NioEventLoopGroup
Bootstrap: Netty客戶端啓動類，負責客戶端的啓動和初始化過程
channel()類型：指定Channel的類型，由於這裏是客戶端，因此使用的是NioSocketChannel，服務端會使用NioServerSocketChannel
Handler：設置數據的處理器
bootstrap.connect(): 客戶端鏈接netty服務的方法

二、NioEventLoopGroup 流程解析

咱們先從NioEventLoopGroup開始，一行行代碼解析，先看看其類結構：網絡

上面是大體的類結構，而 EventLoop 又繼承自EventLoopGroup，因此類的大體結構咱們可想而知。這裏一些核心邏輯會在MultithreadEventExecutorGroup中，包含EventLoopGroup的建立和初始化操做等。app

接着從NioEventLoopGroup構造方法開始看起，一步步往下跟（代碼都只展現重點的部分，省去不少暫時不須要關心的代碼，如下代碼都遵循這個原則）：socket

EventLoopGroup group = new NioEventLoopGroup();

public NioEventLoopGroup() {
    this(0);
}

public NioEventLoopGroup(int nThreads, Executor executor, final SelectorProvider selectorProvider) {
    this(nThreads, executor, selectorProvider, DefaultSelectStrategyFactory.INSTANCE);
}

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

這裏經過調用this()和super()方法一路往下傳遞，期間會構造一些默認屬性，一直傳遞到MultithreadEventExecutorGroup類中，接着往西看。ide

2.一、MultithreadEventExecutorGroup

上面構造函數有一個重要的參數傳遞：DEFAULT_EVENT_LOOP_THREADS，這個值默認是CPU核數 * 2。

爲何要傳遞這個參數呢？咱們以前說過EventLoopGroup能夠理解成一個線程池，MultithreadEventExecutorGroup有一個線程數組EventExecutor[] children屬性，而傳遞過來的DEFAULT_EVENT_LOOP_THREADS就是數組的長度。

先看下MultithreadEventExecutorGroup中的構造方法：

protected MultithreadEventExecutorGroup(int nThreads, Executor executor,
                                            EventExecutorChooserFactory chooserFactory, Object... args) {
    if (executor == null) {
        executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
    }
    
    children = new EventExecutor[nThreads];
    
    for (int i = 0; i < nThreads; i ++) {
        children[i] = newChild(executor, args);
    }
    
    // ... 省略
}

這段代碼執行邏輯能夠理解爲：

經過ThreadPerTaskExecutor構造一個Executor執行器，後面會細說，裏面包含了線程執行的execute()方法
接着建立一個EventExecutor數組對象，大小爲傳遞進來的threads數量，這個所謂的EventExecutor能夠理解爲咱們的EventLoop，在這個demo中就是NioEventLoop對象
最後調用 newChild 方法逐個初始化EventLoopGroup中的EventLoop對象

上面只是大概說了下MultithreadEventExecutorGroup中的構造方法作的事情，後面還會一個個詳細展開，先不用着急，咱們先有個總體的認知就好。

再回到MultithreadEventExecutorGroup中的構造方法入參中，有個EventExecutorChooserFactory對象，這裏面是有個很亮眼的細節設計，經過它咱們來洞悉Netty的良苦用心。

2.一、亮點設計：DefaultEventExecutorChooserFactory

EventExecutorChooserFactory這個類的做用是用來選擇EventLoop執行器的，咱們知道EventLoopGroup是一個包含了CPU * 2個數量的EventLoop數組對象，那每次選擇EventLoop來執行任務是選擇數組中的哪個呢？

咱們看一下這個類的具體實現，紅框中都是須要重點查看的地方：

DefaultEventExecutorChooserFactory是一個選擇器工廠類，調用裏面的next()方法達到一個輪詢選擇的目的。

數組的長度是length，執行第n次，取數組中的哪一個元素就是對length取餘

繼續回到代碼的實現，這裏的優化就是在於先經過isPowerOfTwo()方法判斷數組的長度是否爲2的n次冪，判斷的方式很巧妙，使用val & -val == val，這裏我不作過多的解釋，網上還有不少判斷2的n次冪的優秀解法，我就不班門弄斧了。(可參考：https://leetcode-cn.com/problems/power-of-two/solution/2de-mi-by-leetcode/)

固然我認爲這裏還有更容易理解的一個算法：x & (x - 1) == 0 你們能夠看下面的圖就懂了，這裏就不延展了：

BUT!!! 這裏爲何要去煞費苦心的判斷數組的長度是2的n次冪？

不知道小夥伴們是否還記得大明湖畔的HashMap？通常咱們要求HashMap數組的長度須要是2的n次冪，由於在key值尋找數組位置的方法：(n - 1) & hash n是數組長度，這裏若是數組長度是2的n次冪就能夠經過位運算來提高性能，當length爲2的n次冪時下面公式是等價的：

n & (length - 1) <=> n % length

還記得上面說過，數組的長度默認都是CPU * 2，而通常服務器CPU核心數都是二、四、八、16等等，因此這一個小優化就很實用了，再仔細想一想，原來數組長度的初始化也是很講究的。

這裏位運算的好處就是效率遠遠高於與運算，Netty針對於這個小細節都作了優化，真是太棒了。

2.三、線程執行器：ThreadPerTaskExecutor

接着看下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;
    }

    @Override
    public void execute(Runnable command) {
        threadFactory.newThread(command).start();
    }
}

傳遞進來的threadFactory爲DefaultThreadFactory，這裏面會構造NioEventLoop線程命名規則爲nioEventLoop-1-xxx，咱們就不細看這個了。當線程執行的時候會調用execute()方法，這裏會建立一個FastThreadLocalThread線程，具體看代碼：

public class DefaultThreadFactory implements ThreadFactory {
    @Override
    public Thread newThread(Runnable r) {
        Thread t = newThread(FastThreadLocalRunnable.wrap(r), prefix + nextId.incrementAndGet());
        return t;
    }

    protected Thread newThread(Runnable r, String name) {
        return new FastThreadLocalThread(threadGroup, r, name);
    }
}

這裏經過newThread()來建立一個線程，而後初始化線程對象數據，最終會調用到Thread.init()中。

2.四、EventLoop初始化

接着繼續看MultithreadEventExecutorGroup構造方法:

protected MultithreadEventExecutorGroup(int nThreads, Executor executor,
                                            EventExecutorChooserFactory chooserFactory, Object... args) {
    children = new EventExecutor[nThreads];
    for (int i = 0; i < nThreads; i ++) {
        children[i] = newChild(executor, args);
        // .... 省略部分代碼
    }
}

上面代碼的最後一部分是 newChild 方法, 這個是一個抽象方法, 它的任務是實例化 EventLoop 對象. 咱們跟蹤一下它的代碼, 能夠發現, 這個方法在 NioEventLoopGroup 類中實現了, 其內容很簡單:

@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(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider,
                 SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler) {
    super(parent, executor, false, DEFAULT_MAX_PENDING_TASKS, 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;
}

其實就是實例化一個 NioEventLoop 對象, 而後返回。NioEventLoop構造函數中會保存provider和事件輪詢器selector，在其父類中還會建立一個MpscQueue隊列，而後保存線程執行器executor。

再回過頭來想想，MultithreadEventExecutorGroup 內部維護了一個 EventExecutor[] children數組, Netty 的 EventLoopGroup 的實現機制其實就創建在 MultithreadEventExecutorGroup 之上。

每當 Netty 須要一個 EventLoop 時, 會調用 next() 方法從EventLoopGroup數組中獲取一個可用的 EventLoop對象。其中next方法的實現是經過NioEventLoopGroup.next()來完成的，就是用的上面有過講解的經過輪詢算法來計算得出的。

最後總結一下整個 EventLoopGroup 的初始化過程:

EventLoopGroup(實際上是MultithreadEventExecutorGroup) 內部維護一個類型爲 EventExecutor children 數組，數組長度是nThreads
若是咱們在實例化 NioEventLoopGroup 時, 若是指定線程池大小, 則 nThreads 就是指定的值, 反之是處理器核心數 * 2
MultithreadEventExecutorGroup 中會調用 newChild 抽象方法來初始化 children 數組
抽象方法 newChild 是在 NioEventLoopGroup 中實現的, 它返回一個 NioEventLoop 實例.
NioEventLoop 屬性:
- SelectorProvider provider 屬性: NioEventLoopGroup 構造器中經過 SelectorProvider.provider() 獲取一個 SelectorProvider
- Selector selector 屬性: NioEventLoop 構造器中經過調用經過 selector = provider.openSelector() 獲取一個 selector 對象.

2.五、NioSocketChannel

在Netty中，Channel是對Socket的抽象，每當Netty創建一個鏈接後，都會有一個與其對應的Channel實例。

咱們在開頭的Demo中，設置了channel(NioSocketChannel.class)，NioSocketChannel的類結構以下：

接着分析代碼，當咱們調用b.channel()時實際上會進入AbstractBootstrap.channel()邏輯，接着看AbstractBootstrap中代碼：

public B channel(Class<? extends C> channelClass) {
    if (channelClass == null) {
        throw new NullPointerException("channelClass");
    }
    return channelFactory(new ReflectiveChannelFactory<C>(channelClass));
}

public ReflectiveChannelFactory(Class<? extends T> clazz) {
    if (clazz == null) {
        throw new NullPointerException("clazz");
    }
    this.clazz = clazz;
}

public B channelFactory(ChannelFactory<? extends C> channelFactory) {
    if (channelFactory == null) {
        throw new NullPointerException("channelFactory");
    }
    if (this.channelFactory != null) {
        throw new IllegalStateException("channelFactory set already");
    }

    this.channelFactory = channelFactory;
    return self();
}

能夠看到，這裏ReflectiveChannelFactory其實就是返回咱們指定的channelClass：NioSocketChannel, 而後指定AbstractBootstrap中的channelFactory = new ReflectiveChannelFactory()。

2.六、Channel初始化流程

到了這一步，咱們已經知道NioEventLoopGroup和channel()的流程，接着來看看Channel的初始化流程，這也是Netty客戶端啓動的的核心流程之一：

ChannelFuture f = b.connect(HOST, PORT).sync();

接着就開始從b.connect()爲入口一步步日後跟，先看下NioSocketChannel構造的總體流程：

從connet日後梳理下總體流程：

Bootstrap.connect -> Bootstrap.doResolveAndConnect -> AbstractBootstrap.initAndRegister

final ChannelFuture initAndRegister() {
    Channel channel = channelFactory.newChannel();
    init(channel);
    
    ChannelFuture regFuture = config().group().register(channel);
    return regFuture;
}

爲了更易讀，這裏代碼都作了簡化，只保留了一些重要的代碼。

緊接着咱們看看channelFactory.newChannel()作了什麼，這裏channelFactory是ReflectiveChannelFactory，咱們在上面的章節分析過：

@Override
public T newChannel() {
    try {
        return clazz.getConstructor().newInstance();
    } catch (Throwable t) {
        throw new ChannelException("Unable to create Channel from class " + clazz, t);
    }
}

這裏的clazz是NioSocketChannel，一樣是在上面章節講到過，這裏是調用NioSocketChannel的構造函數而後初始化一個Channel實例。

public class NioSocketChannel extends AbstractNioByteChannel implements io.netty.channel.socket.SocketChannel {
    public NioSocketChannel() {
        this(DEFAULT_SELECTOR_PROVIDER);
    }

    public NioSocketChannel(SelectorProvider provider) {
        this(newSocket(provider));
    }

    private static SocketChannel newSocket(SelectorProvider provider) {
        try {
            return provider.openSocketChannel();
        } catch (IOException e) {
            throw new ChannelException("Failed to open a socket.", e);
        }
    }
}

這裏其實也很簡單，就是建立一個Java NIO SocketChannel而已，接着看看NioSocketChannel的父類還作了哪些事情，這裏梳理下類的關係：

NioSocketChannel -> extends AbstractNioByteChannel -> exntends AbstractNioChannel

public abstract class AbstractNioChannel extends AbstractChannel {
    protected AbstractNioByteChannel(Channel parent, SelectableChannel ch) {
        super(parent, ch, SelectionKey.OP_READ);
    }

    protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
        super(parent);
        ch.configureBlocking(false);
    }
}

這裏會調用父類的構造參數，而且傳遞readInterestOp = SelectionKey.OP_READ:，這裏還有一個很重要的點，配置 Java NIO SocketChannel 爲非阻塞的，咱們以前在NIO章節的時候講解過，這裏也再也不贅述。

接着繼續看AbstractChannel的構造函數：

public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {
    protected AbstractChannel(Channel parent) {
        this.parent = parent;
        id = newId();
        unsafe = newUnsafe();
        pipeline = newChannelPipeline();
    }
}

這裏建立一個ChannelId，建立一個Unsafe對象，這裏的Unsafe並非Java中的Unsafe，後面也會講到。而後建立一個ChannelPipeline，後面也會講到，到了這裏，一個完整的NioSocketChannel 就初始化完成了，咱們再來總結一下：

Netty 的 SocketChannel 會與 Java 原生的 SocketChannel 綁定在一塊兒；
會註冊 Read 事件；
會爲每個 Channel 分配一個 channelId；
會爲每個 Channel 建立一個Unsafe對象；
會爲每個 Channel 分配一個 ChannelPipeline；

2.七、Channel 註冊流程

仍是回到最上面initAndRegister方法，咱們上面都是在分析裏面newChannel的操做，這個方法是NioSocketChannel建立的一個流程，接着咱們在繼續跟init()和register()的過程：

public abstract class AbstractBootstrap<B extends AbstractBootstrap<B, C>, C extends Channel> implements Cloneable {
    final ChannelFuture initAndRegister() {
        Channel channel = channelFactory.newChannel();
        init(channel);
        ChannelFuture regFuture = config().group().register(channel);
    }
}

init()就是將一些參數options和attrs設置到channel中，咱們重點須要看的是register方法，其調用鏈爲：

AbstractBootstrap.initAndRegister -> MultithreadEventLoopGroup.register -> SingleThreadEventLoop.register -> AbstractUnsafe.register

這裏最後到了unsafe的register()方法，最終調用到AbstractNioChannel.doRegister():

@Override
protected void doRegister() throws Exception {
    boolean selected = false;
    for (;;) {
        selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
        return;
    }
}

javaChannel()就是Java NIO中的SocketChannel，這裏是將SocketChannel註冊到與eventLoop相關聯的selector上。

最後咱們整理一下服務啓動的總體流程：

initAndRegister()初始化並註冊什麼呢？

channelFactory.newChannel()
經過反射建立一個 NioSocketChannel
將 Java 原生 Channel 綁定到 NettyChannel 中
註冊 Read 事件
爲 Channel 分配 id
爲 Channel 建立 unsafe對象
爲 Channel 建立 ChannelPipeline（默認是 head<=>tail 的雙向鏈表）

`init(channel)``

把 Bootstrap 中的配置設置到 Channel 中

register(channel)

把 Channel 綁定到一個 EventLoop 上
把 Java 原生 Channel、Netty 的 Channel、Selector 綁定到 SelectionKey 中
觸發 Register 相關的事件

2.8 unsafe初始化

上面有提到過在初始化Channel的過程當中會建立一個Unsafe的對象，而後綁定到Channel上：

protected AbstractChannel(Channel parent) {
    this.parent = parent;
    id = newId();
    unsafe = newUnsafe();
    pipeline = newChannelPipeline();
}

newUnsafe直接調用到了NioSocketChannel中的方法：

@Override
protected AbstractNioUnsafe newUnsafe() {
    return new NioSocketChannelUnsafe();
}

NioSocketChannelUnsafe是NioSocketChannel中的一個內部類，而後向上還有幾個父類繼承，這裏主要是對應到相關Java底層的Socket操做。

2.9 pipeline初始化

咱們仍是回到pipeline初始化的過程，來看一下newChannelPipeline()的具體實現：

protected DefaultChannelPipeline newChannelPipeline() {
    return new DefaultChannelPipeline(this);
}

protected DefaultChannelPipeline(Channel channel) {
    this.channel = ObjectUtil.checkNotNull(channel, "channel");
    succeededFuture = new SucceededChannelFuture(channel, null);
    voidPromise =  new VoidChannelPromise(channel, true);

    tail = new TailContext(this);
    head = new HeadContext(this);

    head.next = tail;
    tail.prev = head;
}

咱們調用 DefaultChannelPipeline 的構造器, 傳入了一個 channel, 而這個 channel 其實就是咱們實例化的 NioSocketChannel。

DefaultChannelPipeline 會將這個 NioSocketChannel 對象保存在channel 字段中. DefaultChannelPipeline 中, 還有兩個特殊的字段, 即 head 和 tail, 而這兩個字段是一個雙向鏈表的頭和尾. 其實在 DefaultChannelPipeline 中, 維護了一個以 AbstractChannelHandlerContext 爲節點的雙向鏈表, 這個鏈表是 Netty 實現 Pipeline 機制的關鍵.

關於 DefaultChannelPipeline 中的雙向鏈表以及它所起的做用, 咱們會在後續章節詳細講解。這裏只是對pipeline作個初步的認識。

HeadContext 的繼承層次結構以下所示:

TailContext 的繼承層次結構以下所示:

咱們能夠看到, 鏈表中 head 是一個 ChannelOutboundHandler, 而 tail 則是一個 ChannelInboundHandler.

3.0、客戶端connect過程

客戶端鏈接的入口方法仍是在Bootstrap.connect()中，上面也分析過一部份內容，請求的具體流程是：

Bootstrap.connect() -> AbstractChannel.coonnect() -> NioSocketChannel.doConnect()

public static boolean connect(final SocketChannel socketChannel, final SocketAddress remoteAddress)
            throws IOException {
    try {
        return AccessController.doPrivileged(new PrivilegedExceptionAction<Boolean>() {
            @Override
            public Boolean run() throws IOException {
                return socketChannel.connect(remoteAddress);
            }
        });
    } catch (PrivilegedActionException e) {
        throw (IOException) e.getCause();
    }
}

看到這裏，仍是用Java NIO SocketChannel發送的connect請求進行客戶端鏈接請求。

總結

本篇文章以一個Netty Client demo爲入口，而後解析了NioEventLoopGroup建立的流程、Channel的建立和註冊的流程，以及客戶端發起connect的具體流程，這裏對於不少細節並無很深的深刻下去，這些會放到後續的源碼分析文章，敬請期待~