highlight: arduino-light
客户端启动主要流程
如果看了服务器端的启动流程,这里简单看下就可以了。
java 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() 的源码。
java 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 的构造函数。
```java //private static final SelectorProvider //DEFAULTSELECTORPROVIDER = SelectorProvider.provider();
//SelectorProvider.provider() //1.读取配置根据配置的class获取provider 獲取不到到第二步 //2.通过spi获取provider 获取不到到第三步 //3.DefaultSelectorProvider#create创建provider //根据不同的系统创建不同的Selector 或者是说jdk不同 //Linux 下JOK 的下载和安装与Windows 下并没有太大的不同,只是对一些环境的设置稍有不同。 //在windows环境下的是 WindowsSelectorProvider public NioSocketChannel() { //DEFAULTSELECTORPROVIDER: 根据不同的系统返回不同的SelectorProvider this(DEFAULTSELECTORPROVIDER); }
public NioSocketChannel(SelectorProvider provider) {
// 很熟悉啊,newSocket(DEFAULT_SELECTOR_PROVIDER)是创建 JDK 底层的 SocketChannelthis(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.OPREADread=1 事件 super(parent, ch, SelectionKey.OPREAD); }
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 的源码实现:
java 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 的构造函数:
java 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 的过程我们已经讲完了,简单总结下其中几个重要的步骤:
java 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() 方法的源码:
```java @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() {@Overridepublic 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() 方法进行注册:
```java 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 事件 底层调用了callHandlerAdded0pipeline.invokeHandlerAddedIfNeeded();safeSetSuccess(promise);//3.触发 channelRegistered 事件pipeline.fireChannelRegistered();//此时 Channel 还未注册绑定地址,所以处于非活跃状态//socket的注册不会走进下面if//socket接受连接创建的socket可以走进去。因为accept后就active了。if (isActive()) {//firstRegistrationif (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() 方法的实现逻辑。
```java @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 的方式跟踪调用栈,如下图所示。
java 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(); } }
```java 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;}
}
```
java @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(); } } } }
```java 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));}}
}
```
```java final void callHandlerAdded() throws Exception {
if (setAddComplete()) {//ChannelInitializer继承自ChannelHandlerAdapter//此处调用的handleradded方法是ChannelHandlerAdapter#handlerAdded//其实就是触发添加处理器事件handler().handlerAdded(this);}
}
```
我们在客户端中指定的ChannelInitializer也是1个ChannelInitializer重写了initChannel。
看到这里恍然大悟,这不就是他妈的模板模式吗?!
我们首先抓住 ChannelInitializer 中的handlerAdded核心源码,逐层进行分析。
java // 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();
2.
```java 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 anymoreunsafe.close(unsafe.voidPromise());return;}try {// k.readyOps() = 8int readyOps = k.readyOps();//SelectionKey.OP_CONNECT=8//2个都是8进入判断if ((readyOps & SelectionKey.OP_CONNECT) != 0) {//当前的注册事件是0int 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());}
}
```
```java @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,则监听readInterestOpif ((interestOps & readInterestOp) == 0) {//NioServerSocketChannel: readInterestOp = 1logger.info("interest ops: " + readInterestOp);selectionKey.interestOps(interestOps | readInterestOp);}
}
```
整个服务端 Channel 注册的流程我们已经讲完,注册过程中 Pipeline 结构的变化值得你再反复梳理,从而加深理解。目前服务端还是不能工作的,还差最后一步就是进行端口绑定,我们继续向下分析。
端口绑定
回到 ServerBootstrap 的 bind() 方法,我们继续跟进端口绑定 doBind0() 的源码。
java 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() 方法:
java 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() 方法,发现其中比较重要的部分:
java // 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.OPACCEPT 事件,所以 OPACCEPT 事件会被注册到 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接受服务器数据