作者： 一字馬胡
轉(zhuǎn)載標(biāo)志 【2017-11-03】

更新日志

NI/O C/S通信過(guò)程

下面分別展示了NI/O模式下的客戶(hù)端/服務(wù)端編程模型：

NI/O服務(wù)端

NI/O客戶(hù)端

Netty是一種基于NI/O的網(wǎng)絡(luò)框架，網(wǎng)絡(luò)層面的操作只是對(duì)NI/O提供的API的封裝，所以，它的服務(wù)端流程和客戶(hù)端流程是和NI/O的一致的，對(duì)于客戶(hù)端而言，它要做的事情就是連接到服務(wù)端，然后發(fā)送/接收消息。對(duì)于服務(wù)端而言，它要bind一個(gè)端口，然后等待客戶(hù)端的連接，接收客戶(hù)端的連接使用的是Accept操作，一個(gè)服務(wù)端需要為多個(gè)客戶(hù)端提供服務(wù)，而每次accept都會(huì)生成一個(gè)新的Channel，Channel是一個(gè)服務(wù)端和客戶(hù)端之間數(shù)據(jù)傳輸?shù)耐ǖ溃梢韵蚱鋡rite數(shù)據(jù)，或者從中read數(shù)據(jù)。本文將分析Netty框架的Accept流程。

Netty的Accept流程

Netty是一個(gè)較為復(fù)雜的網(wǎng)絡(luò)框架，想要理解它的設(shè)計(jì)需要首先了解NI/O的相關(guān)知識(shí)，為了對(duì)Netty框架有一個(gè)大概的了解，你可以參考Netty線程模型及EventLoop詳解，該文章詳解解析了Netty中重要的事件循環(huán)處理流程，包含EventLoop的初始化和啟動(dòng)等相關(guān)內(nèi)容。下面首先展示了Netty中的EventLoop的分配模型，Netty服務(wù)端會(huì)為每一個(gè)新建立的Channel分配一個(gè)EventLoop，并且這個(gè)EventLoop將服務(wù)于這個(gè)Channel得整個(gè)生命周期不會(huì)改變，而一個(gè)EventLoop可能會(huì)被分配給多個(gè)Channel，也就是一個(gè)EventLoop可能會(huì)服務(wù)于多個(gè)Channel的讀寫(xiě)事件，這對(duì)于想要使用ThreadLocal的場(chǎng)景需要認(rèn)真考慮。

Netty NI/O模式下EventLoop分配模型

在文章Netty線程模型及EventLoop詳解中已經(jīng)分析了EventLoop的流程，現(xiàn)在從事件循環(huán)的起點(diǎn)開(kāi)始看起，也就是NioEventLoop的run方法，本文關(guān)心的是Netty的Accept事件，當(dāng)在Channel上發(fā)生了事件之后，會(huì)執(zhí)行processSelectedKeysPlain方法，看一下這個(gè)方法：

 private void processSelectedKeysPlain(Set<SelectionKey> selectedKeys) {
        // check if the set is empty and if so just return to not create garbage by
        // creating a new Iterator every time even if there is nothing to process.
        // See https://github.com/netty/netty/issues/597
        if (selectedKeys.isEmpty()) {
            return;
        }

        Iterator<SelectionKey> i = selectedKeys.iterator();
        for (;;) {
            final SelectionKey k = i.next();
            final Object a = k.attachment();
            i.remove();

            if (a instanceof AbstractNioChannel) {
                processSelectedKey(k, (AbstractNioChannel) a);
            } else {
                @SuppressWarnings("unchecked")
                NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
                processSelectedKey(k, task);
            }

            if (!i.hasNext()) {
                break;
            }

            if (needsToSelectAgain) {
                selectAgain();
                selectedKeys = selector.selectedKeys();

                // Create the iterator again to avoid ConcurrentModificationException
                if (selectedKeys.isEmpty()) {
                    break;
                } else {
                    i = selectedKeys.iterator();
                }
            }
        }
    }

接著會(huì)執(zhí)行processSelectedKey這個(gè)方法，下面是它的細(xì)節(jié)：

    private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
        final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
        try {
            int readyOps = k.readyOps();
            // We first need to call finishConnect() before try to trigger a read(...) or write(...) as otherwise
            // the NIO JDK channel implementation may throw a NotYetConnectedException.
            if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
                // remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
                // See https://github.com/netty/netty/issues/924
                int ops = k.interestOps();
                ops &= ~SelectionKey.OP_CONNECT;
                k.interestOps(ops);

                unsafe.finishConnect();
            }

            // Process OP_WRITE first as we may be able to write some queued buffers and so free memory.
            if ((readyOps & SelectionKey.OP_WRITE) != 0) {
                // Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to write
                ch.unsafe().forceFlush();
            }

            // Also check for readOps of 0 to workaround possible JDK bug which may otherwise lead
            // to a spin loop
            if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
                unsafe.read();
            }
        } catch (CancelledKeyException ignored) {
            unsafe.close(unsafe.voidPromise());
        }
    }

現(xiàn)在我們可以看到和NI/O一樣的類(lèi)似OP_READ和OP_ACCEPT之類(lèi)的東西了，OP_ACCEPT表示的是有Accept事件發(fā)生了，需要我們處理，但是發(fā)現(xiàn)好像OP_READ 事件和OP_ACCEPT事件的處理都是通過(guò)一個(gè)read方法進(jìn)行的，我們先來(lái)找到這個(gè)read方法：

-------------------------------------------------
AbstractNioMessageChannel.NioMessageUnsafe.read
-------------------------------------------------
        public void read() {
            try {
                try {
                    do {
                        int localRead = doReadMessages(readBuf);
                        if (localRead == 0) {
                            break;
                        }
                        if (localRead < 0) {
                            closed = true;
                            break;
                        }

                        allocHandle.incMessagesRead(localRead);
                    } while (allocHandle.continueReading());
                }

                int size = readBuf.size();
                for (int i = 0; i < size; i ++) {
                    readPending = false;
                    pipeline.fireChannelRead(readBuf.get(i));
                }
                
                readBuf.clear();
                allocHandle.readComplete();
                pipeline.fireChannelReadComplete();

            } 
         }
    }

本文中的所有代碼都是已經(jīng)處理過(guò)的，完整的代碼參考源代碼，本文為了控制篇幅去除了一些不影響閱讀（影響邏輯）的代碼，上面的read方法中有一個(gè)關(guān)鍵的方法doReadMessages，下面是它的實(shí)現(xiàn)：

--------------------------------------------
NioServerSocketChannel. doReadMessages
--------------------------------------------

    protected int doReadMessages(List<Object> buf) throws Exception {
        SocketChannel ch = SocketUtils.accept(javaChannel());

        try {
            if (ch != null) {
                buf.add(new NioSocketChannel(this, ch));
                return 1;
            }
        } catch (Throwable t) {
            logger.warn("Failed to create a new channel from an accepted socket.", t);

            try {
                ch.close();
            } catch (Throwable t2) {
                logger.warn("Failed to close a socket.", t2);
            }
        }

        return 0;
    }

這個(gè)方法就是處理accept類(lèi)型的事件的，為了更好的理解上面的代碼，下面展示一段在NI/O中服務(wù)端的代碼：

int port = 8676;
ServerSocketChannel serverChannel = ServerSocketChannel.open();
serverChannel.configureBlocking (false);
ServerSocket serverSocket = serverChannel.socket();
serverSocket.bind (new InetSocketAddres(port));
Selector selector = Selector.open();
serverChannel.register (selector, SelectionKey.OP_ACCEPT);
while (true) {
    int n = selector.select();
    Iterator it = selector.selectedKeys().iterator();
    while (it.hasNext()) {
        SelectionKey key = (SelectionKey) it.next();
        if (key.isAcceptable()) {
            ServerSocketChannel server = (ServerSocketChannel) key.channel();
            SocketChannel channel = server.accept();
            channel.configureBlocking (false);
            channel.register (selector, SelectionKey.OP_READ);
        }
        if (key.isReadable( )) {
            processReadEvent(key);
        }
        
        it.remove( );
    }
} 

可以看到，服務(wù)端accept一次就會(huì)產(chǎn)生一個(gè)新的Channel，Netty也是，每次Accept都會(huì)new一個(gè)新的NioSocketChannel，當(dāng)然，這個(gè)Channel需要分配一個(gè)EventLoop給他才能開(kāi)始事件循環(huán)，但是Netty服務(wù)端的Accept事件到此應(yīng)該可以清楚流程了，下面分析這個(gè)新的Channel是怎么開(kāi)始事件循環(huán)的。繼續(xù)看AbstractNioMessageChannel.NioMessageUnsafe.read這個(gè)方法，其中有一個(gè)句話：

pipeline.fireChannelRead(readBuf.get(i));

現(xiàn)在來(lái)跟蹤一下這個(gè)方法的調(diào)用鏈：

 -> ChannelPipeline.fireChannelRead
 -> DefaultChannelPipeline.fireChannelRead
 -> AbstractChannelHandlerContext.invokeChannelRead
 -> ChannelInboundHandler.channelRead
 ->ServerBootstrapAcceptor.channelRead

上面列出的是主要的調(diào)用鏈路，只是為了分析Accept的過(guò)程，到ServerBootstrapAcceptor.channelRead這個(gè)方法就可以看到是怎么分配EventLoop給Channel的了，下面展示了ServerBootstrapAcceptor.channelRead這個(gè)方法的細(xì)節(jié)：

        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            final Channel child = (Channel) msg;

            child.pipeline().addLast(childHandler);        【1】

            setChannelOptions(child, childOptions, logger);

            for (Entry<AttributeKey<?>, Object> e: childAttrs) {
                child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());   【2】
            }

            try {
                childGroup.register(child).addListener(new ChannelFutureListener() {   【3】
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        if (!future.isSuccess()) {
                            forceClose(child, future.cause());
                        }
                    }
                });
            } catch (Throwable t) {
                forceClose(child, t);
            }
        }

• 【1】首先將服務(wù)端的處理邏輯賦值給新建立的這個(gè)Channel得pipeline，使得這個(gè)新建立的Channel可以得到服務(wù)端提供的服務(wù)
• 【2】屬性賦值
• 【3】將這個(gè)新建立的Channel添加到EventLoopGroup中去，這里進(jìn)行EventLoop的分配

下面來(lái)仔細(xì)看一下【3】這個(gè)register方法：

========================================
MultithreadEventLoopGroup.register
========================================

    public ChannelFuture register(Channel channel) {
        return next().register(channel);
    }
    
========================================
SingleThreadEventLoop.register
========================================

    public ChannelFuture register(Channel channel) {
        return register(new DefaultChannelPromise(channel, this));
    }
    

========================================
AbstractUnsafe.register
========================================    
    
        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);
                }
            }
        }

上面的流程展示了這個(gè)新的Channel是怎么獲得一個(gè)EventLoop的，而EventLoopGroup分配EventLoop的關(guān)鍵在于MultithreadEventLoopGroup.register這個(gè)方法中的next方法，而這部分的分析已經(jīng)在Netty線程模型及EventLoop詳解中做過(guò)了，不再贅述。當(dāng)一個(gè)新的連接被服務(wù)端Accept之后，會(huì)創(chuàng)建一個(gè)新的Channel來(lái)維持服務(wù)端與客戶(hù)端之間的通信，而每個(gè)新建立的Channel都會(huì)被分配一個(gè)EventLoop來(lái)實(shí)現(xiàn)事件循環(huán)，本文分析了Netty服務(wù)端Accept的詳細(xì)過(guò)程，至此，對(duì)于Netty的EventLoop、EventLoopGroup以及EventLoop是如何被運(yùn)行起來(lái)的，以及服務(wù)端是如何Accept新的連接的這些問(wèn)題應(yīng)該都已經(jīng)有答案了。