Bootstrap初始化過程源碼分析--netty客戶端的啓動

Bootstrap初始化過程

netty的客戶端引導類是Bootstrap，咱們看一下spark的rpc中客戶端部分對Bootstrap的初始化過程

TransportClientFactory.createClient(InetSocketAddress address)

只須要貼出Bootstrap初始化部分的代碼

// 客戶端引導對象
Bootstrap bootstrap = new Bootstrap();
// 設置各類參數
bootstrap.group(workerGroup)
  .channel(socketChannelClass)
  // Disable Nagle's Algorithm since we don't want packets to wait
  // 關閉Nagle算法
  .option(ChannelOption.TCP_NODELAY, true)
  .option(ChannelOption.SO_KEEPALIVE, true)
  .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, conf.connectionTimeoutMs())
  .option(ChannelOption.ALLOCATOR, pooledAllocator);

// socket接收緩衝區
if (conf.receiveBuf() > 0) {
  bootstrap.option(ChannelOption.SO_RCVBUF, conf.receiveBuf());
}

// socket發送緩衝區
// 對於接收和發送緩衝區的設置應該用以下的公式計算：
// 延遲 *帶寬
// 例如延遲是1ms，帶寬是10Gbps，那麼緩衝區大小應該設爲1.25MB
if (conf.sendBuf() > 0) {
  bootstrap.option(ChannelOption.SO_SNDBUF, conf.sendBuf());
}

final AtomicReference<TransportClient> clientRef = new AtomicReference<>();
final AtomicReference<Channel> channelRef = new AtomicReference<>();

// 設置handler（處理器對象）
bootstrap.handler(new ChannelInitializer<SocketChannel>() {
  @Override
  public void initChannel(SocketChannel ch) {
    TransportChannelHandler clientHandler = context.initializePipeline(ch);
    clientRef.set(clientHandler.getClient());
    channelRef.set(ch);
  }
});

// Connect to the remote server
long preConnect = System.nanoTime();
// 與服務端創建鏈接，啓動方法
ChannelFuture cf = bootstrap.connect(address);

分爲幾個主要的步驟：

• 首先建立一個Bootstrap對象，調用的是無參構造器
• 設置各類參數，如通道類型，關閉Nagle算法，接收和發送緩衝區大小，設置處理器
• 調用connect與服務端創建鏈接

接下來，咱們主要經過兩條線索來分析Bootstrap的啓動過程，即構造器和connect兩個方法，而對於設置參數的過程僅僅是給內部的一些成員變量賦值，因此不須要詳細展開。

Bootstrap.Bootstrap()

Bootstrap繼承了AbstractBootstrap，看了一下他們的無參構造方法，都是個空方法。。。。。。因此這一步，咱們就省了，瞬間感受飛起來了有沒有^_^

Bootstrap.connect(SocketAddress remoteAddress)

public ChannelFuture connect(SocketAddress remoteAddress) {
    // 檢查非空
    ObjectUtil.checkNotNull(remoteAddress, "remoteAddress");
    // 一樣是對一些成員變量檢查非空，主要檢查EventLoopGroup，ChannelFactory，handler對象
    validate();
    return doResolveAndConnect(remoteAddress, config.localAddress());
}

主要是作了一些非空檢查，須要注意的是，ChannelFactory對象的設置，前面的spark中在對Bootstrap初始化設置的時候調用了.channel(socketChannelClass)方法，這個方法以下：

public B channel(Class<? extends C> channelClass) {
    return channelFactory(new ReflectiveChannelFactory<C>(
            ObjectUtil.checkNotNull(channelClass, "channelClass")
    ));
}

建立了一個ReflectiveChannelFactory對象，並賦值給內部的channelFactory成員。這個工廠類會根據傳進來的Class對象經過反射建立一個Channel實例。

doResolveAndConnect

從這個方法的邏輯中能夠看出來，建立一個鏈接的過程分爲兩個主要的步驟;

• 初始化一個Channel對象並註冊到EventLoop中
• 調用doResolveAndConnect0方法完成tcp鏈接的創建

值得注意的是，initAndRegister方法返回一個Future對象，這個類型一般用於異步機制的實現。在這裏，若是註冊沒有當即成功的話，會給返回的futrue對象添加一個監聽器，在註冊成功之後創建tcp鏈接。

private ChannelFuture doResolveAndConnect(final SocketAddress remoteAddress, final SocketAddress localAddress) {
    // 初始化一個Channel對象並註冊到EventLoop中
    final ChannelFuture regFuture = initAndRegister();
    final Channel channel = regFuture.channel();

    if (regFuture.isDone()) {
        // 若是註冊失敗，世界返回失敗的future對象
        if (!regFuture.isSuccess()) {
            return regFuture;
        }
        return doResolveAndConnect0(channel, remoteAddress, localAddress, channel.newPromise());
    } else {// 若是註冊還在進行中，須要向future對象添加一個監聽器，以便在註冊成功的時候作一些工做，監聽器實際上就是一個回調對象
        // Registration future is almost always fulfilled already, but just in case it's not.
        final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
        regFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                // Directly obtain the cause and do a null check so we only need one volatile read in case of a
                // failure.
                Throwable cause = future.cause();
                if (cause != null) {
                    // Registration on the EventLoop failed so fail the ChannelPromise directly to not cause an
                    // IllegalStateException once we try to access the EventLoop of the Channel.
                    promise.setFailure(cause);
                } else {
                    // Registration was successful, so set the correct executor to use.
                    // See https://github.com/netty/netty/issues/2586
                    promise.registered();
                    // 註冊成功後仍然調用doResolveAndConnect0方法完成鏈接創建的過程
                    doResolveAndConnect0(channel, remoteAddress, localAddress, promise);
                }
            }
        });
        return promise;
    }

initAndRegister

仍然分爲兩個步驟：

• 經過channel工廠類建立一個channel對象，經過反射獲取指定的channel類型的無參構造器，調用構造器來建立對象
• 調用init方法對channel對象進行初始化，init方法是一個抽象方法，Bootstrap和ServerBootstrap的實現不一樣
• 將channel註冊到EventLoopGroup中

注意看源碼中的一段註釋，這段註釋對netty的線程模型的理解頗有幫助，大體意思是說：

• 若是當前的代碼是在EventLoopEvent線程中執行的，那麼代碼運行到這裏說明channel已經成功註冊到EventLoopEvent上了，此時再調用bind() 或 connect()方法確定是沒有問題的
• 若是當前代碼不是在EventLoopEvent線程中執行的，也就是說當前線程是另外的線程，在這裏繼續調用bind() 或 connect()方法仍然是安全的，並不會因爲併發引發方法執行順序的錯亂，緣由是netty中一個channel只會綁定到一個線程上，全部關於這個channel的操做包括註冊，bind或connect都會以排隊任務的形式在一個線程中串行執行，這種作法也爲netty規避了不少線程安全問題，從而減小了不少加鎖，同步的代碼，減小了線程之間的競爭資源致使的線程切換，側面上提升了線程執行效率。

final ChannelFuture initAndRegister() {
Channel channel = null;
try {
// 經過channel工廠類建立一個channel對象
channel = channelFactory.newChannel();
// 調用init方法對channel進行一些初始化的設置
init(channel);
} catch (Throwable t) {
if (channel != null) {
// channel can be null if newChannel crashed (eg SocketException("too many open files"))
channel.unsafe().closeForcibly();
// as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor
return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t);
}
// as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor
return new DefaultChannelPromise(new FailedChannel(), GlobalEventExecutor.INSTANCE).setFailure(t);
}

// 註冊到EventLoopGroup中
    ChannelFuture regFuture = config().group().register(channel);
    // 若是發生異常，須要關閉已經創建的鏈接
    if (regFuture.cause() != null) {
        if (channel.isRegistered()) {
            channel.close();
        } else {
            channel.unsafe().closeForcibly();
        }
    }

    // If we are here and the promise is not failed, it's one of the following cases:
    // 1) If we attempted registration from the event loop, the registration has been completed at this point.
    //    i.e. It's safe to attempt bind() or connect() now because the channel has been registered.
    // 2) If we attempted registration from the other thread, the registration request has been successfully
    //    added to the event loop's task queue for later execution.
    //    i.e. It's safe to attempt bind() or connect() now:
    //         because bind() or connect() will be executed *after* the scheduled registration task is executed
    //         because register(), bind(), and connect() are all bound to the same thread.
    
    return regFuture;
}

NioSocketChannel初始化

DEFAULT_SELECTOR_PROVIDER是默認的SelectorProvider對象，這時jdk中定義的一個類，主要做用是生成選擇器selector對象和通道channel對象

public NioSocketChannel() {
    this(DEFAULT_SELECTOR_PROVIDER);
}

newSocket中經過調用provider.openSocketChannel()方法建立了一個SocketChannel對象，它的默認實現是SocketChannelImpl。
public NioSocketChannel(SelectorProvider provider) {
this(newSocket(provider));
}

而後通過幾回調用，最後調用了下面的構造器，首先調用了父類AbstractNioByteChannel的構造器，
而後建立了一個SocketChannelConfig對象，這個類有點相似於門面模式，對NioSocketChannel對象和Socket對象的一些參數設置和獲取的接口進行封裝。
public NioSocketChannel(Channel parent, SocketChannel socket) {
super(parent, socket);
config = new NioSocketChannelConfig(this, socket.socket());
}

咱們在接着看父類AbstractNioByteChannel的構造方法

AbstractNioByteChannel(Channel parent, SelectableChannel ch)

沒有作任何工做，直接調用了父類的構造方法，注意這裏多了一個參數SelectionKey.OP_READ，這個參數表示channel初始時的感興趣的事件，channel剛建立好以後對read事件感興趣
protected AbstractNioByteChannel(Channel parent, SelectableChannel ch) {
super(parent, ch, SelectionKey.OP_READ);
}

AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp)

主要仍是調用父類的構造方法

protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
    // 父類構造方法
    super(parent);
    this.ch = ch;
    this.readInterestOp = readInterestOp;
    try {
        // 設置非阻塞
        ch.configureBlocking(false);
    } catch (IOException e) {
        try {
            // 若是發生異常，關閉該channel
            ch.close();
        } catch (IOException e2) {
            if (logger.isWarnEnabled()) {
                logger.warn(
                        "Failed to close a partially initialized socket.", e2);
            }
        }

        throw new ChannelException("Failed to enter non-blocking mode.", e);
    }
}

AbstractChannel(Channel parent)

最關鍵的初始化邏輯在這個最頂層的基類中，其中很重的兩個對象Unsafe對象和ChannelPipeline對象，前者封裝了jdk底層api的調用，後者是實現netty對事件的鏈式處理的核心類。

protected AbstractChannel(Channel parent) {
    this.parent = parent;
    // 建立一個ChannelId對象，惟一標識該channel
    id = newId();
    // Unsafe對象，封裝了jdk底層的api調用
    unsafe = newUnsafe();
    // 建立一個DefaultChannelPipeline對象
    pipeline = newChannelPipeline();
}

小結

前面一小節，咱們主要簡單分析了一下NioSocketChannel的初始化過程，能夠看到最主要的邏輯在AbstractChannel的構造方法中，這裏咱們看到了兩個重要的類的建立過程。

Bootstrap.init

回到AbstractBootstrap.initAndRegister方法中，在完成經過反射調用NioSocketChannel構造方法並建立一個實例後，緊接着就要對這個新建立的Channel實例進行初始化設置工做，咱們看一下Bootstrap對新建立的Channel的初始化過程：

• 向channel的Pipeline中添加一個處理器，ChannelPipeline咱們能夠理解爲一個流水線，在這條流水線上有各類各樣的處理器，一個channel事件產生後會在這個流水線上進行傳播，依次通過全部的處理器
• 設置參數，也就是以ChannelOption爲key的一些參數，能夠經過DefaultChannelConfig.setOption方法看到具體能夠設置哪些參數。

• 設置屬性

  void init(Channel channel) throws Exception {
  ChannelPipeline p = channel.pipeline();
  // 向ChannelPipeline中添加一個處理器，這個處理器就是咱們以前設置的處理器
  p.addLast(config.handler());

  final Map<ChannelOption<?>, Object> options = options0();
    // 設置參數，最終是經過調用SocketChannelConfig的一些參數設置接口設置參數
    synchronized (options) {
        setChannelOptions(channel, options, logger);
    }
  
    final Map<AttributeKey<?>, Object> attrs = attrs0();
    // 設置屬性
    synchronized (attrs) {
        for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) {
            channel.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());
        }
    }

  }

MultithreadEventLoopGroup.register

在完成channel的建立和初始化以後，咱們就要將這個channel註冊到一個EventLoop中，NioNioEventLoop繼承自MultithreadEventLoopGroup, 經過調用SingleThreadEventLoop的register方法完成註冊

public ChannelFuture register(Channel channel) {
    return next().register(channel);
}

能夠看到，經過next()方法選出了其中的一個EventLoop進行註冊。MultithreadEventLoopGroup是對多個真正的EventLoopGroup的封裝，每一個實現了實際功能的真正的EventLoopGroup運行在一個線程內，
因此咱們接下來應該看單個的EventLoopGroup的註冊方法。

SingleThreadEventLoop.register

這裏建立了一個DefaultChannelPromise對象，用於做爲返回值。

public ChannelFuture register(Channel channel) {
    return register(new DefaultChannelPromise(channel, this));
}

最終調用了Unsafe的register方法將channel綁定到當前的EventLoopGroup對象上。
public ChannelFuture register(final ChannelPromise promise) {
ObjectUtil.checkNotNull(promise, "promise");
promise.channel().unsafe().register(this, promise);
return promise;
}

AbstractChannel.AbstractUnsafe.register

• 首先是作一些前置檢查，包括變量非空檢查，重複註冊檢查，檢查channel類型和EventLoopGroup類型是否匹配
• 將這個channel綁定到指定的eventLoop對象上，

• 調用register0完成註冊

  public final void register(EventLoop eventLoop, final ChannelPromise promise) {
        // 作一些非空檢查
        if (eventLoop == null) {
            throw new NullPointerException("eventLoop");
        }
        // 若是重複註冊，經過future對象拋出一個異常
        // 一個channel只能註冊到一個EventLoopGroup對象上
        if (isRegistered()) {
            promise.setFailure(new IllegalStateException("registered to an event loop already"));
            return;
        }
        // 檢查channel類型和EventLoopGroup類型是否匹配
        if (!isCompatible(eventLoop)) {
            promise.setFailure(
                    new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));
            return;
        }
  
        // 將channel內部的eventLoop成員設置爲相應的對象
        // 也就是將這個channel綁定到指定頂eventLoop上
        AbstractChannel.this.eventLoop = eventLoop;
  
        // 這裏作了一個判斷，若是當前處於eventLoop對應的線程內，那麼直接執行代碼
        // 若是當前運行的線程與eventLoop不是同一個，那麼將這個註冊的任務添加到eventLoop的任務隊列中
        if (eventLoop.inEventLoop()) {
            register0(promise);
        } else {
            try {
                eventLoop.execute(new Runnable() {
                    @Override
                    public void run() {
                        register0(promise);
                    }
                });
            } catch (Throwable t) {
                logger.warn(
                        "Force-closing a channel whose registration task was not accepted by an event loop: {}",
                        AbstractChannel.this, t);
                closeForcibly();
                closeFuture.setClosed();
                safeSetFailure(promise, t);
            }
        }
    }

AbstractChannel.AbstractUnsafe.register0

這個方法實現了實際的註冊邏輯，

• 依然要作一些前置的設置和檢查工做，包括在註冊過程當中不可取消，檢查channel是否存活，
• 調用jdk的api完成註冊。例如，對於jdk Nio的通道的註冊就是調用SelectableChannel.register(Selector sel, int ops, Object att)
• 調用全部的已添加的處理器節點的ChannelHandler.handlerAdded方法,實際上這也會調用handler.handlerRemoved方法，若是在此以前有handler被移除掉的話
• 通知future對象已經註冊成功了
• 觸發一個channel註冊成功的事件，這個事件會在pipeline中傳播，全部註冊的handler會依次接收到該事件並做出相應的處理

• 若是是第一次註冊，還須要觸發一個channel存活的事件，讓全部的handler做出相應的處理

  private void register0(ChannelPromise promise) {
        try {
            // check if the channel is still open as it could be closed in the mean time when the register
            // call was outside of the eventLoop
            // 將ChannelPromise設置爲不可取消，並檢查channel是否還存活,經過內部的jdk的channel檢查是否存活
            if (!promise.setUncancellable() || !ensureOpen(promise)) {
                return;
            }
            // 是否第一次註冊，
            // TODO 說明狀況下會註冊屢次？？
            boolean firstRegistration = neverRegistered;
            // 完成實際的註冊，即底層api的調用
            // 若是對於jdk Nio的通道的註冊就是調用SelectableChannel.register(Selector sel, int ops, Object att)
            doRegister();
            // 更新標誌變量
            neverRegistered = false;
            registered = true;
  
            // Ensure we call handlerAdded(...) before we actually notify the promise. This is needed as the
            // user may already fire events through the pipeline in the ChannelFutureListener.
            // 調用全部的已添加的處理器節點的ChannelHandler.handlerAdded方法
            pipeline.invokeHandlerAddedIfNeeded();
  
            // 經過future對象已經註冊成功了
            safeSetSuccess(promise);
            // 觸發一個channel註冊成功的事件，這個事件會在pipeline中傳播，
            // 全部註冊的handler會依次接收到該事件並做出相應的處理
            pipeline.fireChannelRegistered();
            // Only fire a channelActive if the channel has never been registered. This prevents firing
            // multiple channel actives if the channel is deregistered and re-registered.
            if (isActive()) {
                if (firstRegistration) {
                    // 若是是第一次註冊，還須要觸發一個channel存活的事件，讓全部的handler做出相應的處理
                    pipeline.fireChannelActive();
                } else if (config().isAutoRead()) {
                    // This channel was registered before and autoRead() is set. This means we need to begin read
                    // again so that we process inbound data.
                    //
                    // See https://github.com/netty/netty/issues/4805
                    // 開始接收讀事件
                    // 對於Nio類型的channel, 經過調用jdk的相關api註冊讀事件爲感興趣的事件
                    beginRead();
                }
            }
        } catch (Throwable t) {
            // Close the channel directly to avoid FD leak.
            closeForcibly();
            closeFuture.setClosed();
            safeSetFailure(promise, t);
        }
    }

小結

到此，咱們就完成了對channel的建立，初始化，和註冊到EventLoop過程的分析，整個過程看下來，其實並不複雜，只不過代碼的嵌套比較深，繼承結構複雜，有些簡單的功能可能要看好幾層才能找到真正實現的地方，因此還須要耐心和熟悉。這裏，我把主幹邏輯再提煉一下，去掉全部細枝末節的邏輯，一遍能有一個總體的認識：

• 首先經過反射建立了一個NioSocketChannel(經過反射調用無參構造器)
• 而後對channel對象進行初始化，主要是想這個channel的ChannelPipeline中添加用戶設置的handler
• 最後將這個channel註冊到一個EventLoop上，註冊過程設計jdk底層的selector註冊api的調用，調用handler的回調方法，在channelPipeline中觸發一個channel註冊的事件，這些事件最終回調各個handler對象的channelRegistered方法。

接下來，咱們回到Bootstrap.doResolveAndConnect方法中，繼續完成創建鏈接的過程的分析。

Bootstrap.doResolveAndConnect0

鏈接的創建在方法doResolveAndConnect0中實現:

這個方法的主要工做就是對遠程地址進行解析，好比經過dns服務器對域名進行解析，
而後使用解析後的地址進行鏈接的創建，鏈接創建調用doConnect方法

private ChannelFuture doResolveAndConnect0(final Channel channel, SocketAddress remoteAddress,
                                           final SocketAddress localAddress, final ChannelPromise promise) {
    try {
        final EventLoop eventLoop = channel.eventLoop();
        // 獲取一個地址解析器
        final AddressResolver<SocketAddress> resolver = this.resolver.getResolver(eventLoop);

        // 若是解析器不支持該地址或者改地址已經被解析過了，那麼直接開始建立鏈接
        if (!resolver.isSupported(remoteAddress) || resolver.isResolved(remoteAddress)) {
            // Resolver has no idea about what to do with the specified remote address or it's resolved already.
            doConnect(remoteAddress, localAddress, promise);
            return promise;
        }

        // 對遠程地址進行解析
        final Future<SocketAddress> resolveFuture = resolver.resolve(remoteAddress);

        if (resolveFuture.isDone()) {
            final Throwable resolveFailureCause = resolveFuture.cause();

            if (resolveFailureCause != null) {
                // Failed to resolve immediately
                channel.close();
                promise.setFailure(resolveFailureCause);
            } else {
                // Succeeded to resolve immediately; cached? (or did a blocking lookup)
                // 解析成功後進行鏈接
                doConnect(resolveFuture.getNow(), localAddress, promise);
            }
            return promise;
        }

        // Wait until the name resolution is finished.
        // 給future對象添加一個回調，採用異步方法進行鏈接，
        resolveFuture.addListener(new FutureListener<SocketAddress>() {
            @Override
            public void operationComplete(Future<SocketAddress> future) throws Exception {
                if (future.cause() != null) {
                    channel.close();
                    promise.setFailure(future.cause());
                } else {
                    doConnect(future.getNow(), localAddress, promise);
                }
            }
        });
    } catch (Throwable cause) {
        promise.tryFailure(cause);
    }
    return promise;
}

Bootstrap.doConnect

調用channel的connect方法完成鏈接過程。
也許是以前看scala代碼習慣了，回過頭來看java代碼感受很冗餘，一大堆代碼就表達了那一點邏輯，感受信息密度過低，如今有不少人認爲java會漸漸的沒落，而最優可能取代java的語言中，scala絕對是強有力的競爭者之一，沒有對比就沒有傷害，跟java比,scala語言真的是簡潔太多了，幾句話就能把所要表達的邏輯精準而又直接地表達出來。好像向聲明式編程更靠近了一點。

private static void doConnect(
        final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise connectPromise) {

    // This method is invoked before channelRegistered() is triggered.  Give user handlers a chance to set up
    // the pipeline in its channelRegistered() implementation.
    final Channel channel = connectPromise.channel();
    channel.eventLoop().execute(new Runnable() {
        @Override
        public void run() {
            if (localAddress == null) {
                // 調用 channel.connect方法完成鏈接
                channel.connect(remoteAddress, connectPromise);
            } else {
                channel.connect(remoteAddress, localAddress, connectPromise);
            }
            connectPromise.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
        }
    });
}

AbstractChannel.connect

public ChannelFuture connect(SocketAddress remoteAddress, ChannelPromise promise) {
    return pipeline.connect(remoteAddress, promise);
}

DefaultChannelPipeline.connect

這裏稍微說明一下，tail是整個鏈條的尾節點，若是對netty比較熟悉的話，應該知道netty對於io事件的處理採用責任鏈的模式，即用戶能夠設置多個處理器，這些處理器組成一個鏈條，io事件在這個鏈條上傳播，被特定的一些處理器所處理，而其中有兩個特殊的處理器head和tail，他們分別是這個鏈條的頭和尾，他們的存在主要是爲了實現一些特殊的邏輯。

public final ChannelFuture connect(SocketAddress remoteAddress, ChannelPromise promise) {
    return tail.connect(remoteAddress, promise);
}

AbstractChannelHandlerContext.connect

中間通過幾個調用以後，最終調用該方法。這裏有一句關鍵代碼findContextOutbound(MASK_CONNECT)，這個方法的代碼我就不貼了，大概說一下它的做用，更爲具體的機制等後面分析Channelpipeline是在詳細說明。這個方法會在處理器鏈中從後向前遍歷，直到找到可以處理connect事件的處理器，可否處理某種類型的事件是經過比特位判斷的，每一個AbstractChannelHandlerContext對象內部有一個int型變量用於存儲標誌各類類型事件的比特位。通常，connect事件會有頭結點head來處理，也就是DefaultChannelPipeline.HeadContext類，因此咱們直接看DefaultChannelPipeline.HeadContext.connect方法

public ChannelFuture connect(
        final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {

    if (remoteAddress == null) {
        throw new NullPointerException("remoteAddress");
    }
    if (isNotValidPromise(promise, false)) {
        // cancelled
        return promise;
    }

    // 找到下一個可以進行connect操做的，這裏用比特位來標記各類不一樣類型的操做，
    final AbstractChannelHandlerContext next = findContextOutbound(MASK_CONNECT);
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        // 調用AbstractChannelHandlerContext.invokeConnect
        next.invokeConnect(remoteAddress, localAddress, promise);
    } else {
        safeExecute(executor, new Runnable() {
            @Override
            public void run() {
                next.invokeConnect(remoteAddress, localAddress, promise);
            }
        }, promise, null);
    }
    return promise;
}

DefaultChannelPipeline.HeadContext.connect

public void connect(
            ChannelHandlerContext ctx,
            SocketAddress remoteAddress, SocketAddress localAddress,
            ChannelPromise promise) {
        unsafe.connect(remoteAddress, localAddress, promise);
    }

unsafe對象的賦值：

HeadContext(DefaultChannelPipeline pipeline) {
        super(pipeline, null, HEAD_NAME, HeadContext.class);
        unsafe = pipeline.channel().unsafe();
        setAddComplete();
    }

因此咱們直接看unsafe.connect

AbstractNioChannel.connect

主要邏輯：

• 狀態檢查，非空檢查
• 調用doConnect方法進行鏈接
• 若是當即就鏈接成功了，那麼將future對象設置爲成功
• 若是超時大於0，會提交一個延遲調度的任務，在超時時間到達後執行這個任務檢查是否鏈接成功，若是爲鏈接成功鏈接說明鏈接超時，須要關閉通道
• 向future對象添加一個回調，在future被外部調用者取消時將通道關閉

可見創建鏈接的核心方法是doConnect，這是一個抽象方法，咱們看NioSocketChannel，也就是tcp鏈接的創建過程，查看AbstractNioChannel的實現類發現還有UDP,SCTP等協議

public final void connect(
            final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {
        // 檢查promise狀態，channel存活狀態
        if (!promise.setUncancellable() || !ensureOpen(promise)) {
            return;
        }

        try {
            // 防止重複鏈接
            if (connectPromise != null) {
                // Already a connect in process.
                throw new ConnectionPendingException();
            }

            boolean wasActive = isActive();
            // 調用doConnect方法進行鏈接
            if (doConnect(remoteAddress, localAddress)) {
                // 若是當即就鏈接成功了，那麼將future對象設置爲成功
                fulfillConnectPromise(promise, wasActive);
            } else {
                connectPromise = promise;
                requestedRemoteAddress = remoteAddress;

                // Schedule connect timeout.
                int connectTimeoutMillis = config().getConnectTimeoutMillis();
                // 若是超時大於0，那麼會在超時到達後檢查是否鏈接成功
                if (connectTimeoutMillis > 0) {
                    connectTimeoutFuture = eventLoop().schedule(new Runnable() {
                        @Override
                        public void run() {
                            ChannelPromise connectPromise = AbstractNioChannel.this.connectPromise;
                            ConnectTimeoutException cause =
                                    new ConnectTimeoutException("connection timed out: " + remoteAddress);
                            // 若是connectPromise可以標記爲失敗，說明此時尚未鏈接成功，也就是鏈接超時了
                            // 此時須要關閉該通道
                            if (connectPromise != null && connectPromise.tryFailure(cause)) {
                                close(voidPromise());
                            }
                        }
                    }, connectTimeoutMillis, TimeUnit.MILLISECONDS);
                }

                // 向future對象添加一個回調，在future被外部調用者取消時將通道關閉
                promise.addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        if (future.isCancelled()) {
                            if (connectTimeoutFuture != null) {
                                connectTimeoutFuture.cancel(false);
                            }
                            connectPromise = null;
                            close(voidPromise());
                        }
                    }
                });
            }
        } catch (Throwable t) {
            promise.tryFailure(annotateConnectException(t, remoteAddress));
            closeIfClosed();
        }
    }

NioSocketChannel.doConnect

• 首先綁定指定的本地地址

• 調用SocketUtils.connect創建鏈接

  protected boolean doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception {
  // 綁定指定的本地地址
  if (localAddress != null) {
  doBind0(localAddress);
  }

  // 這個變量標記創建鏈接的動做是否發起成功
    // 成功發起創建鏈接的工做並不表示鏈接已經成功創建
    boolean success = false;
    try {
        // 實際創建鏈接的語句
        boolean connected = SocketUtils.connect(javaChannel(), remoteAddress);
        if (!connected) {
            selectionKey().interestOps(SelectionKey.OP_CONNECT);
        }
        success = true;
        // 返回鏈接是否已經成功創建
        return connected;
    } finally {
        if (!success) {
            doClose();
        }
    }

  }

SocketUtils.connect

能夠看到，最終是經過調用jdk的api來實現鏈接的創建，也就是SocketChannel.connect方法

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 {
                // 調用jdk api創建鏈接，SocketChannel.connect
                return socketChannel.connect(remoteAddress);
            }
        });
    } catch (PrivilegedActionException e) {
        throw (IOException) e.getCause();
    }
}

總結

一句話，這代碼是真的很深! 很是不直接，初次看的話，若是沒有一個代碼框架圖在旁邊參考，很容易迷失在層層的繼承結構中，不少代碼層層調用，真正有用的邏輯隱藏的很深，因此看這中代碼必需要有耐心，有毅力，要有打破砂鍋問到底的決心。不過這樣的複雜的代碼結構好處也是顯而易見的，那就是良好的擴展性，你能夠在任意層級進行擴展。

總結一下創建鏈接的過程，我認爲能夠歸結爲三個主要的方面：

• 第一, 實際創建邏輯的代碼確定仍是jdk api
• 第二，這麼多方法調用，主要的做用就是迎合框架的要求，本質上是爲了代碼的擴展性，好比ChannelPipeline的處理器鏈
• 第三，另外一個主要的工做就是對future對象的處理，這時實現異步的重要手段，future對象也是外部調用者和對象內部狀態之間的鏈接紐帶，調用者經過future對象完成一些功能，如查狀態，發出取消動做，實現阻塞等待等。