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對象完成一些功能,如查狀態,發出取消動作,實現阻塞等待等。

posted on 2019-06-23 02:15 _朱葛 閱讀(...) 評論(...) 編輯 收藏

公告

統計

  • 隨筆 - 37
  • 文章 - 0
  • 評論 - 0
内蒙古快3开奖结果