node.js中l(wèi)ibuv事件輪詢的示例分析

這篇文章主要為大家展示了“node.js中l(wèi)ibuv事件輪詢的示例分析”，內(nèi)容簡而易懂，條理清晰，希望能夠幫助大家解決疑惑，下面讓小編帶領(lǐng)大家一起研究并學(xué)習(xí)一下“node.js中l(wèi)ibuv事件輪詢的示例分析”這篇文章吧。

提到 Node.js，相信大部分前端工程師都會想到基于它來開發(fā)服務(wù)端，只需要掌握 JavaScript 一門語言就可以成為全棧工程師，但其實 Node.js 的意義并不僅于此。

很多高級語言，執(zhí)行權(quán)限都可以觸及操作系統(tǒng)，而運(yùn)行在瀏覽器端的 JavaScript 則例外，瀏覽器為其創(chuàng)建的沙箱環(huán)境，把前端工程師封閉在一個編程世界的象牙塔里。不過 Node.js 的出現(xiàn)則彌補(bǔ)了這個缺憾，前端工程師也可以觸達(dá)計算機(jī)世界的底層。

所以 Nodejs 對于前端工程師的意義不僅在于提供了全棧開發(fā)能力，更重要的是為前端工程師打開了一扇通向計算機(jī)底層世界的大門。本文通過分析 Node.js 的實現(xiàn)原理來打開這扇大門。

Node.js源碼結(jié)構(gòu)

Node.js 源碼倉庫的 /deps 目錄下有十幾個依賴，其中既有 C 語言編寫的模塊（如 libuv、V8）也有JavaScript 語言編寫的模塊（如 acorn、acorn-plugins），如下圖所示。

• acorn：用 JavaScript 編寫的輕量級 JavaScript 解析器。

• acorn-plugins：acorn 的擴(kuò)展模塊，讓 acorn 支持 ES6 特性解析，比如類聲明。

• brotli：C 語言編寫的 Brotli 壓縮算法。

• cares：應(yīng)該寫為 “c-ares”，C 語言編寫的用來處理異步 DNS 請求。

• histogram：C 語言編寫，實現(xiàn)柱狀圖生成功能。

• icu-small：C 語言編寫，為 Node.js 定制的 ICU（International Components for Unicode）庫，包括一些用來操作 Unicode 的函數(shù)。

• llhttp：C 語言編寫，輕量級的 http 解析器。

• nghttp2/nghttp3/ngtcp2：處理 HTTP/2、HTTP/3、TCP/2 協(xié)議。

• node-inspect：讓 Node.js 程序支持 CLI debug 調(diào)試模式。

• npm：JavaScript 編寫的 Node.js 模塊管理器。

• openssl：C 語言編寫，加密相關(guān)的模塊，在 tls 和 crypto 模塊中都有使用。

• uv：C 語言編寫，采用非阻塞型的 I/O 操作，為 Node.js 提供了訪問系統(tǒng)資源的能力。

• uvwasi：C 語編寫，實現(xiàn) WASI 系統(tǒng)調(diào)用 API。

• v8：C 語言編寫，JavaScript 引擎。

• zlib：用于快速壓縮，Node.js 使用 zlib 創(chuàng)建同步、異步和數(shù)據(jù)流壓縮、解壓接口。

其中最重要的是 v8 和 uv 兩個目錄對應(yīng)的模塊。v8本身并沒有異步運(yùn)行的能力，而是借助瀏覽器的其他線程實現(xiàn)的，這也正是我們常說js是單線程的原因，因為其解析引擎只支持同步解析代碼。 但在 Node.js 中，異步實現(xiàn)主要依賴于 libuv，下面我們來重點(diǎn)分析 libuv 的實現(xiàn)原理。

什么是libuv

libuv 是一個用 C 編寫的支持多平臺的異步 I/O 庫，主要解決 I/O 操作容易引起阻塞的問題。 最開始是專門為 Node.js 使用而開發(fā)的，但后來也被 Luvit、Julia、pyuv 等其他模塊使用。下圖是 libuv 的結(jié)構(gòu)圖。

libuv有兩種異步的實現(xiàn)方式，分別是上圖左右兩個被黃框選中的部分。

左邊部分為網(wǎng)絡(luò) I/O 模塊，在不同平臺下有不同的實現(xiàn)機(jī)制，Linux 系統(tǒng)下通過 epoll 實現(xiàn)，OSX 和其他 BSD 系統(tǒng)采用 KQueue，SunOS 系統(tǒng)采用 Event ports，Windows 系統(tǒng)采用的是 IOCP。由于涉及操作系統(tǒng)底層 API，理解起來比較復(fù)雜，這里就不多介紹了。

右邊部分包括文件 I/O 模塊、DNS 模塊和用戶代碼，通過線程池來實現(xiàn)異步操作。文件 I/O 與網(wǎng)絡(luò) I/O不同，libuv 沒有依賴于系統(tǒng)底層的 API，而是在全局線程池中執(zhí)行阻塞的文件 I/O 操作。

libuv中的事件輪詢

下圖是 libuv 官網(wǎng)給出的事件輪詢工作流程圖，我們結(jié)合代碼來一起分析。

libuv 事件循環(huán)的核心代碼是在 uv_run() 函數(shù)中實現(xiàn)的，下面是 Unix 系統(tǒng)下的部分核心代碼。雖然是用 C 語言編寫的，但和 JavaScript 一樣都是高級語言，所以理解起來也不算太困難。最大的區(qū)別可能是星號和箭頭，星號我們可以直接忽略。例如，函數(shù)參數(shù)中 uv_loop_t* loop 可以理解為 uv_loop_t 類型的變量 loop。箭頭 “→” 可以理解為點(diǎn)號“.”，例如，loop→stop_flag 可以理解為 loop.stop_flag。

int uv_run(uv_loop_t* loop, uv_run_mode mode) {
  ... 
r = uv__loop_alive(loop);
if (!r) uv__update_time(loop);
while (r != 0 && loop - >stop_flag == 0) {
    uv__update_time(loop);
    uv__run_timers(loop);
    ran_pending = uv__run_pending(loop);
    uv__run_idle(loop);
    uv__run_prepare(loop);...uv__io_poll(loop, timeout);
    uv__run_check(loop);
    uv__run_closing_handles(loop);...
}...
}

uv__loop_alive

這個函數(shù)用于判斷事件輪詢是否要繼續(xù)進(jìn)行，如果 loop 對象中不存在活躍的任務(wù)則返回 0 并退出循環(huán)。

在 C 語言中這個 “任務(wù)” 有個專業(yè)的稱呼，即“句柄”，可以理解為指向任務(wù)的變量。句柄又可以分為兩類：request 和 handle，分別代表短生命周期句柄和長生命周期句柄。具體代碼如下：

static int uv__loop_alive(const uv_loop_t * loop) {
    return uv__has_active_handles(loop) || uv__has_active_reqs(loop) || loop - >closing_handles != NULL;
}

uv__update_time

為了減少與時間相關(guān)的系統(tǒng)調(diào)用次數(shù)，同構(gòu)這個函數(shù)來緩存當(dāng)前系統(tǒng)時間，精度很高，可以達(dá)到納秒級別，但單位還是毫秒。

具體源碼如下：

UV_UNUSED(static void uv__update_time(uv_loop_t * loop)) {
    loop - >time = uv__hrtime(UV_CLOCK_FAST) / 1000000;
}

uv__run_timers

執(zhí)行 setTimeout() 和 setInterval() 中到達(dá)時間閾值的回調(diào)函數(shù)。這個執(zhí)行過程是通過 for 循環(huán)遍歷實現(xiàn)的，從下面的代碼中也可以看到，定時器回調(diào)是存儲于一個最小堆結(jié)構(gòu)的數(shù)據(jù)中的，當(dāng)這個最小堆為空或者還未到達(dá)時間閾值時退出循環(huán)。

在執(zhí)行定時器回調(diào)函數(shù)前先移除該定時器，如果設(shè)置了 repeat，需再次加到最小堆里，然后執(zhí)行定時器回調(diào)。

具體代碼如下：

void uv__run_timers(uv_loop_t * loop) {
    struct heap_node * heap_node;
    uv_timer_t * handle;
    for (;;) {
        heap_node = heap_min(timer_heap(loop));
        if (heap_node == NULL) break;
        handle = container_of(heap_node, uv_timer_t, heap_node);
        if (handle - >timeout > loop - >time) break;
        uv_timer_stop(handle);
        uv_timer_again(handle);
        handle - >timer_cb(handle);
    }
}

uv__run_pending

遍歷所有存儲在 pending_queue 中的 I/O 回調(diào)函數(shù)，當(dāng) pending_queue 為空時返回 0；否則在執(zhí)行完pending_queue 中的回調(diào)函數(shù)后返回 1。

代碼如下：

static int uv__run_pending(uv_loop_t * loop) {
    QUEUE * q;
    QUEUE pq;
    uv__io_t * w;
    if (QUEUE_EMPTY( & loop - >pending_queue)) return 0;
    QUEUE_MOVE( & loop - >pending_queue, &pq);
    while (!QUEUE_EMPTY( & pq)) {
        q = QUEUE_HEAD( & pq);
        QUEUE_REMOVE(q);
        QUEUE_INIT(q);
        w = QUEUE_DATA(q, uv__io_t, pending_queue);
        w - >cb(loop, w, POLLOUT);
    }
    return 1;
}

uvrun_idle / uvrun_prepare / uv__run_check

這 3 個函數(shù)都是通過一個宏函數(shù) UV_LOOP_WATCHER_DEFINE 進(jìn)行定義的，宏函數(shù)可以理解為代碼模板，或者說用來定義函數(shù)的函數(shù)。3 次調(diào)用宏函數(shù)并分別傳入 name 參數(shù)值 prepare、check、idle，同時定義了 uvrun_idle、uvrun_prepare、uv__run_check 3 個函數(shù)。

所以說它們的執(zhí)行邏輯是一致的，都是按照先進(jìn)先出原則循環(huán)遍歷并取出隊列 loop->name##_handles 中的對象，然后執(zhí)行對應(yīng)的回調(diào)函數(shù)。

#define UV_LOOP_WATCHER_DEFINE(name, type)
void uv__run_##name(uv_loop_t* loop) {
  uv_##name##_t* h;
  QUEUE queue;
  QUEUE* q;
  QUEUE_MOVE(&loop->name##_handles, &queue);
  while (!QUEUE_EMPTY(&queue)) {
    q = QUEUE_HEAD(&queue);
    h = QUEUE_DATA(q, uv_##name##_t, queue);
    QUEUE_REMOVE(q);
    QUEUE_INSERT_TAIL(&loop->name##_handles, q);
    h->name##_cb(h);
  }
}
UV_LOOP_WATCHER_DEFINE(prepare, PREPARE) 
UV_LOOP_WATCHER_DEFINE(check, CHECK) 
UV_LOOP_WATCHER_DEFINE(idle, IDLE)

uv__io_poll

uv__io_poll 主要是用來輪詢 I/O 操作。具體實現(xiàn)根據(jù)操作系統(tǒng)的不同會有所區(qū)別，我們以 Linux 系統(tǒng)為例進(jìn)行分析。

uv__io_poll 函數(shù)源碼較多，核心為兩段循環(huán)代碼，部分代碼如下：

void uv__io_poll(uv_loop_t * loop, int timeout) {
    while (!QUEUE_EMPTY( & loop - >watcher_queue)) {
        q = QUEUE_HEAD( & loop - >watcher_queue);
        QUEUE_REMOVE(q);
        QUEUE_INIT(q);
        w = QUEUE_DATA(q, uv__io_t, watcher_queue);
        e.events = w - >pevents;
        e.data.fd = w - >fd;
        if (w - >events == 0) op = EPOLL_CTL_ADD;
        else op = EPOLL_CTL_MOD;
        if (epoll_ctl(loop - >backend_fd, op, w - >fd, &e)) {
            if (errno != EEXIST) abort();
            if (epoll_ctl(loop - >backend_fd, EPOLL_CTL_MOD, w - >fd, &e)) abort();
        }
        w - >events = w - >pevents;
    }
    for (;;) {
        for (i = 0; i < nfds; i++) {
            pe = events + i;
            fd = pe - >data.fd;
            w = loop - >watchers[fd];
            pe - >events &= w - >pevents | POLLERR | POLLHUP;
            if (pe - >events == POLLERR || pe - >events == POLLHUP) pe - >events |= w - >pevents & (POLLIN | POLLOUT | UV__POLLRDHUP | UV__POLLPRI);
            if (pe - >events != 0) {
                if (w == &loop - >signal_io_watcher) have_signals = 1;
                else w - >cb(loop, w, pe - >events);
                nevents++;
            }
        }
        if (have_signals != 0) loop - >signal_io_watcher.cb(loop, &loop - >signal_io_watcher, POLLIN);
    }...
}

在 while 循環(huán)中，遍歷觀察者隊列 watcher_queue，并把事件和文件描述符取出來賦值給事件對象 e，然后調(diào)用 epoll_ctl 函數(shù)來注冊或修改 epoll 事件。

在 for 循環(huán)中，會先將 epoll 中等待的文件描述符取出賦值給 nfds，然后再遍歷 nfds，執(zhí)行回調(diào)函數(shù)。

uv__run_closing_handles

遍歷等待關(guān)閉的隊列，關(guān)閉 stream、tcp、udp 等 handle，然后調(diào)用 handle 對應(yīng)的 close_cb。代碼如下：

static void uv__run_closing_handles(uv_loop_t * loop) {
    uv_handle_t * p;
    uv_handle_t * q;
    p = loop - >closing_handles;
    loop - >closing_handles = NULL;
    while (p) {
        q = p - >next_closing;
        uv__finish_close(p);
        p = q;
    }
}

process.nextTick 和 Promise

雖然 process.nextTick 和 Promise 都是異步 API，但并不屬于事件輪詢的一部分，它們都有各自的任務(wù)隊列，在事件輪詢的每個步驟完成后執(zhí)行。所以當(dāng)我們使用這兩個異步 API 的時候要注意，如果在傳入的回調(diào)函數(shù)中執(zhí)行長任務(wù)或遞歸，則會導(dǎo)致事件輪詢被阻塞，從而 “餓死”I/O 操作。

下面的代碼就是通過遞歸調(diào)用 prcoess.nextTick 而導(dǎo)致 fs.readFile 的回調(diào)函數(shù)無法執(zhí)行的例子。

fs.readFile('config.json', (err, data) = >{...
}) const traverse = () = >{
    process.nextTick(traverse)
}

要解決這個問題，可以使用 setImmediate 來替代，因為 setImmediate 會在事件輪詢中執(zhí)行回調(diào)函數(shù)隊列。process.nextTick 任務(wù)隊列優(yōu)先級比 Promise任務(wù)隊列更高，具體的原因可以參看下面的代碼：

function processTicksAndRejections() {
    let tock;
    do {
        while (tock = queue.shift()) {
            const asyncId = tock[async_id_symbol];
            emitBefore(asyncId, tock[trigger_async_id_symbol], tock);
            try {
                const callback = tock.callback;
                if (tock.args === undefined) {
                    callback();
                } else {
                    const args = tock.args;
                    switch (args.length) {
                    case 1:
                        callback(args[0]);
                        break;
                    case 2:
                        callback(args[0], args[1]);
                        break;
                    case 3:
                        callback(args[0], args[1], args[2]);
                        break;
                    case 4:
                        callback(args[0], args[1], args[2], args[3]);
                        break;
                    default:
                        callback(...args);
                    }
                }
            } finally {
                if (destroyHooksExist()) emitDestroy(asyncId);
            }
            emitAfter(asyncId);
        }
        runMicrotasks();
    } while (! queue . isEmpty () || processPromiseRejections());
    setHasTickScheduled(false);
    setHasRejectionToWarn(false);
}

從 processTicksAndRejections() 函數(shù)中可以看出，首先通過 while 循環(huán)取出 queue 隊列的回調(diào)函數(shù)，而這個 queue 隊列中的回調(diào)函數(shù)就是通過 process.nextTick 來添加的。當(dāng) while 循環(huán)結(jié)束后才調(diào)用runMicrotasks() 函數(shù)執(zhí)行 Promise 的回調(diào)函數(shù)。

以上是“node.js中l(wèi)ibuv事件輪詢的示例分析”這篇文章的所有內(nèi)容，感謝各位的閱讀！相信大家都有了一定的了解，希望分享的內(nèi)容對大家有所幫助，如果還想學(xué)習(xí)更多知識，歡迎關(guān)注億速云行業(yè)資訊頻道！