31.Netty源码之客户端启动流程
客户端启动主要流程
如果看了服务器端的启动流程,这里简单看下就可以了。
package io.netty.server;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
public final class EchoClient {
public static void main(String[] args) throws Exception {
// Configure the client.
EventLoopGroup group = new NioEventLoopGroup();
ChannelInitializer<SocketChannel> channelInitializer = new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
// p.addLast(new LoggingHandler(LogLevel.INFO));
p.addLast(new EchoClientHandler());
}
};
try {
Bootstrap b = new Bootstrap();
b.group(group)
.channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.handler(channelInitializer);
// Start the client.
ChannelFuture f = b.connect("127.0.0.1", 8090).sync();
// Wait until the connection is closed.
f.channel().closeFuture().sync();
} finally {
// Shut down the event loop to terminate all threads.
group.shutdownGracefully();
}
}
}
创建客户端NioSocketChannel
1.创建NioSocketChannel
首先看下创建 Channel 的过程,直接跟进 channelFactory.newChannel() 的源码。
public class ReflectiveChannelFactory<T extends Channel> implements ChannelFactory<T> {
private final Constructor<? extends T> constructor;
public ReflectiveChannelFactory(Class<? extends T> clazz) {
ObjectUtil.checkNotNull(clazz, "clazz");
try {
//这里通过泛型反射+工厂 获取无参构造方法
//传进来的clazz是NioSocketChannel.class
this.constructor = clazz.getConstructor();
} catch (NoSuchMethodException e) {
throw new IllegalArgumentException("Class " + StringUtil.simpleClassName(clazz) +
" does not have a public non-arg constructor", e);
}
}
@Override
public T newChannel() {
try {
// 反射创建对象
return constructor.newInstance();
} catch (Throwable t) {
throw new ChannelException("Unable to create Channel from class " + constructor.getDeclaringClass(), t);
}
}
// 省略其他代码
}
在前面 EchoServer的示例中,我们通过 channel(NioSocketChannel.class) 配置 Channel 的类型,工厂类 ReflectiveChannelFactory 是在该过程中被创建的。
从 constructor.newInstance() 我们可以看出,ReflectiveChannelFactory 通过反射创建出 NioSocketChannel 对象,所以我们重点需要关注 NioSocketChannel 的构造函数。
//private static final SelectorProvider
//DEFAULT_SELECTOR_PROVIDER = SelectorProvider.provider();
//SelectorProvider.provider()
//1.读取配置根据配置的class获取provider 獲取不到到第二步
//2.通过spi获取provider 获取不到到第三步
//3.DefaultSelectorProvider#create创建provider
//根据不同的系统创建不同的Selector 或者是说jdk不同
//Linux 下JOK 的下载和安装与Windows 下并没有太大的不同,只是对一些环境的设置稍有不同。
//在windows环境下的是 WindowsSelectorProvider
public NioSocketChannel() {
//DEFAULT_SELECTOR_PROVIDER: 根据不同的系统返回不同的SelectorProvider
this(DEFAULT_SELECTOR_PROVIDER);
}
public NioSocketChannel(SelectorProvider provider) {
// 很熟悉啊,newSocket(DEFAULT_SELECTOR_PROVIDER)是创建 JDK 底层的 SocketChannel
this(newSocket(provider));
}
//根据不同的 SelectorProvider 创建不同的JDK 底层的 SocketChannel
private static SocketChannel newSocket(SelectorProvider provider) {
try {
// 创建 JDK 底层的 SocketChannel 实现类是SocketChannelImpl
return provider.openSocketChannel();
} catch (IOException e) {
throw new ChannelException("Failed to open a socket.", e);
}
}
public NioSocketChannel(Channel parent, SocketChannel socket) {
super(parent, socket);
config = new NioSocketChannelConfig(this, socket.socket());
}
protected AbstractNioByteChannel(Channel parent, SelectableChannel ch) {
//SelectionKey.OP_READread=1 事件
super(parent, ch, SelectionKey.OP_READ);
}
protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
super(parent);
this.ch = ch;
//这里不是注册 SelectionKey.OP_READ=1 事件
//只是赋值
this.readInterestOp = readInterestOp;
try {
//非阻塞模式
ch.configureBlocking(false);
} catch (IOException e) {
try {
ch.close();
} catch (IOException e2) {
logger.warn(
"Failed to close a partially initialized socket.", e2);
}
throw new ChannelException("Failed to enter non-blocking mode.", e);
}
}
protected AbstractChannel(Channel parent) {
this.parent = parent;
id = newId();
unsafe = newUnsafe();
pipeline = newChannelPipeline();
}
SelectorProvider 是 JDK NIO 中的抽象类实现,通过 openServerSocketChannel() 方法可以用于创建服务端的 ServerSocketChannel。而且 SelectorProvider 会根据操作系统类型和版本的不同,返回不同的实现类,具体可以参考 DefaultSelectorProvider 的源码实现:
public static SelectorProvider create() {
String osname = AccessController
.doPrivileged(new GetPropertyAction("os.name"));
if (osname.equals("SunOS"))
return createProvider("sun.nio.ch.DevPollSelectorProvider");
if (osname.equals("Linux"))
return createProvider("sun.nio.ch.EPollSelectorProvider");
//默认返回的是Poll
return new sun.nio.ch.PollSelectorProvider();
}
在这里我们只讨论 Linux 操作系统的场景,在 Linux 内核 2.6版本及以上都会默认采用 EPollSelectorProvider。如果是旧版本则使用 PollSelectorProvider。对于目前的主流 Linux 平台而言,都是采用 Epoll 机制实现的。
创建完 ServerSocketChannel,我们回到 NioServerSocketChannel 的构造函数,接着它会通过 super() 依次调用到父类的构造进行初始化工作,最终我们可以定位到 AbstractNioChannel 和 AbstractChannel 的构造函数:
protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
super(parent);
// 省略其他代码
//设置为16
this.readInterestOp = readInterestOp;
try {
ch.configureBlocking(false);
} catch (IOException e) {
// 省略其他代码
}
}
protected AbstractChannel(Channel parent) {
this.parent = parent;
// Channel 全局唯一 id
id = newId();
// unsafe 操作底层读写
unsafe = newUnsafe();
// pipeline 负责业务处理器编排
// 会初始化TailContext和HeadContext
pipeline = newChannelPipeline();
}
2.设置pipeline
首先调用 AbstractChannel 的构造函数创建了三个重要的成员变量,分别为 id、unsafe、pipeline。
id 表示全局唯一的 Channel,
unsafe 用于操作底层数据的读写操作,
pipeline 负责业务处理器的编排。
3.设置非阻塞模式
初始化状态,pipeline 的内部结构只包含头尾两个节点,如下图所示。三个核心成员变量创建好之后,会回到 AbstractNioChannel 的构造函数,通过 ch.configureBlocking(false) 设置 Channel 是非阻塞模式。
创建服务端 Channel 的过程我们已经讲完了,简单总结下其中几个重要的步骤:
ReflectiveChannelFactory 通过反射创建 NioSocketChannel 实例;
创建 JDK 底层的SocketChannel;包装为NioSocketChannel
为 Channel 创建 id、unsafe、pipeline 三个重要的成员变量;
设置 Channel 为非阻塞模式。
将底层的SocketChannel包装为 NioSocketChannel。
初始化Channel
回到 ServerBootstrap 的 initAndRegister() 方法,继续跟进用于初始化服务端 Channel 的 init() 方法源码:
@Override
@SuppressWarnings("unchecked")
void init(Channel channel) {
//获取pipeline
ChannelPipeline p = channel.pipeline();
//添加客户端的handler方法指定的处理器到pipeline
p.addLast(config.handler());
//设置选项
setChannelOptions
(channel, options0().entrySet().toArray(newOptionArray(0)), logger);
//设置属性
setAttributes(channel, attrs0().entrySet().toArray(newAttrArray(0)));
}
init() 方法的源码比较长,我们依然拆解成两个部分来看:
1.添加客户端handler方法的处理器到pipeline
添加客户端的handler方法指定的处理器到pipeline
2.设置OPTION参数
设置 Socket 参数以及用户自定义属性。在创建客户端 Channel 时,Channel 的配置参数保存在 NioSocketChannelConfig 中,在初始化 Channel 的过程中,Netty 会将这些参数设置到 JDK 底层的 Socket 上,并把用户自定义的属性绑定在 Channel 上。
注册客户端 Channel
回到 initAndRegister() 的主流程,创建完客户端 Channel 之后,继续一层层跟进 register() 方法的源码:
@Override
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
if (eventLoop == null) {
throw new NullPointerException("eventLoop");
}
if (isRegistered()) {
promise.setFailure
(new IllegalStateException("registered to an event loop already"));
return;
}
if (!isCompatible(eventLoop)) {
promise.setFailure(
new IllegalStateException
("incompatible event loop type: " +
eventLoop.getClass().getName()));
return;
}
AbstractChannel.this.eventLoop = eventLoop;
if (eventLoop.inEventLoop()) {
register0(promise);
} else {
try {
eventLoop.execute(new Runnable() {
@Override
public void run() {
register0(promise);
}
});
} catch (Throwable t) {
closeForcibly();
closeFuture.setClosed();
safeSetFailure(promise, t);
}
}
}
Netty 会在线程池 EventLoopGroup 中选择一个 EventLoop 与当前 Channel 进行绑定,之后 Channel 生命周期内的所有 I/O 事件都由这个 EventLoop 负责处理,如 accept、connect、read、write 等 I/O 事件。
可以看出,不管是 EventLoop 线程本身调用,还是外部线程用,最终都会通过 register0() 方法进行注册:
private void register0(ChannelPromise promise) {
try {
if (!promise.setUncancellable() || !ensureOpen(promise)) {
return;
}
boolean firstRegistration = neverRegistered;
// 1.调用 JDK 底层的 register() 进行注册
doRegister();
neverRegistered = false;
registered = true;
// 2.触发 handlerAdded 事件 底层调用了callHandlerAdded0
pipeline.invokeHandlerAddedIfNeeded();
safeSetSuccess(promise);
//3.触发 channelRegistered 事件
pipeline.fireChannelRegistered();
//此时 Channel 还未注册绑定地址,所以处于非活跃状态
//socket的注册不会走进下面if
//socket接受连接创建的socket可以走进去。因为accept后就active了。
if (isActive()) {
//firstRegistration
if (firstRegistration) {
// Channel 当前状态为活跃时,触发 channelActive 事件
pipeline.fireChannelActive();
} else if (config().isAutoRead()) {
//开始读
beginRead();
}
}
} catch (Throwable t) {
// Close the channel directly to avoid FD leak.
closeForcibly();
closeFuture.setClosed();
safeSetFailure(promise, t);
}
}
register0() 主要做了四件事:
1.调用 JDK 底层进行 Channel 注册、
2.触发 handlerAdded 事件、
3.触发 channelRegistered 事件、
4.Channel 当前状态为活跃时,触发 channelActive 事件。
1.注册Channel 绑定选择器和注册事件0
为什么注册0?因为还没初始化完成
我们对它们逐一进行分析。
首先看下 JDK 底层注册 Channel 的过程,对应 doRegister() 方法的实现逻辑。
@Override
protected void doRegister() throws Exception {
boolean selected = false;
for (;;) {
try {
logger.info("initial register: " + 0);
// 调用 JDK 底层的 register() 进行注册
// eventLoop().unwrappedSelector()指的是未包装的selector
// 包装的selector指的是 selectKey
// 注意这里注册的事件是 0 是 0 是 0
// 注意这里注册的事件是 0 是 0 是 0
// 注意这里注册的事件是 0 是 0 是 0
// this = NioServerSocketChannel
selectionKey = javaChannel()
.register(eventLoop()
.unwrappedSelector(), 0, this);
return;
} catch (CancelledKeyException e) {
if (!selected) {
eventLoop().selectNow();
selected = true;
} else {
throw e;
}
}
}
}
public final SelectionKey register(Selector sel, int ops,
Object att)throws ClosedChannelException{
synchronized (regLock) {
// 省略其他代码
SelectionKey k = findKey(sel);
if (k != null) {
k.interestOps(ops);
//att = NioSocketChannel
k.attach(att);
}
if (k == null) {
synchronized (keyLock) {
if (!isOpen())
throw new ClosedChannelException();
k = ((AbstractSelector)sel).register(this, ops, att);
addKey(k);
}
}
return k;
}
}
javaChannel().register() 负责调用 JDK 底层,将 Channel 注册到 Selector 上,register() 的第三个入参传入的是 Netty 自己实现的 NioSocketChannel 对象,调用 register() 方法会将NioSocketChannel 绑定在 JDK 底层 Channel 的 attachment 上。
这样在每次 Selector 对象进行事件循环时,Netty 都可以从返回的 JDK 底层 Channel 中获得自己的 Channel 对象。
2.触发handlerAdded 事件
完成 Channel 向 Selector 注册后,接下来就会触发 Pipeline 一系列的事件传播。在事件传播之前,用户自定义的业务处理器是如何被添加到 Pipeline 中的呢?
答案就在pipeline.invokeHandlerAddedIfNeeded() 当中,我们重点看下 handlerAdded 事件的处理过程。invokeHandlerAddedIfNeeded() 方法的调用层次比较深,推荐你结合上述 Echo 服务端示例,使用 IDE Debug 的方式跟踪调用栈,如下图所示。
final void invokeHandlerAddedIfNeeded() {
assert channel.eventLoop().inEventLoop();
if (firstRegistration) {
firstRegistration = false;
// We are now registered to the EventLoop. It's time to call the callbacks for the ChannelHandlers,
// that were added before the registration was done.
callHandlerAddedForAllHandlers();
}
}
private void callHandlerAddedForAllHandlers() {
final PendingHandlerCallback pendingHandlerCallbackHead;
synchronized (this) {
assert !registered;
// This Channel itself was registered.
registered = true;
pendingHandlerCallbackHead = this.pendingHandlerCallbackHead;
// Null out so it can be GC'ed.
this.pendingHandlerCallbackHead = null;
}
PendingHandlerCallback task = pendingHandlerCallbackHead;
while (task != null) {
//task是PendingHandlerAddedTask
//进入PendingHandlerAddedTask的execute方法
task.execute();
task = task.next;
}
}
@Override
void execute() {
EventExecutor executor = ctx.executor();
if (executor.inEventLoop()) {
//调用callHandlerAdded0
callHandlerAdded0(ctx);
} else {
try {
executor.execute(this);
} catch (RejectedExecutionException e) {
remove0(ctx);
ctx.setRemoved();
}
}
}
}
private void callHandlerAdded0(final AbstractChannelHandlerContext ctx) {
try {
ctx.callHandlerAdded();
} catch (Throwable t) {
boolean removed = false;
try {
remove0(ctx);
ctx.callHandlerRemoved();
removed = true;
} catch (Throwable t2) {
if (logger.isWarnEnabled()) {
logger.warn("Failed to remove a handler: " + ctx.name(), t2);
}
}
if (removed) {
fireExceptionCaught(new ChannelPipelineException(
ctx.handler().getClass().getName() +
".handlerAdded() has thrown an exception; removed.", t));
} else {
fireExceptionCaught(new ChannelPipelineException(
ctx.handler().getClass().getName() +
".handlerAdded() has thrown an exception; also failed to remove.", t));
}
}
}
final void callHandlerAdded() throws Exception {
if (setAddComplete()) {
//ChannelInitializer继承自ChannelHandlerAdapter
//此处调用的handleradded方法是ChannelHandlerAdapter#handlerAdded
//其实就是触发添加处理器事件
handler().handlerAdded(this);
}
}
我们在客户端中指定的ChannelInitializer也是1个ChannelInitializer重写了initChannel。
看到这里恍然大悟,这不就是他妈的模板模式吗?!
我们首先抓住 ChannelInitializer 中的handlerAdded核心源码,逐层进行分析。
// ChannelInitializer
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
if (ctx.channel().isRegistered()) {
//调用初始化方法
//调用的其实就是我们在客户端指定的handler方法中返回的处理器
if (initChannel(ctx)) {
//移除我们在客户端指定的handler方法中返回的处理器
removeState(ctx);
}
}
}
其中有一个点不要混淆,handler() 方法中的handler是添加到客户端的Pipeline 上
完成 这一步之后,handler() 方法中的ChannelInitializer的initChannel已经被调用,添加处理器到客户端的Pipeline 上。
3.监听Read事件
具体流程
1.在nio的run方法中processSelectedKeys();
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
if (!k.isValid()) {
final EventLoop eventLoop;
try {
eventLoop = ch.eventLoop();
} catch (Throwable ignored) {
return;
}
if (eventLoop != this || eventLoop == null) {
return;
}
// close the channel if the key is not valid anymore
unsafe.close(unsafe.voidPromise());
return;
}
try {
// k.readyOps() = 8
int readyOps = k.readyOps();
//SelectionKey.OP_CONNECT=8
//2个都是8进入判断
if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
//当前的注册事件是0
int ops = k.interestOps();
ops &= ~SelectionKey.OP_CONNECT;
k.interestOps(ops);
//完成连接
unsafe.finishConnect();
}
if ((readyOps & SelectionKey.OP_WRITE) != 0) {
ch.unsafe().forceFlush();
}
//处理读请求(断开连接)或接入连接
if ((readyOps & (SelectionKey.OP_READ
| SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
unsafe.read();
}
} catch (CancelledKeyException ignored) {
unsafe.close(unsafe.voidPromise());
}
}
@Override
protected void doBeginRead() throws Exception {
// Channel.read() or ChannelHandlerContext.read() was called
final SelectionKey selectionKey = this.selectionKey;
if (!selectionKey.isValid()) {
return;
}
readPending = true;
final int interestOps = selectionKey.interestOps();
//假设之前没有监听readInterestOp,则监听readInterestOp
if ((interestOps & readInterestOp) == 0) {
//NioServerSocketChannel: readInterestOp = 1
logger.info("interest ops: " + readInterestOp);
selectionKey.interestOps(interestOps | readInterestOp);
}
}
整个服务端 Channel 注册的流程我们已经讲完,注册过程中 Pipeline 结构的变化值得你再反复梳理,从而加深理解。目前服务端还是不能工作的,还差最后一步就是进行端口绑定,我们继续向下分析。
端口绑定
回到 ServerBootstrap 的 bind() 方法,我们继续跟进端口绑定 doBind0() 的源码。
public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
assertEventLoop();
// 省略其他代码
boolean wasActive = isActive();
try {
// 调用 JDK 底层进行端口绑定
doBind(localAddress);
} catch (Throwable t) {
safeSetFailure(promise, t);
closeIfClosed();
return;
}
if (!wasActive && isActive()) {
invokeLater(new Runnable() {
@Override
public void run() {
// 触发 channelActive 给ServerSocketChannel注册
// SelectionKey.OP_ACCEPT事件
// 所有事件的触发都是通过pipeline
pipeline.fireChannelActive();
}
});
}
safeSetSuccess(promise);
}
bind() 方法主要做了两件事,分别为调用 JDK 底层进行端口绑定;绑定成功后并触发 channelActive 事件。下面我们逐一进行分析。
首先看下调用 JDK 底层进行端口绑定的 doBind() 方法:
protected void doBind(SocketAddress localAddress) throws Exception {
if (PlatformDependent.javaVersion() >= 7) {
javaChannel().bind(localAddress, config.getBacklog());
} else {
javaChannel().socket().bind(localAddress, config.getBacklog());
}
}
Netty 会根据 JDK 版本的不同,分别调用 JDK 底层不同的 bind() 方法。我使用的是 JDK8,所以会调用 JDK 原生 Channel 的 bind() 方法。执行完 doBind() 之后,服务端 JDK 原生的 Channel 真正已经完成端口绑定了。
完成端口绑定之后,Channel 处于活跃 Active 状态,然后会调用 pipeline.fireChannelActive() 方法触发 channelActive 事件。 即Channel 处于就绪状态,可以被读写。
我们可以一层层跟进 fireChannelActive() 方法,发现其中比较重要的部分:
// DefaultChannelPipeline#channelActive
public void channelActive(ChannelHandlerContext ctx) {
ctx.fireChannelActive();
readIfIsAutoRead();
}
// AbstractNioChannel#doBeginRead
protected void doBeginRead() throws Exception {
// Channel.read() or ChannelHandlerContext.read() was called
final SelectionKey selectionKey = this.selectionKey;
if (!selectionKey.isValid()) {
return;
}
readPending = true;
final int interestOps = selectionKey.interestOps();
if ((interestOps & readInterestOp) == 0) {
// 注册 OP_ACCEPT 事件到服务端 Channel 的事件集合
selectionKey.interestOps(interestOps | readInterestOp);
}
}
可以看出,在执行完 channelActive 事件传播之后,会调用 readIfIsAutoRead() 方法触发 Channel 的 read 事件,而它最终调用到 AbstractNioChannel 中的 doBeginRead() 方法,其中 readInterestOp 参数就是在前面初始化 Channel 所传入的 SelectionKey.OP_ACCEPT 事件,所以 OP_ACCEPT 事件会被注册到 Channel 的事件集合中。
到此为止,整个服务端已经真正启动完毕。我们总结一下服务端启动的全流程,如下图所示。
创建服务端 Channel:本质是创建 JDK 底层原生的 Channel,并初始化几个重要的属性,包括 id、unsafe、pipeline 等。
初始化服务端 Channel:设置 Socket 参数以及用户自定义属性,并添加1个特殊的处理器 ChannelInitializer,ChannelInitializer的功能是添加 LoggingHandler 和 ServerBootstrapAcceptor,但是并没有添加进去。
注册服务端 Channel:调用 JDK 底层将 Channel 注册到 Selector上。执行ChannelInitializer的initChannel真正添加handler
端口绑定:调用 JDK 底层进行端口绑定,并触发 channelActive 事件,把 OP_ACCEPT 事件注册到NioServerSocketChannel 的事件集合中。
1.添加handler方法中的指定handlerA到pipeline
2.执行pipeline中handlerA
3.将handlerA中 添加的hanlders添加到pipeline
4.移除handler方法中的指定handlerA
5.和服务器建立连接
5.执行hanlders向服务器发送数据
6.执行hanlders接受服务器数据
查看28道真题和解析