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对象完成一些功能,如查状态,发出取消动作,实现阻塞等待等。
(0)

相关推荐