在本章节准备分析下客户端的启动流程，其实其中流程已经涉及到了netty的几大基本模块，但是本文不会详细深入每个模块，重点在于走通流程，对于netty有一个框架上的大概认识。先上netty源码的demo：
public final class EchoClient {
static final boolean SSL = System.getProperty("ssl") != null;
static final String HOST = System.getProperty("host", "127.0.0.1");
static final int PORT = Integer.parseInt(System.getProperty("port", "8007"));
static final int SIZE = Integer.parseInt(System.getProperty("size", "256"));
public static void main(String[] args) throws Exception {
// Configure SSL.git
final SslContext sslCtx;
if (SSL) {
sslCtx = SslContextBuilder.forClient()
.trustManager(InsecureTrustManagerFactory.INSTANCE).build();
} else {
sslCtx = null;
}
// Configure the client.
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(group)
.channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
if (sslCtx != null) {
p.addLast(sslCtx.newHandler(ch.alloc(), HOST, PORT));
}
//p.addLast(new LoggingHandler(LogLevel.INFO));
p.addLast(new EchoClientHandler());
}
});
// Start the client.
ChannelFuture f = b.connect(HOST, PORT).sync();
// Wait until the connection is closed.
f.channel().closeFuture().sync();
} finally {
// Shut down the event loop to terminate all threads.
group.shutdownGracefully();
}
}
}
上述代码并不复杂，不过却基本上涵盖了客户端初始化所需要的全部内容：
- EventLoopGroup: NioEventLoopGroup是与Reactor 线程模型有对应关系的，主要管理eventLoop的生命周期，这个后续再细说。EventLoopGroup。不论是服务器端还是客户端, 都必须指定 EventLoopGroup. 在这个例子中, 指定了 NioEventLoopGroup, 表示一个 NIO 的EventLoopGroup.
- ChannelType: 指定 Channel 的类型. 因为是客户端, 因此使用了 NioSocketChannel.
- Handler: 设置数据的处理器.
接下来我们跟着代码分析下客户端的启动过程都做了什么工作。
EventLoopGroup group = new NioEventLoopGroup();
可以看作传统IO编程模型的线程组，其中的初始化后续再详说。
Bootstrap b = new Bootstrap();
b.group(group)
上述代码指定引导类（Bootstrap）的线程模型
.channel(NioSocketChannel.class)
然后，我们指定我们客户端IO 模型为NIO，我们通过过.channel(NioSocketChannel.class)来指定IO模型，当然，这里也有其他的选择，如果你想指定 IO 模型为 BIO，那么这里配置上OioServerSocketChannel.class类型即可。
.option(ChannelOption.TCP_NODELAY, true)
option() 方法可以给连接设置一些 TCP 底层相关的属性：
- ChannelOption.CONNECT_TIMEOUT_MILLIS 表示连接的超时时间，超过这个时间还是建立不上的话则代表连接失败
- ChannelOption.SO_KEEPALIVE 表示是否开启 TCP 底层心跳机制，true 为开启
- ChannelOption.TCP_NODELAY 表示是否开始 Nagle 算法，true 表示关闭，false 表示开启，通俗地说，如果要求高实时性，有数据发送时就马上发送，就设置为 true 关闭，如果需要减少发送次数减少网络交互，就设置为 false 开启。
.channel(...)
给引导类指定一个 handler，这里主要就是定义连接的业务处理逻辑，后面详细分析。
ChannelFuture f = b.connect(HOST, PORT).sync();
配置完线程模型、IO 模型、业务处理逻辑之后，调用 connect 方法进行连接，可以看到 connect 方法有两个参数，第一个参数可以填写 IP 或者域名，第二个参数填写的是端口号，由于 connect 方法返回的是一个 Future，也就是说这个方是异步的，我们通过 addListener 方法可以监听到连接是否成功，进而打印出连接信息。
上述说明了客户端启动的基本配置，接下来从代码层面来详细说明下各项配置初始化和连接过程。
EventLoopGroup的初始化
EventLoopGroup group = new NioEventLoopGroup();
NioEventLoopGroup有几个重载的构造器, 不过内容都没有什么大的区别, 最终都是调用的父类MultithreadEventLoopGroup构造器:
protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {
super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
}
如果我们传入的线程数 nThreads 是0, 那么 Netty 会为我们设置默认的线程数 DEFAULT_EVENT_LOOP_THREADS，很明显DEFAULT_EVENT_LOOP_THREADS是cpu核心数*2。
DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt(
"io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));
回到MultithreadEventLoopGroup会继续调用其父类MultithreadEventExecutorGroup构造器，让我们来看下其中的关键代码：
protected MultithreadEventExecutorGroup(int nThreads, Executor executor,
EventExecutorChooserFactory chooserFactory, Object... args) {
children = new EventExecutor[nThreads];
for (int i = 0; i < nThreads; i ++) {
children[i] = newChild(executor, args);
}
chooser = chooserFactory.newChooser(children);
}
public EventExecutorChooser newChooser(EventExecutor[] executors) {
if (isPowerOfTwo(executors.length)) {
return new PowerOfTwoEventExecutorChooser(executors);
} else {
return new GenericEventExecutorChooser(executors);
}
}
根据代码, 我们就很清楚 MultithreadEventExecutorGroup 中的处理逻辑了:
- 创建一个大小为 nThreads 的 Executor 数组；
- 调用 newChhild 方法初始化 children 数组；
- 根据 nThreads 的大小, 创建不同的 Chooser, 即如果 nThreads 是 2 的幂, 则使用 PowerOfTwoEventExecutorChooser, 反之使用GenericEventExecutorChooser. 不论使用哪个 Chooser, 它们的功能都是一样的, 即从 children 数组中选出一个合适的 EventExecutor 实例.
根据上面的代码, 我们知道, MultithreadEventExecutorGroup 内部维护了一个 EventExecutor 数组, Netty 的 EventLoopGroup 的实现机制其实就建立在 MultithreadEventExecutorGroup 之上. 每当 Netty 需要一个 EventLoop 时, 会调用 next() 方法获取一个可用的 EventLoop.我们来看下newChild(executor, args);其具体实现是在NioEventLoopGroup。
protected EventLoop newChild(Executor executor, Object... args) throws Exception {
return new NioEventLoop(this, executor, (SelectorProvider) args[0],
((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]);
}
具体的NioEventLoop等到我们讲NioEventLoop的时候再来具体说明。
总结下NioEventLoopGroup初始化所做的事情：
- EventLoopGroup(其实是MultithreadEventExecutorGroup) 内部维护一个类型为 - EventExecutor children 数组, 其大小是 nThreads, 这样就构成了一个线程池
如果我们在实例化 NioEventLoopGroup 时, 如果指定线程池大小, 则 nThreads 就是指定的值, 反之是处理器核心数 * 2 - MultithreadEventExecutorGroup 中会调用 newChild 抽象方法来初始化 children 数组
- 抽象方法 newChild 是在 NioEventLoopGroup 中实现的, 它返回一个 NioEventLoop 实例。
NioSocketChannel初始化
.channel(NioSocketChannel.class)：
public B channel(Class<? extends C> channelClass) {
if (channelClass == null) {
throw new NullPointerException("channelClass");
}
return channelFactory(new ReflectiveChannelFactory<C>(channelClass));
}
继续往下走可以看到以下代码：
public B channelFactory(ChannelFactory<? extends C> channelFactory) {
this.channelFactory = channelFactory;
}
注意其中的this.channelFactory = channelFactory; 后续真正生成NioSocketChannel对象的的地方用的就是ReflectiveChannelFactory.newChannel()。
@Override
public T newChannel() {
try {
return clazz.getConstructor().newInstance();
} catch (Throwable t) {
throw new ChannelException("Unable to create Channel from class " + clazz, t);
}
}
channel的实例化
由前面分析可知channel是通过ReflectiveChannelFactory.newChannel()实例化的，那么问题来了，具体实例化是在哪里呢？别急，我们继续往下走。从b.connect(HOST, PORT)往下看，可以发现最终的调用链是如何的：
Bootstrap.connet() --> Bootstrap.doResolveAndConnect() --> AbstractBootstrap.initAndRegister()
我们来看下initAndRegister()方法，我这边只列了关键代码：
final ChannelFuture initAndRegister() {
channel = channelFactory.newChannel();
init(channel);
ChannelFuture regFuture = config().group().register(channel);
}
创建一个channel对象，调用NioSocketChannel默认构造函数，默认构造函数如下：
private static final SelectorProvider DEFAULT_SELECTOR_PROVIDER = SelectorProvider.provider();
public NioSocketChannel() {
this(DEFAULT_SELECTOR_PROVIDER);
}
public NioSocketChannel(SelectorProvider provider) {
this(newSocket(provider));
}
private static SocketChannel newSocket(SelectorProvider provider) {
return provider.openSocketChannel();
}
从上面代码可以看出在newSocket会打开一个SocketChannel，Java NIO。继续往下探索，会进入父类AbstractNioByteChannel
protected AbstractNioByteChannel(Channel parent, SelectableChannel ch) {
super(parent, ch, SelectionKey.OP_READ);
}
然后继续调用父类AbstractNioChannel的构造函数：
protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
super(parent);
this.ch = ch;
this.readInterestOp = readInterestOp;
ch.configureBlocking(false);
}
上述代码应该有点熟悉吧，ch.configureBlocking(false);在Java NIO概述中有讲到如果要用nio需要设置为非阻塞的。在此处再次调用了父类AbstractChannel构造函数：
protected AbstractChannel(Channel parent) {
this.parent = parent;
id = newId();
unsafe = newUnsafe();
pipeline = newChannelPipeline();
}
在这里注意下unsafe和pipeline，尤其是pipeline，说明netty在一个channel中包含了pipeline，这个对理解netty结构有帮助。
到这里，一个NioSocketChannel初始化就完成了。然后我们稍微总结下NioSocketChannel初始化所需要的工作：
- 调用 NioSocketChannel.newSocket(DEFAULT_SELECTOR_PROVIDER) 打开一个新的 Java NIO SocketChannel
- AbstractChannel(Channel parent) 中初始化 AbstractChannel 的属性:
unsafe 通过newUnsafe() 实例化一个 unsafe 对象, 它的类型是 AbstractNioByteChannel.NioByteUnsafe 内部类
pipeline 是 new DefaultChannelPipeline(this) 新创建的实例. 这里体现了:Each channel has its own pipeline and it is created automatically when a new channel is created. - AbstractNioChannel 中的属性:
SelectableChannel ch 被设置为 Java SocketChannel, 即NioSocketChannel#newSocket 返回的 Java NIO SocketChannel.
readInterestOp 被设置为 SelectionKey.OP_READ
SelectableChannel ch 被配置为非阻塞的 ch.configureBlocking(false) -
NioSocketChannel 中的属性:
SocketChannelConfig config = new NioSocketChannelConfig(this, socket.socket())
在上述代码跟踪的过程中，有两个字段是非常重要的，在此我们也来走一遍初始化过程。这两个字段就是：unsafe和pipeline。
unsafe初始化
在 NioSocketChannel初始化的过程中, 会在父类 AbstractChannel 的构造器中, 调用 newUnsafe() 来获取一个 unsafe 实例. 那么 unsafe 是怎么初始化的呢? 它的作用是什么?
interface Unsafe {
RecvByteBufAllocator.Handle recvBufAllocHandle();
SocketAddress localAddress();
SocketAddress remoteAddress();
void register(EventLoop eventLoop, ChannelPromise promise);
void bind(SocketAddress localAddress, ChannelPromise promise);
void connect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise);
void disconnect(ChannelPromise promise);
void close(ChannelPromise promise);
void closeForcibly();
void deregister(ChannelPromise promise);
void beginRead();
void write(Object msg, ChannelPromise promise);
void flush();
ChannelPromise voidPromise();
ChannelOutboundBuffer outboundBuffer();
}
通过方法名我们基本上可以知道unsafe方法都是涉及到java底层的socket操作，在NioSocketChannel中的unsafe即是NioSocketChannelUnsafe：
protected AbstractNioUnsafe newUnsafe() {
return new NioSocketChannelUnsafe();
}
pipieline的初始化
Each channel has its own pipeline and it is created automatically when a new channel is created.实例化一个channel的时候伴随着实例化一个pipeline，即DefaultChannelPipeline。pipeline是netty中非常重要的一个概念，这个后续再详细说明。
protected DefaultChannelPipeline newChannelPipeline() {
return new DefaultChannelPipeline(this);
}
现在看下其初始化的过程，对它有个大概的印象就好。
protected DefaultChannelPipeline(Channel channel) {
this.channel = ObjectUtil.checkNotNull(channel, "channel");
succeededFuture = new SucceededChannelFuture(channel, null);
voidPromise = new VoidChannelPromise(channel, true);
tail = new TailContext(this);
head = new HeadContext(this);
head.next = tail;
tail.prev = head;
}
其中的重点自然是tail和head，双向链表是pipeline的关键，这是一个重点。然后细看下TailContext和HeadContext
TailContext(DefaultChannelPipeline pipeline) {
super(pipeline, null, TAIL_NAME, true, false);
setAddComplete();
}
HeadContext(DefaultChannelPipeline pipeline) {
super(pipeline, null, HEAD_NAME, false, true);
unsafe = pipeline.channel().unsafe();
setAddComplete();
}
其中super具体是AbstractChannelHandlerContext(DefaultChannelPipeline pipeline, EventExecutor executor, String name,boolean inbound, boolean outbound) ，其中的inbound和outbound是很重要的属性，等我们介绍pipeline的时候具体说明。
channel的注册过程
之前的分析我们看到channel的初始化是在Bootstrap.initAndRegister进行的，初始化之后进行了channel的注册过程：
ChannelFuture regFuture = config().group().register(channel);
让我们来跟踪下调用链，config().group()其实得到的是
AbstractBootstrap.initAndRegister -> MultithreadEventLoopGroup.register -> SingleThreadEventLoop.register -> AbstractUnsafe.register
关键代码：
register0(promise)->doRegister()
protected void doRegister() throws Exception {
boolean selected = false;
for (;;) {
try {
selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
return;
} catch (CancelledKeyException e) {
if (!selected) {
// Force the Selector to select now as the "canceled" SelectionKey may still be
// cached and not removed because no Select.select(..) operation was called yet.
eventLoop().selectNow();
selected = true;
} else {
// We forced a select operation on the selector before but the SelectionKey is still cached
// for whatever reason. JDK bug ?
throw e;
}
}
}
}
javaChannel()返回的就是Java NIO SocketChannel，SocketChannel 注册到与 eventLoop 关联的 selector 上了。
整个注册过程就不再总结了，跟着代码走一遍基本上就清楚了，Channel 注册过程所做的工作就是将 Channel 与对应的 EventLoop 关联, 因此这也体现了, 在 Netty 中, 每个 Channel 都会关联一个特定的 EventLoop, 并且这个 Channel 中的所有 IO 操作都是在这个 EventLoop 中执行的; 当关联好 Channel 和 EventLoop 后, 会继续调用底层的 Java NIO SocketChannel 的 register 方法, 将底层的 Java NIO SocketChannel 注册到指定的 selector 中.
handler的添加过程
可以把netty分为两部分，一个是启动，一个是业务处理，之前所说的模块可以归于启动部分，还有一部分是主要给用户使用的，就是hander部分，业务处理模块，用户可以根据自身的需要加handler来处理业务。类似于过滤器或者拦截器，非常灵活。具体实现过程在后续的pipeline中说明。
言归正传，看代码：
.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
if (sslCtx != null) {
p.addLast(sslCtx.newHandler(ch.alloc(), HOST, PORT));
}
//p.addLast(new LoggingHandler(LogLevel.INFO));
p.addLast(new EchoClientHandler());
}
});
从代码里可以看到，入参是实现了ChannelInitializer的匿名类，该类中的抽象方法就是initChannel，在这个方法中添加了用户自定义的handler类来实现业务内容。关键就是initChannel方法在哪里被调用，答案是handlerAdded()和ChannelInitializer.channelRegistered 方法中，其实是先会调用handlerAdded()中的初始化方法，至于为什么要多次调用，怀疑是担心用户重写会覆盖handlerAdded()方法，那么channelRegistered还有一次调用机会。至于handlerAdded()和channelRegistered()又是在哪里被调用的呢，pipeline部分将为您揭开答案。
public final void channelRegistered(ChannelHandlerContext ctx) throws Exception {
if (initChannel(ctx)) {
ctx.pipeline().fireChannelRegistered();
} else {
ctx.fireChannelRegistered();
}
}
private boolean initChannel(ChannelHandlerContext ctx) throws Exception {
if (initMap.putIfAbsent(ctx, Boolean.TRUE) == null) { // Guard against re-entrance.
try {
initChannel((C) ctx.channel());
} catch (Throwable cause) {
// Explicitly call exceptionCaught(...) as we removed the handler before calling initChannel(...).
// We do so to prevent multiple calls to initChannel(...).
exceptionCaught(ctx, cause);
} finally {
remove(ctx);
}
return true;
}
return false;
}
可以看到，在 channelRegistered 方法中, 会调用 initChannel方法, 将自定义的 handler 添加到 ChannelPipeline 中, 然后调用 ctx.pipeline().remove(this) 将自己从 ChannelPipeline 中删除.
步骤一，ChannelPipeline 中只有三个 ChannelInitializer, head, tail。ChannelInitializer是在init方法中加的：
oid init(Channel channel) throws Exception {
ChannelPipeline p = channel.pipeline();
p.addLast(config.handler());
}
步骤二，新增自定义的handler
步骤三，删除ChannelInitializer。
上述基本上是自定义的handler的添加过程，当然没说的那么简单，具体过程在pipeline中说明。
连接
基本工作都做完了以后，我们来看下netty客户端是怎么进行连接的。让我们来跟踪下代码：
Bootstrap.connect()-->Bootstrap.doResolveAndConnect()-->Bootstrap.doResolveAndConnect0()-->Bootstrap.doConnect()-->AbstractChannel.connect()-->DefaultChannelPipeline.connect()-->AbstractChannelHandlerContext.invokeConnect()-->AbstractNioUnsafe.connect()-->NioSocketChannel.doConnect()-->SocketUtils.connect()
最后在SocketUtils.connect()中是通过socketChannel.connect(remoteAddress)完成了连接。这里简述下我认为的几个关键节点：
一、Bootstrap.doConnect()
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(remoteAddress, connectPromise);
} else {
channel.connect(remoteAddress, localAddress, connectPromise);
}
connectPromise.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
}
});
}
上述代码中channel.eventLoop().execute其中是SingleThreadEventExecutor.execute()，可以看到是新增一个任务。我们继续看核心代码channel.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;
}
final AbstractChannelHandlerContext next = findContextOutbound();
EventExecutor executor = next.executor();
if (executor.inEventLoop()) {
next.invokeConnect(remoteAddress, localAddress, promise);
} else {
safeExecute(executor, new Runnable() {
@Override
public void run() {
next.invokeConnect(remoteAddress, localAddress, promise);
}
}, promise, null);
}
return promise;
}
findContextOutbound()函数内容如下：
private AbstractChannelHandlerContext findContextOutbound() {
AbstractChannelHandlerContext ctx = this;
do {
ctx = ctx.prev;
} while (!ctx.outbound);
return ctx;
}
其中的this表示的是TailContext，然后沿着链表往前查找，找到第一个outbound为true的AbstractChannelHandlerContext, 然后调用它的 invokeConnect 方法，如果invokeConnect中handler不是HeadContext，则会继续调用connect直到HeadContext，最后调用HeadContext.connect()，
public void connect(
ChannelHandlerContext ctx,
SocketAddress remoteAddress, SocketAddress localAddress,
ChannelPromise promise) throws Exception {
unsafe.connect(remoteAddress, localAddress, promise);
}
然后继续往下走即可，可以看到最终关键代码：
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 {
return socketChannel.connect(remoteAddress);
}
});
} catch (PrivilegedActionException e) {
throw (IOException) e.getCause();
}
}
以上即使客户端的初始化和连接过程。
