netty源码分析 - accept过程分析

系列

Netty源码分析 - Bootstrap服务端
Netty源码分析 - Bootstrap客户端
netty源码分析 - ChannelHandler
netty源码分析 - EventLoop类关系
netty源码分析 - register分析
Netty源码分析 - NioEventLoop事件处理
netty源码分析 - accept过程分析
Netty源码分析 - ByteBuf
Netty源码分析 - 粘包和拆包问题

开篇

• 这篇文章是分析Netty Server在处理Client端连接过程中如何实现bossGroup的接受请求到workerGroup的注册Channel的过程。
• 在这篇文章中会了解bossGroup如何处理来自client端的连接，并将该连接注册到workerGroup当中。
• 在Netty源码分析 - NioEventLoop事件处理文章中分析了Netty的消息处理机制，连接请求就是消息处理的一个小分支，所以如果想了解完整的消息处理过程建议先阅读NioEventLoop事件处理这篇文章。

NioEventLoop

accept调用链

• NioEventLoop启动消息处理线程的调用链如下：
  SingleThreadEventExecutor#doStartThread => executor#execute => NioEventLoop#run

• NioEventLoop处理消息的调用链如下：
  NioEventLoop#processSelectedKeys => NioEventLoop#processSelectedKeysOptimized => NioEventLoop#processSelectedKey => AbstractNioMessageChannel.NioMessageUnsafe#read

• 下面的分析基于源码层面分析NioEventLoop如何处理accept事件类型。
  

NioEventLoop#processSelectedKey

public final class NioEventLoop extends SingleThreadEventLoop {

    private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
        final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
        // 省略代码
        try {
            int readyOps = k.readyOps();
            if ((readyOps & SelectionKey.OP_CONNECT) != 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) {
                // AbstractNioMessageChannel#NioMessageUnsafe
                unsafe.read();
            }
        } catch (CancelledKeyException ignored) {
            unsafe.close(unsafe.voidPromise());
        }
    }
}

• processSelectedKey内部会各类事件消息，包括OP_ACCEPT事件，就是我们关心的server侧的连接请求消息类型。
• 针对OP_ACCEPT类型事件消息，由unsafe.read()继续处理，这里的unsafe指的是AbstractNioMessageChannel#NioMessageUnsafe。

AbstractNioMessageChannel#NioMessageUnsafe

public abstract class AbstractNioMessageChannel extends AbstractNioChannel {

    private final class NioMessageUnsafe extends AbstractNioUnsafe {

        private final List<Object> readBuf = new ArrayList<Object>();

        @Override
        public void read() {
            assert eventLoop().inEventLoop();
            final ChannelConfig config = config();
            final ChannelPipeline pipeline = pipeline();
            final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
            allocHandle.reset(config);

            boolean closed = false;
            Throwable exception = null;
            try {
                try {
                    do {
                        // 读取连接请求并生成NioSocketChannel放到readBuf当中
                        int localRead = doReadMessages(readBuf);

                        // 省略代码
                        allocHandle.incMessagesRead(localRead);
                    } while (allocHandle.continueReading());
                } catch (Throwable t) {
                    exception = t;
                }

                int size = readBuf.size();
                for (int i = 0; i < size; i ++) {
                    readPending = false;
                    // 触发NioServerSocketChannel的pipeline处理链
                    pipeline.fireChannelRead(readBuf.get(i));
                }
                readBuf.clear();
                allocHandle.readComplete();
                pipeline.fireChannelReadComplete();
                
                // 省略代码
            } finally {
                // 省略代码
            }
        }
    }
}


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

    protected int doReadMessages(List<Object> buf) throws Exception {
        // 负责accept新的SocketChannel对象
        SocketChannel ch = SocketUtils.accept(javaChannel());

        try {
            if (ch != null) {
                // 封装新的SocketChannel对象为NioSocketChannel并添加到buf当中
                buf.add(new NioSocketChannel(this, ch));
                return 1;
            }
        } catch (Throwable t) {
        }

        return 0;
    }
}

• NioMessageUnsafe#read的doReadMessages方法内部会获取来连接的SocketChannel对象，然后封装成NioSocketChannel对象放置到readBuf当中。
• 针对readBuf中的NioSocketChannel对象，逐个通过NioServerSocketChannel对应的pipeline进行处理，注意、注意、注意，是监听的NioServerSocketChannel的pipeline，相当于accept的NioSocketChannel首先是需要通过NioServerSocketChannel的pipeline的处理链进行处理的。
• 在netty源码分析 - ChannelHandler文章中已经分析过NioServerSocketChannel对应的pipeline，这里我们直接进入pipeline当中的ServerBootstrapAcceptor的handler的分析。

ServerBootstrapAcceptor

public class ServerBootstrap extends AbstractBootstrap<ServerBootstrap, ServerChannel> {

    void init(Channel channel) throws Exception {
        final Map<ChannelOption<?>, Object> options = options0();
        synchronized (options) {
            setChannelOptions(channel, options, logger);
        }

        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) {
            currentChildOptions = childOptions.entrySet().toArray(newOptionArray(childOptions.size()));
        }
        synchronized (childAttrs) {
            currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(childAttrs.size()));
        }

        p.addLast(new ChannelInitializer<Channel>() {
            @Override
            public void initChannel(Channel ch) throws Exception {
                final ChannelPipeline pipeline = ch.pipeline();
                ChannelHandler handler = config.handler();
                if (handler != null) {
                    pipeline.addLast(handler);
                }

                ch.eventLoop().execute(new Runnable() {
                    @Override
                    public void run() {
                        pipeline.addLast(new ServerBootstrapAcceptor(
                                currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                    }
                });
            }
        });
    }
}

• ServerBootstrap#init过程中会在NioServerSocketChannel的pipeline当中增加ServerBootstrapAcceptor对象。
• ServerBootstrapAcceptor负责处理server端的accept请求。

ServerBootstrapAcceptor#channelRead

public class ServerBootstrap extends AbstractBootstrap<ServerBootstrap, ServerChannel> {

    private static class ServerBootstrapAcceptor extends ChannelInboundHandlerAdapter {

        private final EventLoopGroup childGroup;
        private final ChannelHandler childHandler;
        private final Entry<ChannelOption<?>, Object>[] childOptions;
        private final Entry<AttributeKey<?>, Object>[] childAttrs;

        ServerBootstrapAcceptor(
                EventLoopGroup childGroup, ChannelHandler childHandler,
                Entry<ChannelOption<?>, Object>[] childOptions, Entry<AttributeKey<?>, Object>[] childAttrs) {
            this.childGroup = childGroup;
            this.childHandler = childHandler;
            this.childOptions = childOptions;
            this.childAttrs = childAttrs;
        }

        @Override
        @SuppressWarnings("unchecked")
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            final Channel child = (Channel) msg;

            // accept的socketChannel负责绑定childHandler对象
            child.pipeline().addLast(childHandler);
            // accept的socketChannel负责添加具体的属性
            setChannelOptions(child, childOptions, logger);
            for (Entry<AttributeKey<?>, Object> e: childAttrs) {
                child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());
            }

            try {
                // 注册到childGroup当中
                childGroup.register(child).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        if (!future.isSuccess()) {
                            forceClose(child, future.cause());
                        }
                    }
                });
            } catch (Throwable t) {
                forceClose(child, t);
            }
        }
    }
}

• ServerBootstrapAcceptor#channelRead当中首先会针对新连接的NioSocketChannel的pipeline添加对应的childHandler，这个childHandler就是在初始化ServerBootstrap中指定的。
• NioSocketChannel绑定额外的属性，这些属性在初始化ServerBootstrap中指定的。
• 绑定NioSocketChannel到childGroup当中，通过childGroup.register(child)来完成，这里的childGroup在初始化ServerBootstrap中指定的。
• childGroup.register(child)实现了从bossGroup到workerGroup的注册转换。