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的启动过程，即构造器和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从这个方法的逻辑中可以看出来，创建一个连接的过程分为两个主要的步骤;
值得注意的是，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,promise);
}
}
});
return promise;
}
initAndRegister仍然分为两个步骤：
注意看源码中的一段注释，这段注释对netty的线程模型的理解很有帮助，大致意思是说：
final ChannelFuture initAndRegister() { // 注册到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。 然后经过几次调用，最后调用了下面的构造器，首先调用了父类AbstractNioByteChannel的构造器， 我们在接着看父类AbstractNioByteChannel的构造方法 AbstractNioByteChannel(Channel parent,SelectableChannel ch)没有做任何工作，直接调用了父类的构造方法，注意这里多了一个参数SelectionKey.OP_READ，这个参数表示channel初始时的感兴趣的事件，channel刚创建好之后对read事件感兴趣 AbstractNioChannel(Channel parent,SelectableChannel ch,int readInterestOp)主要还是调用父类的构造方法 protected AbstractNioChannel(Channel parent,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的初始化过程：
MultithreadEventLoopGroup.register在完成channel的创建和初始化之后，我们就要将这个channel注册到一个EventLoop中，NioNioEventLoop继承自MultithreadEventLoopGroup,通过调用SingleThreadEventLoop的register方法完成注册 public ChannelFuture register(Channel channel) {
return next().register(channel);
}
可以看到，通过next()方法选出了其中的一个EventLoop进行注册。MultithreadEventLoopGroup是对多个真正的EventLoopGroup的封装，每个实现了实际功能的真正的EventLoopGroup运行在一个线程内， SingleThreadEventLoop.register这里创建了一个DefaultChannelPromise对象，用于作为返回值。 public ChannelFuture register(Channel channel) {
return register(new DefaultChannelPromise(channel,this));
}
最终调用了Unsafe的register方法将channel绑定到当前的EventLoopGroup对象上。 AbstractChannel.AbstractUnsafe.register
AbstractChannel.AbstractUnsafe.register0这个方法实现了实际的注册逻辑，
小结到此，我们就完成了对channel的创建，初始化，和注册到EventLoop过程的分析，整个过程看下来，其实并不复杂，只不过代码的嵌套比较深，继承结构复杂，有些简单的功能可能要看好几层才能找到真正实现的地方，所以还需要耐心和熟悉。这里，我把主干逻辑再提炼一下，去掉所有细枝末节的逻辑，一遍能有一个整体的认识：
接下来，我们回到Bootstrap.doResolveAndConnect方法中，继续完成建立连接的过程的分析。 Bootstrap.doResolveAndConnect0连接的建立在方法doResolveAndConnect0中实现: 这个方法的主要工作就是对远程地址进行解析，比如通过dns服务器对域名进行解析， 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,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(),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(),promise);
}
}
});
} catch (Throwable cause) {
promise.tryFailure(cause);
}
return promise;
}
Bootstrap.doConnect调用channel的connect方法完成连接过程。 private static void doConnect(
final SocketAddress remoteAddress,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,connectPromise);
}
connectPromise.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
}
});
}
AbstractChannel.connectpublic 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 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,promise);
} else {
safeExecute(executor,new Runnable() {
@Override
public void run() {
next.invokeConnect(remoteAddress,promise);
}
},promise,null);
}
return promise;
}
DefaultChannelPipeline.HeadContext.connectpublic void connect(
ChannelHandlerContext ctx,SocketAddress localAddress,ChannelPromise promise) {
unsafe.connect(remoteAddress,promise);
}
unsafe对象的赋值： HeadContext(DefaultChannelPipeline pipeline) {
super(pipeline,null,HEAD_NAME,HeadContext.class);
unsafe = pipeline.channel().unsafe();
setAddComplete();
}
所以我们直接看unsafe.connect AbstractNioChannel.connect主要逻辑：
可见建立连接的核心方法是doConnect，这是一个抽象方法，我们看NioSocketChannel，也就是tcp连接的建立过程，查看AbstractNioChannel的实现类发现还有UDP,SCTP等协议 public final void connect(
final SocketAddress remoteAddress,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可以看到，最终是通过调用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();
}
}
总结一句话，这代码是真的很深! 非常不直接，初次看的话，如果没有一个代码框架图在旁边参考，很容易迷失在层层的继承结构中，很多代码层层调用，真正有用的逻辑隐藏的很深，所以看这中代码必须要有耐心，有毅力，要有打破砂锅问到底的决心。不过这样的复杂的代码结构好处也是显而易见的，那就是良好的扩展性，你可以在任意层级进行扩展。 总结一下建立连接的过程，我认为可以归结为三个主要的方面：
（编辑：李大同） 【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容! |