相对于服务端，Netty 客户端 的创建更加复杂，除了要考虑线程模型、异步连接、客户端连接超时等因素外，还需要对连接过程中的各种异常进行考虑。本章将对 Netty 客户端 创建的关键流程和源码进行分析，以期读者能够了解客户端创建的细节。

基于 Netty 创建客户端的流程分析

Netty 为了向使用者屏蔽 NIO 通信 的底层细节，在和用户交互的边界做了封装，目的就是为了减少用户开发工作量，降低开发难度。Bootstrap 是 Socket 客户端创建工具类，用户通过 Bootstrap 可以方便地创建 Netty 的客户端并发起 异步 TCP 连接操作。

基于 Netty 创建客户端 时序图

avatar

Netty 创建客户端 流程分析

  1. 用户线程创建 Bootstrap 实例，通过 API 设置客户端相关的参数，异步发起客户端连接；
  2. 创建处理客户端连接、I/O 读写的 Reactor 线程组 NioEventLoopGroup。可以通过构造函数指定 IO 线程 的个数，默认为 CPU 内核数的 2 倍；
  3. 通过 Bootstrap 的 ChannelFactory 和用户指定的 Channel 类型 创建用于客户端连接的 NioSocketChannel，它的功能类似于 JDK NIO 类库 提供的 SocketChannel；
  4. 创建默认的 Channel、Handler、Pipeline，用于调度和执行网络事件；
  5. 异步发起 TCP 连接，判断连接是否成功。如果成功，则直接将 NioSocketChannel 注册到多路复用器上，监听读操作位，用于数据报读取和消息发送；如果没有立即连接成功，则注册连接监听位到多路复用器，等待连接结果；
  6. 注册对应的网络监听状态位到多路复用器；
  7. 由多路复用器在 IO 现场中轮询各 Channel，处理连接结果；
  8. 如果连接成功，设置 Future 结果，发送连接成功事件，触发 ChanneIPipeline 执行；
  9. 由 ChannelPipeline 调度执行系统和用户的 ChannelHandler，执行业务逻辑。

Netty 客户端创建源码分析

Netty 客户端 的创建流程比较繁琐，下面我们针对关键步骤和代码进行分析，通过梳理关键流程来掌握客户端创建的原理。

客户端连接辅助类 BootStrap

Bootstrap 是 Netty 提供的客户端连接工具类，主要用于简化客户端的创建，下面我们对它的 主要 API 进行讲解。

设置 lO 线程组：NIO 的特点就是，一个多路复用器可以同时处理上干条链路，这就意味着，NIO 模式中 一个线程可以处理多个 TCP 连接。考虑到 lO 线程 的处理性能，大多数 NIO 框架 都采用线程池的方式处理 IO 读写，Netty 也不例外。客户端相对于服务端，只需要一个处理 IO 读写 的线程组即可，因为 Bootstrap 提供了 设置 IO 线程组 的接口，代码如下。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

public abstract class AbstractBootstrap<B extends AbstractBootstrap<B, C>, C extends Channel> implements Cloneable {

    volatile EventLoopGroup group;

    public B group(EventLoopGroup group) {
        if (group == null) {
            throw new NullPointerException("group");
        } else if (this.group != null) {
            throw new IllegalStateException("group set already");
        } else {
            this.group = group;
            return this;
        }
    }
}

由于 Netty 的 NIO 线程组 默认采用 EventLoopGroup 接口，因此线程组参数使用 EventLoopGroup。

TCP 参数设置接口：无论是 NIO，还是 BIO，创建客户端套接字的时候通常都会设置连接参数，例如接收和发送缓冲区大小、连接超时时间等。Bootstrap 也提供了客户端 TCP 参数设置接口，代码如下。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

public <T> B option(ChannelOption<T> option, T value) {
    if (option == null) {
        throw new NullPointerException("option");
    } else {
        if (value == null) {
            synchronized(this.options) {
                this.options.remove(option);
            }
        } else {
            synchronized(this.options) {
                this.options.put(option, value);
            }
        }

        return this;
    }
}

Netty 提供的 主要 TCP 参数 如下。

1、SO_TIMEOUT：控制读取操作将阻塞多少毫秒。如果返回值为 0，计时器就被禁止了，该线程将无限期阻塞； 2、SO_SNDBUF：套接字使用的发送缓冲区大小； 3、SO_RCVBUF：套接字使用的接收缓冲区大小； 4、SO_REUSEADDR：用于决定 如果网络上仍然有数据向旧的 ServerSocket 传输数据，是否允许新的 ServerSocket 绑定到与旧的 ServerSocket 同样的端口上。SO_REUSEADDR 选项 的默认值与操作系统有关，在某些操作系统中，允许重用端口，而在某些操作系统中不允许重用端口； 5、CONNECT_TIMEOUT_MILLIS：客户端连接超时时间，由于 NIO 原生的客户端 并不提供设置连接超时的接口，因此，Netty 采用的是自定义连接超时定时器负责检测和超时控制；

Channel 接口：用于指定客户端使用的 Channel 接口，对于 TCP 客户端连接，默认使用 NioSocketChannel，代码如下。

1
2
3
4
5
6
7

public B channel(Class<? extends C> channelClass) {
    if (channelClass == null) {
        throw new NullPointerException("channelClass");
    } else {
        return this.channelFactory((io.netty.channel.ChannelFactory)(new ReflectiveChannelFactory(channelClass)));
    }
}

BootstrapChannelFactory 利用 参数 channelClass，通过反射机制创建 NioSocketChannel 对象。

设置 Handler 接口：Bootstrap 为了简化 Handler 的编排，提供了 Channellnitializer，它继承了 ChannelHandlerAdapter，当 TCP 链路 注册成功之后，调用 initChannel 接口，用于设置用户 ChanneIHandler。它的代码如下。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

public abstract class ChannelInitializer<C extends Channel> extends ChannelInboundHandlerAdapter {

    public final void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        if (this.initChannel(ctx)) {
            ctx.pipeline().fireChannelRegistered();
        } else {
            ctx.fireChannelRegistered();
        }
    }
}

最后一个比较重要的接口就是发起客户端连接，代码如下。

1

ChannelFuture f = b.connect(host, port).sync();

客户端连接操作

首先要创建和初始化 NioSocketChannel，代码如下。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91

public class Bootstrap extends AbstractBootstrap<Bootstrap, Channel> {

    public ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress) {
        if (remoteAddress == null) {
            throw new NullPointerException("remoteAddress");
        } else {
            this.validate();
            return this.doResolveAndConnect(remoteAddress, localAddress);
        }
    }

    private ChannelFuture doResolveAndConnect(final SocketAddress remoteAddress, final SocketAddress localAddress) {
    	// 首先要创建和初始化 NioSocketChannel
        ChannelFuture regFuture = this.initAndRegister();
        final Channel channel = regFuture.channel();
        if (regFuture.isDone()) {
        	// 初始化 Channel 之后，将其注册到 Selector 上
            return !regFuture.isSuccess() ? regFuture : this.doResolveAndConnect0(channel, remoteAddress, localAddress, channel.newPromise());
        } else {
            final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
            regFuture.addListener(new ChannelFutureListener() {
                public void operationComplete(ChannelFuture future) throws Exception {
                    Throwable cause = future.cause();
                    if (cause != null) {
                        promise.setFailure(cause);
                    } else {
                        promise.registered();
                        Bootstrap.this.doResolveAndConnect0(channel, remoteAddress, localAddress, promise);
                    }

                }
            });
            return promise;
        }
    }

    private ChannelFuture doResolveAndConnect0(final Channel channel, SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {
        try {
            EventLoop eventLoop = channel.eventLoop();
            AddressResolver<SocketAddress> resolver = this.resolver.getResolver(eventLoop);
            if (!resolver.isSupported(remoteAddress) || resolver.isResolved(remoteAddress)) {
                doConnect(remoteAddress, localAddress, promise);
                return promise;
            }

            Future<SocketAddress> resolveFuture = resolver.resolve(remoteAddress);
            if (resolveFuture.isDone()) {
                Throwable resolveFailureCause = resolveFuture.cause();
                if (resolveFailureCause != null) {
                    channel.close();
                    promise.setFailure(resolveFailureCause);
                } else {
                    doConnect((SocketAddress)resolveFuture.getNow(), localAddress, promise);
                }

                return promise;
            }

            resolveFuture.addListener(new FutureListener<SocketAddress>() {
                public void operationComplete(Future<SocketAddress> future) throws Exception {
                    if (future.cause() != null) {
                        channel.close();
                        promise.setFailure(future.cause());
                    } else {
                        Bootstrap.doConnect((SocketAddress)future.getNow(), localAddress, promise);
                    }

                }
            });
        } catch (Throwable var9) {
            promise.tryFailure(var9);
        }

        return promise;
    }

    private static void doConnect(final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise connectPromise) {
        final Channel channel = connectPromise.channel();
        channel.eventLoop().execute(new Runnable() {
            public void run() {
                if (localAddress == null) {
                    channel.connect(remoteAddress, connectPromise);
                } else {
                    channel.connect(remoteAddress, localAddress, connectPromise);
                }

                connectPromise.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            }
        });
    }
}

需要注意的是，SocketChannel 执行 connect() 操作后有以下三种结果。

  1. 连接成功，返回 True；
  2. 暂时没有连接上，服务端没有返回 ACK 应答，连接结果不确定，返回 False；
  3. 连接失败，直接抛出 IO 异常。

如果是第二种结果，需要将 NioSocketChannel 中的 selectionKey 设置为 OP_CONNECT，监听连接结果。异步连接返回之后，需要判断连接结果，如果连接成功，则触发 ChannelActive 事件。ChannelActive 事件 最终会将 NioSocketChannel 中的 selectionKey 设置为 SelectionKey.OP_READ，用于监听网络读操作。如果没有立即连接上服务端，则注册 SelectionKey.OP_CONNECT 到多路复用器。如果连接过程发生异常，则关闭链路，进入连接失败处理流程。

异步连接结果通知

NioEventLoop 的 Selector 轮询 客户端连接 Channel，当服务端返回握手应答之后，对连接结果进行判断，代码如下。

1
2
3
4
5
6

if ((readyOps & 8) != 0) {
    int ops = k.interestOps();
    ops &= -9;
    k.interestOps(ops);
    unsafe.finishConnect();
}

下面对 finishConnect()方法 进行分析，代码如下。

1
2
3
4
5
6
7

try {
    boolean wasActive = AbstractNioChannel.this.isActive();
    AbstractNioChannel.this.doFinishConnect();
    this.fulfillConnectPromise(AbstractNioChannel.this.connectPromise, wasActive);
} catch (Throwable var5) {
    ......
}

doFinishConnect()方法 用于判断 JDK 的 SocketChannel 的连接结果，如果未出错 表示连接成功，其他值或者发生异常表示连接失败。

1
2
3
4
5

protected void doFinishConnect() throws Exception {
    if (!this.javaChannel().finishConnect()) {
        throw new Error();
    }
}

连接成功之后，调用 fufillConectPromise()方法，触发链路激活事件，该事件由 ChannelPipeline 进行传播。

客户端连接超时机制

对于 SocketChannel 接口，JDK 并没有提供连接超时机制，需要 NIO 框架或者用户自己扩展实现。Netty 利用定时器提供了客户端连接超时控制功能，下面我们对该功能进行详细讲解。

首先,用户在创建 Netty 客户端的时候，可以通过 ChannelOption.CONNECT_TIMEOUT_MILLIS 配置项设置连接超时时间，代码如下。

1
2
3
4
5

Bootstrap b = new Bootstrap();
b.group(workerGroup);
b.channel(NioSocketChannel.class);
b.option(ChannelOption.SO_KEEPALIVE, true);
b.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000);