Netty源码分析系列（一）

2020-01-02

Netty服务端启动

netty版本为4.1.6.Final

本文可以带着两个问题来看
1.服务端的socket在哪里初始化？
2.在哪里accept连接？

启动过程

1.创建服务端channel
2.初始化服务端channel
3.注册selector
4.端口绑定

创建服务端channel

调用过程

代码调用过程

用户代码入口bind()

try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childOption(ChannelOption.TCP_NODELAY, true)
                    .childAttr(AttributeKey.newInstance("childAttr"), "childAttrValue")
                    .handler(new ServerHandler())
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) {
                            ch.pipeline().addLast(new AuthHandler());
                            //..

                        }
                    });

            ChannelFuture f = b.bind(8888).sync();//入口

bind()方法

    /**
     * Create a new {@link Channel} and bind it.
     */
    public ChannelFuture bind(int inetPort) {
        return bind(new InetSocketAddress(inetPort));
    }

    /**
     * Create a new {@link Channel} and bind it.
     */
    public ChannelFuture bind(SocketAddress localAddress) {
        validate();
        if (localAddress == null) {
            throw new NullPointerException("localAddress");
        }
        return doBind(localAddress);//调用doBind方法
    }

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

channelFactory.newChannel()


    final ChannelFuture initAndRegister() {
        Channel channel = null;
        try {
            channel = channelFactory.newChannel();
            init(channel);
        } catch (Throwable t) {
               ... 
        }
        }

ReflectiveChannelFactory.java里面的channelFactory

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

    private final Class<? extends T> clazz;

    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

newSocket（）

newSocket（）通过底层jdk创建jdk channel

public class NioServerSocketChannel extends AbstractNioMessageChannel
                             implements io.netty.channel.socket.ServerSocketChannel {
     ....                        

     /**
     * Create a new instance
     */
    public NioServerSocketChannel() {
        this(newSocket(DEFAULT_SELECTOR_PROVIDER));//newSocket方法
    }

    .....
    }

newSocket方法直接调用provider

private static ServerSocketChannel newSocket(SelectorProvider provider) {
        try {
            /**
             *  Use the {@link SelectorProvider} to open {@link SocketChannel} and so remove condition in
             *  {@link SelectorProvider#provider()} which is called by each ServerSocketChannel.open() otherwise.
             *
             *  See <a href="https://github.com/netty/netty/issues/2308">#2308</a>.
             */
            return provider.openServerSocketChannel();//这里直接创建一个channel
        } catch (IOException e) {
            throw new ChannelException(
                    "Failed to open a server socket.", e);
        }
    }

进入到openServerSocketChannel,发现其所属的包已经为jdk的包

 package java.nio.channels.spi;
      ....
    /**
     * Opens a server-socket channel.
     *
     * @return  The new channel
     *
     * @throws  IOException
     *          If an I/O error occurs
     */
    public abstract ServerSocketChannel openServerSocketChannel()
        throws IOException;
      ....

NioServerSocketChannelConfig

通过NioServerSocketChannelConfig创建tcp参数配置类

    /**
     * Create a new instance using the given {@link ServerSocketChannel}.
     */
    public NioServerSocketChannel(ServerSocketChannel channel) {
        super(null, channel, SelectionKey.OP_ACCEPT);
        config = new NioServerSocketChannelConfig(this, javaChannel().socket());
    }

AbstractNioChannel

调用父类构造函数AbstractNioChannel

    /**
     * Create a new instance using the given {@link ServerSocketChannel}.
     */
    public NioServerSocketChannel(ServerSocketChannel channel) {
        super(null, channel, SelectionKey.OP_ACCEPT); //super调用父类
        config = new NioServerSocketChannelConfig(this, javaChannel().socket());
    }

跟进去，可以看到AbstractNioMessageChannel继续调用

    /**
     * @see {@link AbstractNioChannel#AbstractNioChannel(Channel, SelectableChannel, int)}
     */
    protected AbstractNioMessageChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
        super(parent, ch, readInterestOp);//super调用父类
    }

跟进去，可以看到AbstractNioChannel


    /**
     * Create a new instance
     *
     * @param parent            the parent {@link Channel} by which this instance was created. May be {@code null}
     * @param ch                the underlying {@link SelectableChannel} on which it operates
     * @param readInterestOp    the ops to set to receive data from the {@link SelectableChannel}
     */
    protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
        super(parent); //再往上还有一个客户端，服务端共同父类AbstractChannel
        this.ch = ch;
        this.readInterestOp = readInterestOp;
        try {
            ch.configureBlocking(false); //这是的ch,就是刚刚通过newsocket创建出来的一个底层的jdkchannel，这里是设置服务端的channel为非阻塞
        } 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);
        }
    }

AbstractChannel

这里为什么是AbstractChannel，因为除了服务端的channel还有客户端的channel。AbstractChannel就是对channel的一个抽象。AbstractChannel有三个重要的属性id，unsafe，pipeline.

id :对应每一个channel的唯一标识
unsafe ：跟tcp相关的一些操作，读写
pipeline ：服务端和客户端相关的一个逻辑链

    /**
     * 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();
        unsafe = newUnsafe();
        pipeline = newChannelPipeline();
    }

初始化服务端channel

过程,初始化服务端channel的过程用一句话概括就是保存用户的自定义属性，然后通过这些属性创建一个连接接入器，连接接入器每次accept一个连接后，会使用这些属性对新的连接进行一些配置。

下面来看一下代码调用过程，首先来到用户代码入口bind()方法，可以看到initAndRegister方法。

    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);     //这里用init初始化channel
        } catch (Throwable t) {
          ...
        }

init()方法，总的来说就是获取用户的配置，并

package io.netty.bootstrap;
    ...
    @Override
    void init(Channel channel) throws Exception {
        final Map<ChannelOption<?>, Object> options = options0();//拿到配置的options
        synchronized (options) {
            channel.config().setOptions(options);//通过config将options配置进去，这里的config是上一节创建服务端channel时也会创建的config
        }

        final Map<AttributeKey<?>, Object> attrs = attrs0();//拿到用户自定义的属性
        synchronized (attrs) {
            for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) {
                @SuppressWarnings("unchecked")
                AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();
                channel.attr(key).set(e.getValue()); //绑定
            }
        }

        ChannelPipeline p = channel.pipeline();

        final EventLoopGroup currentChildGroup = childGroup;
        final ChannelHandler currentChildHandler = childHandler;
        final Entry<ChannelOption<?>, Object>[] currentChildOptions;
        final Entry<AttributeKey<?>, Object>[] currentChildAttrs;
        synchronized (childOptions) { //保存childOptions
            currentChildOptions = childOptions.entrySet().toArray(newOptionArray(childOptions.size()));
        }
        synchronized (childAttrs) {   //保存childAttrs
            currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(childAttrs.size()));
        }

        p.addLast(new ChannelInitializer<Channel>() { //配置服务端pipeline
            @Override
            public void initChannel(Channel ch) throws Exception {
                final ChannelPipeline pipeline = ch.pipeline();
                ChannelHandler handler = config.handler();//拿到用户自定义的pipeline
                if (handler != null) {
                    pipeline.addLast(handler);//把用户自定义的处理链添加进去
                }

                // We add this handler via the EventLoop as the user may have used a ChannelInitializer as handler.
                // In this case the initChannel(...) method will only be called after this method returns. Because
                // of this we need to ensure we add our handler in a delayed fashion so all the users handler are
                // placed in front of the ServerBootstrapAcceptor.
                ch.eventLoop().execute(new Runnable() {
                    @Override
                    public void run() {
                        pipeline.addLast(new ServerBootstrapAcceptor(//添加一个ServerBootstrapAcceptor，把前面配置的几个参数都传进去
                                currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                    }
                });
            }
        });
    }

注册selector

在initAndRegister()方法中有调用register方法注册channel。

    final ChannelFuture initAndRegister() {
        Channel channel = null;
        ....

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

AbstractChannel.register()方法


        @Override
        public final void register(EventLoop eventLoop, final ChannelPromise promise) {
            ...

            AbstractChannel.this.eventLoop = eventLoop;//简单的赋值操作，后续一些io操作都交由eventloop来执行

            if (eventLoop.inEventLoop()) {
                register0(promise); //调用register0实际注册
            } else {
                try {
                    eventLoop.execute(new Runnable() {
                        @Override
                        public void run() {
                            register0(promise); //
                        }
                    });
                } catch (Throwable t) {
                   ....
                }
            }
        }

register0()方法

        private void register0(ChannelPromise promise) {
            try {
                ...
                boolean firstRegistration = neverRegistered;
                doRegister();  //调用底层jdk注册channel
                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(); //HandlerAdded对应着用户代码的执行顺序
                ...
                pipeline.fireChannelRegistered();//这对应着用户代码的执行顺序
                ...
        }

用户的ServerHandler

public class ServerHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        System.out.println("channelActive");
    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) {
        System.out.println("channelRegistered");
    }

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        System.out.println("handlerAdded");
    }
    ...
}

先输出handlerAdded，然后是channelRegistered

doRegister()方法


    @Override
    protected void doRegister() throws Exception {
        boolean selected = false;
        for (;;) {
            try {
                //调用jdk api register注册方法,这个javaChannel是在创建服务端channel的时候会创建一个jdk底层的channel，然后把这个channel进行保存，保存到成员变量ch里面
                //register需要三个参数`selector，ops,att`
                selectionKey = javaChannel().register(eventLoop().selector, 0, this);

                return;
            } catch (CancelledKeyException e) {
               ...
                }
            }
        }
    }

端口绑定

端口绑定过程

还是来看AbstractBootstrap.doBind()方法


    private ChannelFuture doBind(final SocketAddress localAddress) {
        //初始化，注册...

        if (regFuture.isDone()) {
            // At this point we know that the registration was complete and successful.
            ChannelPromise promise = channel.newPromise();
            doBind0(regFuture, channel, localAddress, promise);//绑定端口，调用`AbstractChannel.bind()`方法
            return promise;
        } else {
           ...
                }
            });
            return promise;
        }
    }

AbstractChannel.bind()方法

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

            boolean wasActive = isActive();
            try {
                doBind(localAddress);//jdk底层的绑定
            } catch (Throwable t) {
                safeSetFailure(promise, t);
                closeIfClosed();
                return;
            }

            if (!wasActive && isActive()) { //在端口绑定之前不是active，在绑定之后为active状态
                invokeLater(new Runnable() {
                    @Override
                    public void run() {
                        pipeline.fireChannelActive();//触发ChannelActive事件传播
                    }
                });
            }

            safeSetSuccess(promise);
        }

AbstractChannel.bind()方法里面的doBind(localAddress)会调用NioServerSocketChannel.doBind()方法，直接调用了底层jdk的bind()方法

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

端口绑定完成后，调用pipeline.fireChannelActive(),触发ChannelActive事件pipeline.fireChannelActive()传播

传播链为

pipeline.fireChannelActive()–>DefaultChannelPipeline.ChannelActive()–>DefaultChannelPipeline.readIfIsAutoRead()–>AbstractChannel.read()–>DefaultChannelPipeline.read()–>AbstractChannelHandlerContext.read()–>AbstractChannel.beginRead()–>AbstractNioChannel.doBeginRead()

DefaultChannelPipeline.ChannelActive()

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            ctx.fireChannelActive(); //传播从HeadContext.readIfIsAutoRead出发
            readIfIsAutoRead();
        }

DefaultChannelPipeline.readIfIsAutoRead()

        private void readIfIsAutoRead() {
            if (channel.config().isAutoRead()) {
                channel.read();
            }
        }

AbstractChannel.read()

        @Override
        public Channel read() {
            pipeline.read(); //从pipeline开始传播
            return this;
        }

DefaultChannelPipeline.read()

    @Override
    public final ChannelPipeline read() {
        tail.read(); //tail为pipeline最尾部的一个节点    ·
        return this;
    }

AbstractChannelHandlerContext.read()

    @Override
    public ChannelHandlerContext read() {
        final AbstractChannelHandlerContext next = findContextOutbound();
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) {
            next.invokeRead();
        } else {
            Runnable task = next.invokeReadTask;
            if (task == null) {
                next.invokeReadTask = task = new Runnable() {
                    @Override
                    public void run() {
                        next.invokeRead();
                    }
                };
            }
            executor.execute(task);
        }

        return this;
    }

AbstractChannel.beginRead()

        @Override
        public final void beginRead() {
            assertEventLoop();

            if (!isActive()) {
                return;
            }

            try {
                doBeginRead();
            } catch (final Exception e) {
                invokeLater(new Runnable() {
                    @Override
                    public void run() {
                        pipeline.fireExceptionCaught(e);
                    }
                });
                close(voidPromise());
            }
        }

AbstractNioChannel.doBeginRead()

    @Override
    protected void doBeginRead() throws Exception {
        // Channel.read() or ChannelHandlerContext.read() was called
        final SelectionKey selectionKey = this.selectionKey; //这个selectionKey就是前面注册服务端的channel到selector上时返回的selectionKey
        if (!selectionKey.isValid()) {
            return;
        }

        readPending = true;

        final int interestOps = selectionKey.interestOps();//前面注册的时候ops绑定了一个0上去
        if ((interestOps & readInterestOp) == 0) {
            selectionKey.interestOps(interestOps | readInterestOp);
        }
    }

总结端口绑定：当端口绑定完成后，会触发一个active事件，这个active事件最终会调用到channel的一个read事件，read事件对于服务端来说就是可以读了，可以读新的连接。

启动核心路径总结

首先调用newChannel()创建服务端的channel，这个过程实际上就是调用jdk底层的api来创建一个jdk channel，然后netty将其包装成一个自己服务端的channel，同时会创建一些基本的组件绑定在此channel上，比如pipeline，然后调用init()方法初始化服务端channel，这个过程最重要的是为服务端的channel添加一个连接处理器，随后调用register()方法注册selector，这个过程中netty将jdk底层的channel注册到事件轮询器selector上面，并把netty的服务端channel作为一个attachment绑定到对应的jdk底层的channel，最后调用doBind()方法调用jdk底层的api来实现对本地端口的监听，然后netty会重新向selector注册一个ops accept事件，这样netty就可以接受新的连接了。

|阅读量()