什麼是libuv，淺析libuv中的事件輪詢（Node核心依賴）

這篇文章帶大家了解一下 Node 的核心依賴 libuv，介紹一下什麼是libuv，libuv中的事件輪詢，希望對大家有幫助！

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

很多高階語言，執行權限都可以觸及作業系統，而運行在瀏覽器端的JavaScript 則例外，瀏覽器為其創建的沙箱環境，把前端工程師封閉在一個編程世界的象牙塔里。不過 Node.js 的出現則彌補了這個缺憾，前端工程師也可以觸達電腦世界的底層。

所以 Nodejs 對於前端工程師的意義不僅在於提供了全端開發能力，更重要的是為前端工程師打開了一扇通往電腦底層世界的大門。本文透過分析 Node.js 的實作原理來打開這扇大門。

Node.js原始碼結構

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

• acorn：用 JavaScript 寫的輕量級 JavaScript 解析器。
• acorn-plugins：acorn 的擴充模組，讓 acorn 支援 ES6 特性解析，例如類別宣告。
• brotli：C 語言編寫的 Brotli 壓縮演算法。
• cares：應該寫為 “c-ares”，C 語言編寫的用來處理非同步 DNS 請求。
• histogram：C 語言編寫，實作長條圖產生功能。
• icu-small：C 語言編寫，為 Node.js 客製化的 ICU（International Components for Unicode）函式庫，包含一些用來操作 Unicode 的函式。
• llhttp：C 語言編寫，輕量級的 http 解析器。
• nghttp2/nghttp3/ngtcp2：處理 HTTP/2、HTTP/3、TCP/2 協定。
• node-inspect：讓 Node.js 程式支援 CLI debug 偵錯模式。
• npm：JavaScript 編寫的 Node.js 模組管理器。
• openssl：C 語言編寫，加密相關的模組，在 tls 和 crypto 模組中都有使用。
• uv：C 語言編寫，採用非阻塞型的 I/O 操作，為 Node.js 提供了存取系統資源的能力。
• uvwasi：C 語編寫，實作 WASI 系統呼叫 API。
• v8：C 語言編寫，JavaScript 引擎。
• zlib：用於快速壓縮，Node.js 使用 zlib 建立同步、非同步和資料流壓縮、解壓縮介面。

其中最重要的是 v8 和 uv 兩個目錄對應的模組。 v8本身並沒有非同步運行的能力，而是藉助瀏覽器的其他執行緒實現的，這也正是我們常說js是單執行緒的原因，因為其解析引擎只支援同步解析程式碼。 但在 Node.js 中，非同步實作主要依賴 libuv，以下我們將重點放在分析 libuv 的實作原理。

什麼是libuv

libuv 是一個用 C 寫的支援多平台的非同步 I/O 函式庫，主要解​​決 I/O 操作容易造成阻塞的問題。 一開始是專門為 Node.js 使用而開發的，但後來也被 Luvit、Julia、pyuv 等其他模組使用。下圖是 libuv 的結構圖。

libuv有兩種非同步的實作方式，分別是上圖左右兩個被黃框選取的部分。

左邊部分為網路I/O 模組，在不同平台下有不同的實現機制，Linux 系統下透過epoll 實現，OSX 和其他BSD 系統採用KQueue，SunOS 系統採用Event ports，Windows 系統採用的是IOCP。由於涉及作業系統底層 API，理解起來比較複雜，這裡就不多介紹了。

右邊部分包括檔案 I/O 模組、DNS 模組和使用者程式碼，透過執行緒池來實現非同步操作。檔案 I/O 與網路 I/O不同，libuv 沒有依賴系統底層的 API，而是在全域執行緒池中執行阻塞的檔案 I/O 操作。

libuv中的事件輪詢

下圖是 libuv 官網給的事件輪詢工作流程圖，我們結合程式碼一起分析。

libuv 事件循環的核心程式碼是在 uv_run() 函數中實現的，下面是 Unix 系統下的部分核心程式碼。雖然是用 C 語言寫的，但和 JavaScript 一樣都是高階語言，所以要理解也不太困難。最大的差異可能是星號和箭頭，星號我們可以直接忽略。例如，函數參數中 uv_loop_t* loop 可以理解為 uv_loop_t 類型的變數 loop。箭頭 “→” 可以理解為點號“.”，例如，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

這個函數用於判斷事件輪詢是否要繼續進行，如果loop 物件中不存在活躍的任務則返回0 並退出循環。

在 C 語言中這個 “任務” 有個專業的稱呼，即“句柄”，可以理解為指向任務的變數。句柄又可以分為兩類：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

為了減少與時間相關的系統呼叫次數，同構這個函數來快取目前系統時間，精度很高，可以達到奈秒級別，但單位還是毫秒。

具體原始碼如下：

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

uv__run_timers

執行setTimeout() 和setInterval() 中到達時間閾值的回調函數。這個執行過程是透過for 循環遍歷實現的，從下面的程式碼中也可以看到，定時器回調是儲存於一個最小堆結構的資料中的，當這個最小堆為空或還未到達時間閾值時退出循環。

在執行定時器回呼函數前先移除該定時器，如果設定了 repeat，需再次加到最小堆裡，然後執行定時器回調。

具體程式碼如下：

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 回呼函數，當pending_queue 為空時傳回0；否則在執行完pending_queue 中的回呼函數後傳回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 個函數都是透過一個巨集函數UV_LOOP_WATCHER_DEFINE

這3 個函數都是透過一個巨集函數UV_LOOP_WATCHER_DEFINE

這3 個函數都是透過一個巨集函數UV_LOOP_WATCHER_DEFINE進行定義的，巨集函數可以理解為程式碼模板，或者說用來定義函數的函數。 3 次呼叫巨集函數並分別傳入 name 參數值 prepare、check、idle，同時定義了 uvrun_idle、uvrun_prepare、uv__run_check 3 個函數。 所以說它們的執行邏輯是一致的，都是按照先進先出原則循環遍歷並取出隊列 loop->name##_handles 中的對象，然後執行對應的回調函數。

#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 作業。具體實作根據作業系統的不同會有所區別，我們以 Linux 系統為例進行分析。

uv__io_poll 函數原始碼較多，核心為兩段循環程式碼，部分程式碼如下：

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 迴圈中，遍歷觀察者佇列watcher_queue，並把事件和檔案描述符取出來賦值給事件物件e，然後呼叫epoll_ctl 函數來註冊或修改epoll 事件。 在 for 迴圈中，會先將 epoll 中等待的檔案描述子取出賦值給 nfds，然後再遍歷 nfds，執行回呼函數。

uv__run_closing_handles

#遍歷等待關閉的佇列，關閉 stream、tcp、udp 等 handle，然後呼叫 handle 對應的 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，但並不屬於事件輪詢的一部分，它們都有各自的任務隊列，在事件輪詢的每個步驟完成後執行。所以當我們使用這兩個非同步 API 的時候要注意，如果在傳入的回調函數中執行長任務或遞歸，則會導致事件輪詢被阻塞，從而 “餓死”I/O 操作。

下面的程式碼就是透過遞迴呼叫 prcoess.nextTick 而導致 fs.readFile 的回呼函數無法執行的範例。

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

要解決這個問題，可以使用 setImmediate 來替代，因為 setImmediate 會在事件輪詢中執行回呼函數佇列。 process.nextTick 任務佇列優先權比Promise任務佇列更高，具體的原因可以參考下面的程式碼：

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() 函數中可以看出，首先透過while 迴圈取出queue 佇列的回呼函數，而這個queue 佇列中的回呼函數就是透過process.nextTick 來加入的。當 while 迴圈結束後才呼叫runMicrotasks() 函數執行 Promise 的回呼函數。

###總結######Node.js 的核心依賴libuv的結構可以分成兩個部分，一部分是網路I/O，底層實作會根據不同作業系統依賴不同的系統API，另一部分是檔案I/O、DNS、使用者程式碼，這一部分採用線程池來處理。 ###

libuv 處理非同步操作的核心機制是事件輪詢，事件輪詢分成若干步驟，大致操作是遍歷並執行佇列中的回呼函數。

最後提到處理非同步的 API process.nextTick 和 Promise 不屬於事件輪詢，使用不當則會導致事件輪詢阻塞，其中一種解決方式就是使用 setImmediate 來替代。

更多node相關知識，請造訪：nodejs 教學！

以上是什麼是libuv，淺析libuv中的事件輪詢（Node核心依賴）的詳細內容。更多資訊請關注PHP中文網其他相關文章！