世界要聞：【Netty源碼分析】03 客戶端接入流程

• 2023-03-28 22:17:11 來(lái)源：騰訊云

Netty服務(wù)端啟動(dòng)完成，這時(shí)候客戶端連接就可以接入進(jìn)來(lái)了，下面我們就來(lái)分析下客戶端連接接入的流程。

之前分析過(guò)NioEventLoop線程啟動(dòng)方法是startThread()，由于這個(gè)方法里面的邏輯比較復(fù)雜，并沒(méi)有展開(kāi)，這一節(jié)就是從這個(gè)方法開(kāi)始分析。

(資料圖片)

startThread

private void startThread() {    if (state == ST_NOT_STARTED) {        if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {            try {                doStartThread();            } catch (Throwable cause) {                STATE_UPDATER.set(this, ST_NOT_STARTED);                PlatformDependent.throwException(cause);            }        }    }}

復(fù)制

這個(gè)方法主要主要完成2件事：

利用cas將NioEventLoop的狀態(tài)由ST_NOT_STARTED修改成ST_STARTED，即表示NioEventLoop線程啟動(dòng)；執(zhí)行doStartThread()方法；

doStartThread()方法看著比較復(fù)雜，核心邏輯如下，向線程池執(zhí)行器executor提交一個(gè)任務(wù)，而這個(gè)線程池執(zhí)行器類型是ThreadPerTaskExecutor，即每次執(zhí)行任務(wù)都會(huì)創(chuàng)建一個(gè)新線程，而且這個(gè)任務(wù)是無(wú)限循環(huán)的：事件輪詢selector.select()、事件處理processSelectedKeys()和任務(wù)隊(duì)列處理runAllTasks()，這樣NioEventLoop就和具體的Thread線程進(jìn)行了關(guān)聯(lián)：

private void doStartThread() {    assert thread == null;    //executor線程執(zhí)行器，類型是：ThreadPerTaskExecutor，即每次執(zhí)行任務(wù)都會(huì)創(chuàng)建一個(gè)新線程    executor.execute(new Runnable() {        @Override        public void run() {            //將executor線程執(zhí)行器創(chuàng)建的線程：FastThreadLocalThread保存到EventLoop的全局變量中，相當(dāng)于thread和EventLoop的綁定            thread = Thread.currentThread();            if (interrupted) {                thread.interrupt();            }            boolean success = false;            updateLastExecutionTime();            try {                //然后調(diào)用EventLoop中的run方法進(jìn)行啟動(dòng)                SingleThreadEventExecutor.this.run();                success = true;            } catch (Throwable t) {                logger.warn("Unexpected exception from an event executor: ", t);            }        }    });}

復(fù)制

該方法大致完成2件事：

thread = Thread.currentThread();：將executor線程池分配的線程保存起來(lái)，這樣就完成了NioEventLoop和Thread線程的關(guān)聯(lián)；SingleThreadEventExecutor.this.run():具體實(shí)現(xiàn)在NioEventLoop.run()方法，所以，startThread()核心就是分配一個(gè)線程運(yùn)行NioEventLoop.run()方法。

protected void run() {    for (;;) {        try {            try {                switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {                case SelectStrategy.CONTINUE:// 默認(rèn)實(shí)現(xiàn)下，不存在這個(gè)情況                    continue;                case SelectStrategy.BUSY_WAIT:                case SelectStrategy.SELECT:                    //selector.select輪詢io事件                    select(wakenUp.getAndSet(false));                    if (wakenUp.get()) {                        selector.wakeup();                    }                default:                }            } catch (IOException e) {                rebuildSelector0();                handleLoopException(e);                continue;            }            cancelledKeys = 0;            needsToSelectAgain = false;            final int ioRatio = this.ioRatio;            if (ioRatio == 100) {               try {                    // 處理 Channel 感興趣的就緒 IO 事件                    processSelectedKeys();                } finally {                    // 運(yùn)行所有普通任務(wù)和定時(shí)任務(wù)，不限制時(shí)間                    runAllTasks();                }            } else {                final long ioStartTime = System.nanoTime();                try {                    // 處理IO事件                    processSelectedKeys();                } finally {                    // 運(yùn)行所有普通任務(wù)和定時(shí)任務(wù)，限制時(shí)間                    final long ioTime = System.nanoTime() - ioStartTime;                    runAllTasks(ioTime * (100 - ioRatio) / ioRatio);                }            }        } catch (Throwable t) {            handleLoopException(t);        }        //  EventLoop 優(yōu)雅關(guān)閉        try {            if (isShuttingDown()) {                closeAll();                if (confirmShutdown()) {                    return;                }            }        } catch (Throwable t) {            handleLoopException(t);        }    }}

復(fù)制

該方法主要完成三件事：

select(wakenUp.getAndSet(false))：主要執(zhí)行selector.select()方法進(jìn)行事件輪詢processSelectedKeys()：如果輪詢到事件，會(huì)在這里進(jìn)行處理runAllTasks()：處理任務(wù)隊(duì)列和定時(shí)任務(wù)隊(duì)列中的任務(wù)

下面我們就分別來(lái)分析下這三個(gè)方法。

select

private void select(boolean oldWakenUp) throws IOException {    Selector selector = this.selector;    try {        int selectCnt = 0;//計(jì)數(shù)器置0        long currentTimeNanos = System.nanoTime();        /**         * selectDeadLineNanos是select()方法運(yùn)行的截止時(shí)間         *         * currentTimeNanos:可以看成當(dāng)前時(shí)間         * delayNanos(currentTimeNanos)：獲取間隔時(shí)間，這里分為兩種情況：         * 1、netty里面定時(shí)任務(wù)隊(duì)列scheduledTaskQueue是按照延遲時(shí)間從小到大進(jìn)行排序，如果定時(shí)任務(wù)隊(duì)列中有任務(wù)，         * 則只需要獲取到第一個(gè)任務(wù)的啟動(dòng)時(shí)間 - 當(dāng)前時(shí)間 = select()方法可以運(yùn)行的時(shí)間間隔，即：select()方法要在第一個(gè)定時(shí)任務(wù)執(zhí)行之前退出，這樣才能去執(zhí)行定時(shí)任務(wù)         * 2、如果定時(shí)任務(wù)隊(duì)列沒(méi)有任務(wù)，則delayNanos(currentTimeNanos)返回1秒對(duì)應(yīng)的時(shí)間間隔         *         */        long selectDeadLineNanos = currentTimeNanos + delayNanos(currentTimeNanos);        for (;;) {            //計(jì)算超時(shí)時(shí)間            long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L;            /**             * timeoutMillis <= 0表示當(dāng)前已經(jīng)超時(shí)了，不能繼續(xù)向下執(zhí)行select()方法了，需要立即退出select方法，在退出前還有個(gè)判斷：selectCnt == 0             * selectCnt == 0表示第一次進(jìn)入循環(huán)，則執(zhí)行下Selector.selectNow()檢出準(zhǔn)備好的網(wǎng)絡(luò)IO事件，該方法不會(huì)阻塞，             */            if (timeoutMillis <= 0) {                if (selectCnt == 0) {                    selector.selectNow();                    selectCnt = 1;                }                break;            }            /**             * 如果沒(méi)有超時(shí)，但是通過(guò)hasTasks()判斷到taskQueue任務(wù)隊(duì)列中有需要執(zhí)行的任務(wù)，這時(shí)也需要退出select()方法             * 1、利用cas將wakeUp值由false變成true，wakeUp=true表示線程處于喚醒狀態(tài)，可以執(zhí)行任務(wù)，進(jìn)入select()方法前會(huì)把wakeUp設(shè)置成false             * 表示線程處于select()方法阻塞中，不能處理任務(wù)隊(duì)列中的任務(wù)，這時(shí)只要處理Selector.select()             * 2、退出前執(zhí)行一次：selector.selectNow()             */            if (hasTasks() && wakenUp.compareAndSet(false, true)) {                //有任務(wù)，進(jìn)行一次非阻塞式的select                selector.selectNow();                selectCnt = 1;                break;            }            //調(diào)用select方法，阻塞時(shí)間為上面算出的最近一個(gè)將要超時(shí)的定時(shí)任務(wù)時(shí)間            /**             * 未超時(shí)，任務(wù)隊(duì)列中也沒(méi)有需要執(zhí)行的任務(wù)，這時(shí)就可以放心的執(zhí)行Selector.select()方法了，這里帶上之前計(jì)算出的超時(shí)時(shí)間             * 如果之前計(jì)算時(shí)存在定時(shí)任務(wù)，則保證在第一個(gè)定時(shí)任務(wù)啟動(dòng)前喚醒即可，沒(méi)有定時(shí)任務(wù)則默認(rèn)超時(shí)1秒             */            int selectedKeys = selector.select(timeoutMillis);            //輪詢次數(shù)+1            selectCnt ++;            /**             * 發(fā)生如下幾種情況，select()方法都需要退出：             * 1、selectedKeys != 0:表示輪詢到IO事件             * 2、oldWakenUp:這個(gè)是入?yún)ⅲ禐閒alse，是在select()方法中控制是否需要退出，默認(rèn)是沒(méi)有使用到的，沒(méi)有意義             * 3、wakenUp.get()：進(jìn)入select()方法之前，wakeUp被設(shè)置成false，如果這里為true，表示已有外部線程對(duì)線程進(jìn)行喚醒操作，             *      一般就是addTask()添加新任務(wù)時(shí)會(huì)觸發(fā)喚醒，然后及時(shí)去執(zhí)行taskQueue中的任務(wù)             * 4、hasTasks() || hasScheduledTasks()：判斷任務(wù)隊(duì)列和定時(shí)任務(wù)隊(duì)列是否有任務(wù)需要執(zhí)行             */            if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks() || hasScheduledTasks()) {                break;            }            //線程中斷響應(yīng)：如果線程被中斷，計(jì)數(shù)器置1，break退出for循環(huán)，則退出select()檢測(cè)            if (Thread.interrupted()) {                if (logger.isDebugEnabled()) {                    logger.debug("Selector.select() returned prematurely because " +                            "Thread.currentThread().interrupt() was called. Use " +                            "NioEventLoop.shutdownGracefully() to shutdown the NioEventLoop.");                }                selectCnt = 1;                break;            }            /**             * 正常情況下：time >= currentTimeNanos + TimeUnit.MILLISECONDS.toNanos(timeoutMillis)             * 但是，jdk nio中存在一個(gè)bug，selector.select(timeoutMillis)在沒(méi)有IO事件觸發(fā)時(shí)并不會(huì)等待超時(shí)而是立即返回，造成空輪詢             *             * 下面就是Netty解決空輪詢問(wèn)題             * 1、if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos)             *      表示selector.select(timeoutMillis)經(jīng)過(guò)超時(shí)后才被喚醒，屬于正常情況，把selectCnt重置成1             * 2、如果不是，表示可能發(fā)生空輪詢selectCnt不會(huì)被重置成1，for循環(huán)一次selectCnt就會(huì)被累加1次；             * 3、等到 selectCnt > 門限值，默認(rèn)是512，可以通過(guò)io.netty.selectorAutoRebuildThreshold參數(shù)設(shè)置，             *      則判斷真正發(fā)生了nio空循環(huán)bug，則重建Selector替換掉當(dāng)前這個(gè)出問(wèn)題的Selector             */            long time = System.nanoTime();            //判斷執(zhí)行了一次阻塞式select后，當(dāng)前時(shí)間和開(kāi)始時(shí)間是否大于超時(shí)時(shí)間。（大于是很正常的，小于的話，說(shuō)明沒(méi)有執(zhí)行發(fā)生了空輪詢）            if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {                selectCnt = 1;             } else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 &&                    selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {                selector = selectRebuildSelector(selectCnt);                selectCnt = 1;                break;            }            currentTimeNanos = time;        }    } catch (CancelledKeyException e) {        if (logger.isDebugEnabled()) {            logger.debug(CancelledKeyException.class.getSimpleName() + " raised by a Selector {} - JDK bug?",                        selector, e);        }    }}

復(fù)制

select()方法代碼看著很復(fù)雜，其核心思想理解再來(lái)分析就比較簡(jiǎn)單的。select()主要是用來(lái)執(zhí)行selector.select()對(duì)IO事件進(jìn)行輪詢，作為server，這里就是輪詢OP_ACCEPT事件，看是否有客戶端接入進(jìn)來(lái)。但是NioEventLoop是單線程處理模式，不可能讓線程一直處理selector.select()，還有輪詢到的事件以及任務(wù)隊(duì)列中任務(wù)等等都需要使用這個(gè)線程進(jìn)行處理，所以，上面一大堆代碼都是用來(lái)判斷什么時(shí)候退出select()方法的，總結(jié)下退出邏輯主要分為如下幾種情況：

在執(zhí)行selector.select()方法之前，計(jì)算出一個(gè)超時(shí)時(shí)間，超時(shí)時(shí)間默認(rèn)是1秒，如果定時(shí)任務(wù)隊(duì)列有任務(wù)，則取出第一個(gè)任務(wù)(按順序存放)，保證在該定時(shí)任務(wù)執(zhí)行之前退出select()方法即可；如果超時(shí)就退出，退出前判斷是否是第一次進(jìn)入for循環(huán)，如果是在退出之前調(diào)用一次無(wú)阻塞的selector.selectNow()輪詢下判斷任務(wù)隊(duì)列taskQueue中是否有任務(wù)，如果有則將wakenUp利用cas設(shè)置成true，執(zhí)行下無(wú)阻塞的selector.selectNow()輪詢后退出select()方法如果上面情況都不存在，開(kāi)始執(zhí)行阻塞selector.select(timeoutMillis)輪詢，并將之前計(jì)算的超時(shí)時(shí)間帶上；selector.select(timeoutMillis)執(zhí)行完成后，繼續(xù)判斷是否需要退出select()方法，發(fā)生如下任一情況則要退出：輪詢到IO事件，則需要退出select()方法去處理事件外部線程對(duì)對(duì)線程執(zhí)行過(guò)喚醒操作，比如addTask()等操作需要喚醒線程執(zhí)行隊(duì)列任務(wù)，才能及時(shí)去執(zhí)行taskQueue中的任務(wù)：進(jìn)入select()方法之前，wakeUp被設(shè)置成false，如果這里為true，表示已有外部線程對(duì)線程進(jìn)行喚醒操作任務(wù)隊(duì)列taskQueue或定時(shí)任務(wù)隊(duì)列scheduledTaskQueue中有需要處理的任務(wù)，這時(shí)需要退出select()方法，轉(zhuǎn)去執(zhí)行任務(wù)

processSelectedKeys

private void processSelectedKeys() {    //selectedKeys != null表示已對(duì)Selector進(jìn)行優(yōu)化過(guò)，替換掉Selector內(nèi)部的selectedKeys，正常情況下進(jìn)入這個(gè)流程    if (selectedKeys != null) {        processSelectedKeysOptimized();    } else {        processSelectedKeysPlain(selector.selectedKeys());    } }

復(fù)制

processSelectedKeys()主要是對(duì)selector.select()方法輪詢到的事件進(jìn)行處理，作為server，如果輪詢到OP_ACCEPT，就表示有客戶端接入進(jìn)來(lái)了，那我們就跟蹤下這個(gè)方法，看接入進(jìn)來(lái)的客戶端處理流程。

Netty是對(duì)Selector進(jìn)行了優(yōu)化，將selectedKeys由Set實(shí)現(xiàn)替換成了數(shù)組實(shí)現(xiàn)，提升性能，所以，這里一般走的是processSelectedKeysOptimized()這個(gè)流程：

private void processSelectedKeysOptimized() {    for (int i = 0; i < selectedKeys.size; ++i) {        final SelectionKey k = selectedKeys.keys[i];        selectedKeys.keys[i] = null;        //k.attachment()獲取到的就是NioServerSocketChannel        final Object a = k.attachment();        if (a instanceof AbstractNioChannel) {//一般是走這個(gè)分支流程            processSelectedKey(k, (AbstractNioChannel) a);        } else {            @SuppressWarnings("unchecked")            NioTask task = (NioTask) a;            processSelectedKey(k, task);        }        if (needsToSelectAgain) {            selectedKeys.reset(i + 1);            selectAgain();            i = -1;        }    }}

復(fù)制

這里關(guān)鍵一點(diǎn)是Object a = k.attachment();，之前分析過(guò)向selector注冊(cè)時(shí)把NioServerSocketChannel作為attachment添加進(jìn)去，所以，這里取出來(lái)的就是NioServerSocketChannel對(duì)象。processSelectedKey()方法通過(guò)if判斷事件類型進(jìn)行處理，server端這里肯定是OP_ACCEPT：

if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) { unsafe.read();}

復(fù)制

具體的處理邏輯交由Unsafe對(duì)象進(jìn)行處理：

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 {                //doReadMessages()讀出來(lái)一個(gè)客戶端連接的Channel                int localRead = doReadMessages(readBuf);                if (localRead == 0) {                    break;                }                if (localRead < 0) {                    closed = true;                    break;                }                allocHandle.incMessagesRead(localRead);            } while (allocHandle.continueReading());        } catch (Throwable t) {            exception = t;        }        int size = readBuf.size();        for (int i = 0; i < size; i ++) {            readPending = false;            pipeline.fireChannelRead(readBuf.get(i));        }        readBuf.clear();        allocHandle.readComplete();        pipeline.fireChannelReadComplete();        if (exception != null) {            closed = closeOnReadError(exception);            pipeline.fireExceptionCaught(exception);        }        if (closed) {            inputShutdown = true;            if (isOpen()) {                close(voidPromise());            }        }    } finally {        if (!readPending && !config.isAutoRead()) {            removeReadOp();        }    }}

復(fù)制

這個(gè)類主要完成2件事：

doReadMessages(readBuf)：調(diào)用serverSocketChannel.accept()接收到客戶端連接socketChannel，并封裝成Netty中類型：NioSocketChannel，然后放入到readBuf集合中；pipeline.fireChannelRead(readBuf.get(i));：將讀入的客戶端連接作為參數(shù)，即NioSocketChannel對(duì)象，通過(guò)pipeline觸發(fā)channelRead事件進(jìn)行handler間傳播，注意這里的pipeline是NioServerSocketChannel中的，即server端的。最終會(huì)進(jìn)入到ServerBootstrapAcceptor#channelRead方法中進(jìn)行處理。

public void channelRead(ChannelHandlerContext ctx, Object msg) {    final Channel child = (Channel) msg;    child.pipeline().addLast(childHandler);    setChannelOptions(child, childOptions, logger);    for (Entry, Object> e: childAttrs) {        child.attr((AttributeKey