netty5笔记-总体流程分析5-客户端连接过程

前面几篇文章，我们从服务端的视角，分析了从启动到接收连接，到连接的read-write-close。现在我们开始切换到客户端的视角，看看客户端连接服务端的一些实现细节。

还是从snoop的example代码开始，见HttpSnoopClient（稍有修改）：

public static void main(String[] args) throws Exception {
// 配制client.
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(group)
.channel(NioSocketChannel.class)
.handler(new HttpSnoopClientInitializer(null));

// 创建连接.
Channel ch = b.connect("127.0.0.1", 80).sync().channel();

// 构造HTTP请求.
HttpRequest request = new DefaultFullHttpRequest(
HttpVersion.HTTP_1_1, HttpMethod.GET, uri.getRawPath());
request.headers().set(HttpHeaderNames.HOST, host);
request.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE);
request.headers().set(HttpHeaderNames.ACCEPT_ENCODING, HttpHeaderValues.GZIP);

// 发送HTTP请求.
ch.writeAndFlush(request);

// 等待服务端关闭连接
ch.closeFuture().sync();
} finally {
// 关闭group.
group.shutdownGracefully();
}
}

有了前面ServerBootstrap的铺垫，这里就比较简单了，ServerBootstrap和Bootstrap都继承自AbstractBootstrap。但需要注意的是，在ServerBootstrap中group、channel、handler都是针对NioServerSocketChannel的设置，而切到Bootstrap后，group、channel、handler则是针对NioSocketChannel的设置了。相应的在ServerBootstrap中设置NioSocketChannel的属性和选项使用childAttr、childOption，而Bootstrap中设置NioSocketChannel则是直接使用attr、option。

下面直接进入主题，b.connect("127.0.0.1", 80)创建了一个客户端的链接，调用的方法：

public ChannelFuture connect(String inetHost, int inetPort) {
return connect(InetSocketAddress.createUnresolved(inetHost, inetPort));
}

/**
* 创建一个连接.
*/
public ChannelFuture connect(SocketAddress remoteAddress) {
if (remoteAddress == null) {
throw new NullPointerException("remoteAddress");
}
// 主要验证必要的参数是否设置，如group何channelFactory(通过channel方法设置）
validate();
// 解析并连接
 return doResolveAndConnect(remoteAddress, localAddress());
}

    private ChannelFuture doResolveAndConnect(SocketAddress remoteAddress, final SocketAddress localAddress) {
// 初始化并注册到EventLoop（注册这块过程与server端类似，不再细讲）
        final ChannelFuture regFuture = initAndRegister();
        if (regFuture.cause() != null) {
            return regFuture;
        }

        final Channel channel = regFuture.channel();
        final EventLoop eventLoop = channel.eventLoop();
        final NameResolver<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.
            return doConnect(remoteAddress, localAddress, regFuture, channel.newPromise());
        }
// 对remoteAddress进行解析，如果地址为ip(127.0.0.1)则直接返回，如地址为域名(www.baidu.com)则需要解析为ip(180.97.33.107)
        final Future<SocketAddress> resolveFuture = resolver.resolve(remoteAddress);
        final Throwable resolveFailureCause = resolveFuture.cause();

        if (resolveFailureCause != null) {
            // Failed to resolve immediately
            channel.close();
            return channel.newFailedFuture(resolveFailureCause);
        }

// 在解析完成后调用doConnect(最终调用doConnect0)连接远程的端口
        if (resolveFuture.isDone()) {
            // Succeeded to resolve immediately; cached? (or did a blocking lookup)
            return doConnect(resolveFuture.getNow(), localAddress, regFuture, channel.newPromise());
        }

        // Wait until the name resolution is finished.
        final ChannelPromise connectPromise = channel.newPromise();
        resolveFuture.addListener(new FutureListener<SocketAddress>() {
            @Override
            public void operationComplete(Future<SocketAddress> future) throws Exception {
                if (future.cause() != null) {
                    channel.close();
                    connectPromise.setFailure(future.cause());
                } else {
                    doConnect(future.getNow(), localAddress, regFuture, connectPromise);
                }
            }
        });

        return connectPromise;
    }

// 初始化pipeline、各个option和attr
void init(Channel channel) throws Exception {
        ChannelPipeline p = channel.pipeline();
        p.addLast(handler());

        final Map<ChannelOption<?>, Object> options = options();
        synchronized (options) {
            for (Entry<ChannelOption<?>, Object> e: options.entrySet()) {
                try {
                    if (!channel.config().setOption((ChannelOption<Object>) e.getKey(), e.getValue())) {
                        logger.warn("Unknown channel option: " + e);
                    }
                } catch (Throwable t) {
                    logger.warn("Failed to set a channel option: " + channel, t);
                }
            }
        }

        final Map<AttributeKey<?>, Object> attrs = attrs();
        synchronized (attrs) {
            for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) {
                channel.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());
            }
        }
    }

private static void doConnect0(
            final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelFuture regFuture,
            final ChannelPromise connectPromise) {

        // 这个方法在 channelRegistered()之前调用. 这样就给handlers机会在channelRegistered()方法中设置pipeline
        final Channel channel = connectPromise.channel();
        channel.eventLoop().execute(new Runnable() {
            @Override
            public void run() {
                if (regFuture.isSuccess()) {
// 执行channel的connect方法
                    if (localAddress == null) {
                        channel.connect(remoteAddress, connectPromise);
                    } else {
                        channel.connect(remoteAddress, localAddress, connectPromise);
                    }
                    connectPromise.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
                } else {
                    connectPromise.setFailure(regFuture.cause());
                }
            }
        });
    }

这个过程比较简单，1、创建一个NioSocketChannel实例，并用Bootstrap中的参数初始化该实例；2、将创建的channel注册到EventLoop中；3、创建一个解析host的任务(如果是ip，任务直接完成，如果是域名需要将域名解析为ip)；4、解析完成后，在EventLoop中调用channel.connect，连接到远程端口。

channel.connect将connect任务交给pipeline去处理，最终调用到TailContext中的connect方法，该方法调用代码：unsafe.connect(...)。

// AbstractNioUnsafe
public final void connect(final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {
if (!promise.setUncancellable() || !ensureOpen(promise)) {
return;
}

try {
if (connectPromise != null) {
throw new IllegalStateException("connection attempt already made");
}

boolean wasActive = isActive();
// 创建连接，如果连接立即成功了则返回true, 否则会返回false
if (doConnect(remoteAddress, localAddress)) {
// 完成连接过程
fulfillConnectPromise(promise, wasActive);
} else {
connectPromise = promise;
requestedRemoteAddress = remoteAddress;

// 如果设置了连接超时时间，则创建一个超时检测任务，如果超时未连接成功则关闭连接
int connectTimeoutMillis = config().getConnectTimeoutMillis();
if (connectTimeoutMillis > 0) {
connectTimeoutFuture = eventLoop().schedule(new OneTimeTask() {
@Override
public void run() {
ChannelPromise connectPromise = AbstractNioChannel.this.connectPromise;
ConnectTimeoutException cause =
new ConnectTimeoutException("connection timed out: " + remoteAddress);
if (connectPromise != null && connectPromise.tryFailure(cause)) {
close(voidPromise());
}
}
}, connectTimeoutMillis, TimeUnit.MILLISECONDS);
}
// 如果连接被取消，则关闭前面创建的超时检测任务并关闭连接
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();
}
}

private void fulfillConnectPromise(ChannelPromise promise, boolean wasActive) {
            if (promise == null) {
                // Closed via cancellation and the promise has been notified already.
                return;
            }

            // 如果取消了连接则此处会返回false
            boolean promiseSet = promise.trySuccess();

            // 只要是连接确实打开了，则无论是否被取消channelActive都会触发
            if (!wasActive && isActive()) {
                pipeline().fireChannelActive();
            }

            // 如果用户取消了连接，则会调用close方法，此方法会触发channelInactive
            if (!promiseSet) {
                close(voidPromise());
            }
        }

// doConnect来自NioSocketChannel
protected boolean doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception {
        if (localAddress != null) {
            javaChannel().socket().bind(localAddress);
        }

        boolean success = false;
        try {
// 发起连接请求，由于是非阻塞模式，因此主要是两种情况：1、连接本地，可能会立即完成，此时返回true;2、其他情况，没有立即完成连接，返回false
            boolean connected = javaChannel().connect(remoteAddress);
            if (!connected) {
// 注册OP_CONNECT事件，如果连接成功则会进入NioEventLoop中的processSelectedKey方法
                selectionKey().interestOps(SelectionKey.OP_CONNECT);
            }
            success = true;
            return connected;
        } finally {
            if (!success) {
                doClose();
            }
        }
    }

// NioEventLoop中的processSelectedKey方法片段       
if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
                // 移除OP_CONNECT否则会导致cpu 100%
                int ops = k.interestOps();
                ops &= ~SelectionKey.OP_CONNECT;
                k.interestOps(ops);
// 此调用最终也会调用上面的fulfillConnectPromise，进而触发后面的channelActive
                unsafe.finishConnect();
            }

上面的连接过程也是比较简单：

1、发起连接请求。1.1如果立即成功则执行连接成功的后续处理，如channelActive方法的调用；1.2如果未立即成功，则将连接事件注册到Selector；

2、Selector检测到连接事件后触发unsafe.finishConnect，该方法最终也执行连接成功的后续处理（同1.1）;

3、1.2的分支会创建一个超时检测任务，如果超过指定时间未连接成功，则直接关闭此次连接请求。