---

本文均为原创，如需转载请注明出处。

[TOC]

Netty服务端创建流程分析

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

Netty服务端时序图

Netty服务端--Channel的创建

首先基于NIO的学习，思考两个问题

• 服务端的Socket在哪里初始化？

• 在哪里accept连接？

研究服务端是如何创建的，查看源码，首先应该从源头出发，直接调用的方法开始层层深入。

Netty服务端的入口bind（）方法

​ 服务端创建的入口bind（）方法

// 绑定端口，同步等待成功
ChannelFuture f = b.bind(port).sync();

进入bind（）方法发现会执行一个dobind（）方法

    private ChannelFuture doBind(final SocketAddress localAddress) {
        final ChannelFuture regFuture = initAndRegister();
        final Channel channel = regFuture.channel();
        if (regFuture.cause() != null) {
            return regFuture;
        }

该方法调用了一个initAndRegister（）

 final ChannelFuture initAndRegister() {
        Channel channel = null;
        try {
            channel = channelFactory.newChannel();
            init(channel);
        } catch (Throwable t) {
            if (channel != null) {
                // channel can be null if newChannel crashed (eg SocketException("too many open files"))
                channel.unsafe().closeForcibly();
            }
            // as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor
            return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t);
        }

很明显可以看出。这里会创建一个底层的channel，调用Netty维护的一个工厂类，下面的init（）方法是一个抽象类，可以看看那些方法继承并实现了它呢？😊

AbstractBootstrap下的两个子类.png

所以可想而知！在调用bind（）方法后，Netty会调用JDK底层初始化一个Channel，回过头来看看channelFactory.newChannel()的实现。

在ChannelFactory下有一个利用反射的实现子类ReflectiveChannelFactory(),从名字就可以看得出来，是一个利用反射来初始化channel的工厂类。该类下的newChannel（）方法：

    @Override
    public T newChannel() {
        try {
            return clazz.newInstance();
        } catch (Throwable t) {
            throw new ChannelException("Unable to create Channel from class " + clazz, t);
        }
    }

确实是利用了返回来进行实例化的。那么它是如何知道实话化那个对象呢？

​ 让我们回到上一层开始出现channelFactory.newChannel(),在这里可以很清楚的看出来，当前类AbstractBootstrap自身维护着该工厂对象，并且在构造函数中给该工厂类所需要的对象进行了赋值。

    AbstractBootstrap() {
        // Disallow extending from a different package.
    }

    AbstractBootstrap(AbstractBootstrap<B, C> bootstrap) {
        group = bootstrap.group;
        // 那么这里的bootstrap对象又是从哪里得到的呢？
        channelFactory = bootstrap.channelFactory;
        handler = bootstrap.handler;
        localAddress = bootstrap.localAddress;
        synchronized (bootstrap.options) {
            options.putAll(bootstrap.options);
        }
        synchronized (bootstrap.attrs) {
            attrs.putAll(bootstrap.attrs);
        }
    }

这里的AbstractBootstrap<B, C> bootstrap很直接的就可以想到我们在最外层定义的AbstractBootstrap的子类Bootstrap/ServerBootstrap中设置了

          ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .childHandler(new ChildChannelHandler());

其中设置的channel属性就是告诉底层的channelFactory来实例化该对象。

这里我们验证一下，直接从我们的代码中.channel()方法中去，可以很清楚的看出，上面说的调用的是ChannelFactory下的子类ReflectiveChannelFactory

    public B channel(Class<? extends C> channelClass) {
        if (channelClass == null) {
            throw new NullPointerException("channelClass");
        }
        // 调用工厂类，利用反射创建对象
        return channelFactory(new ReflectiveChannelFactory<C>(channelClass));
    }

该类的实现非常的简单：

public class ReflectiveChannelFactory<T extends Channel> implements ChannelFactory<T> {

    private final Class<? extends T> clazz;

    // 直接返回 需要初始化类的 class 对象
    public ReflectiveChannelFactory(Class<? extends T> clazz) {
        if (clazz == null) {
            throw new NullPointerException("clazz");
        }
        this.clazz = clazz;
    }

    // 利用反射 初始化返回实例对象
    @Override
    public T newChannel() {
        try {
            return clazz.newInstance();
        } catch (Throwable t) {
            throw new ChannelException("Unable to create Channel from class " + clazz, t);
        }
    }

    @Override
    public String toString() {
        return StringUtil.simpleClassName(clazz) + ".class";
    }
}

现在明白了channel是由谁创建的，那么到底是怎么创建出来的呢？现在进入NioServerSocketChannel.查看该对象的构造函数，做了那些事情。

通过NioServerSocketChannel加密Channel的创建

    /**
     * Create a new instance
     */
    public NioServerSocketChannel() {
        this(newSocket(DEFAULT_SELECTOR_PROVIDER));
    }
// 该方法 返回的对象为 java.nio.channels;下的。调用的方法也是JDK底层的实现
    private static ServerSocketChannel newSocket(SelectorProvider provider) {
        try {
            return provider.openServerSocketChannel();
        } catch (IOException e) {
            throw new ChannelException(
                    "Failed to open a server socket.", e);
        }
    }
// 原来NioServerSocketChannel 是直接调用的JDK底层的newSocket来来创建Channel 通道

其中还有一个构造

    /**
     * Create a new instance using the given {@link ServerSocketChannel}.
     */
    public NioServerSocketChannel(ServerSocketChannel channel) {
        super(null, channel, SelectionKey.OP_ACCEPT);
        // 这里是可以生成一个NiO channel配置信息
        config = new NioServerSocketChannelConfig(this, javaChannel().socket());
    }

一次点进该方法的父类可以找到在NIO编程中不可避免地一项异步配置：ch.configureBlocking(false);，配置为异步非阻塞。

    protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
        super(parent);
        this.ch = ch;
        this.readInterestOp = readInterestOp;
        try {
            ch.configureBlocking(false);// 配置为异步 非阻塞 （重点）
        } catch (IOException e) {
            try {
                ch.close();
            } catch (IOException e2) {
                if (logger.isWarnEnabled()) {
                    logger.warn(
                            "Failed to close a partially initialized socket.", e2);
                }
            }

            throw new ChannelException("Failed to enter non-blocking mode.", e);
        }
    }

Channel UML图.png

在上面继承关系中AbstractChannel维护着channel通道地内部属性

    /**
     * Creates a new instance.
     *
     * @param parent
     *        the parent of this channel. {@code null} if there's no parent.
     */
    protected AbstractChannel(Channel parent) {
        this.parent = parent;
        id = newId(); // Channel的 唯一标识
        unsafe = newUnsafe();// 对应 底层TCP读写的相关操作
        pipeline = newChannelPipeline();// 后续研究😊
    }

在服务端channel初始化完成之后，下一步就需要将该channel注册到selector上面。

注册selector

​ 在上述代码channel初始化完成的地方为：调用了init（）方法。因此应该从该地方出发查看如何将channel注册到selector上。initAndRegister初始并注册。

    final ChannelFuture initAndRegister() {
        Channel channel = null;
        try {
            // 初始化操作
            channel = channelFactory.newChannel();
            init(channel);
        } catch (Throwable t) {
            if (channel != null) {
                // channel can be null if newChannel crashed (eg SocketException("too many open files"))
                channel.unsafe().closeForcibly();
            }
            // as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor
            return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t);
        }

        // 注册channel
        ChannelFuture regFuture = config().group().register(channel);
        if (regFuture.cause() != null) {
            if (channel.isRegistered()) {
                channel.close();
            } else {
                channel.unsafe().closeForcibly();
            }
        }

其中register的实现在AbstractChannel下。

 @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) {
                    logger.warn(
                            "Force-closing a channel whose registration task was not accepted by an event loop: {}",
                            AbstractChannel.this, t);
                    closeForcibly();
                    closeFuture.setClosed();
                    safeSetFailure(promise, t);
                }
            }
        }

register0(promise);

private void register0(ChannelPromise promise) {
            try {
                // check if the channel is still open as it could be closed in the mean time when the register
                // call was outside of the eventLoop
                if (!promise.setUncancellable() || !ensureOpen(promise)) {
                    return;
                }
                boolean firstRegistration = neverRegistered;
                // 调用Jdk底层注册
                doRegister();
                neverRegistered = false;
                registered = true;

                // Ensure we call handlerAdded(...) before we actually notify the promise. This is needed as the
                // user may already fire events through the pipeline in the ChannelFutureListener.
                // 触发事件
                pipeline.invokeHandlerAddedIfNeeded();

                safeSetSuccess(promise);
                // 传播时间
                pipeline.fireChannelRegistered();
                // Only fire a channelActive if the channel has never been registered. This prevents firing
                // multiple channel actives if the channel is deregistered and re-registered.
                if (isActive()) {
                    if (firstRegistration) {
                        pipeline.fireChannelActive();
                    } else if (config().isAutoRead()) {
                        // This channel was registered before and autoRead() is set. This means we need to begin read
                        // again so that we process inbound data.
                        //
                        // See https://github.com/netty/netty/issues/4805
                        beginRead();
                    }
                }
            } catch (Throwable t) {
                // Close the channel directly to avoid FD leak.
                closeForcibly();
                closeFuture.setClosed();
                safeSetFailure(promise, t);
            }
        }

    @Override
    protected void doRegister() throws Exception {
        boolean selected = false;
        for (;;) {
            try {
                // 调用JDK 方法实现注册
                selectionKey = javaChannel().register(eventLoop().selector, 0, this);
                return;

注册成功后，考虑端口绑定。

端口绑定

还是根据dobind（）方法可以看到里面有个bind0（）方法。channel对象调用bind（）方法，在AbstractChannel（）下有具体实现

@Override
        public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
            assertEventLoop();

            if (!promise.setUncancellable() || !ensureOpen(promise)) {
                return;
            }

            // See: https://github.com/netty/netty/issues/576
            if (Boolean.TRUE.equals(config().getOption(ChannelOption.SO_BROADCAST)) &&
                localAddress instanceof InetSocketAddress &&
                !((InetSocketAddress) localAddress).getAddress().isAnyLocalAddress() &&
                !PlatformDependent.isWindows() && !PlatformDependent.isRoot()) {
                // Warn a user about the fact that a non-root user can't receive a
                // broadcast packet on *nix if the socket is bound on non-wildcard address.
                logger.warn(
                        "A non-root user can't receive a broadcast packet if the socket " +
                        "is not bound to a wildcard address; binding to a non-wildcard " +
                        "address (" + localAddress + ") anyway as requested.");
            }

            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() {
                        // 触发事件
                        pipeline.fireChannelActive();
                    }
                });
            }

            safeSetSuccess(promise);
        }

调用底层的绑定方法

    @Override
    protected void doBind(SocketAddress localAddress) throws Exception {
        if (PlatformDependent.javaVersion() >= 7) {
            // 调用底层JDK的绑定方法
            javaChannel().bind(localAddress, config.getBacklog());
        } else {
            javaChannel().socket().bind(localAddress, config.getBacklog());
        }
    }

总结

​ 服务端启动核心路径总结：

服务端启动核心路径.png

首先调用服务端的newChannel（）创建服务端channel，这个过程实际上就是调用JDK底层的API来创建一个JDK channel，然后Netty将其包装成自己的服务端的channel，同时会创建一些基本的组件绑定在此channel上（比喻：pipeline）。然后调用init（）来初始化服务端channel，这个过程最重要的就是为服务端的channel添加一个连接处理器。随后调用register（）方法注册selector，这个过程Netty将JDK底层生成的channel注册到selector上，最后调用bind（）方法通过jdk底层的API将端口号绑定。来实现，绑定之后，netty会selector绑定一个OP_ACCEPT事件，然后selector就可以接收绑定其他channel了。