31.Netty源码之客户端启动流程-Toy模板网

这篇具有很好参考价值的文章主要介绍了31.Netty源码之客户端启动流程。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

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 底层的 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.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 是非阻塞模式。

31.Netty源码之客户端启动流程,.net

创建服务端 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() {
                    @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() 方法进行注册：

```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 事件 底层调用了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() 方法的实现逻辑。

```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();

```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 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());
    }
}

```

```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，则监听readInterestOp
    if ((interestOps & readInterestOp) == 0) {
        //NioServerSocketChannel: readInterestOp = 1
        logger.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 的事件集合中。

到此为止，整个服务端已经真正启动完毕。我们总结一下服务端启动的全流程，如下图所示。

31.Netty源码之客户端启动流程,.net