作者 | symen

在Linux系統中，實際上所有的I/O設備都被抽象為文件這個概念，一切皆文件（Everything is File）。無論是磁盤、網絡數據、終端，還是進程間通信工具（如：管道pipe）等都被抽象為文件的概念。 這種設計使得 I/O 操作可以通過統一的文件描述符（File Descriptor, FD）來管理。 在了解多路復用select、poll、epoll實現之前，我們先簡單回憶復習以下兩個概念。

一、什么是多路復用

多路: 是指多個網絡連接（Socket）

復用: 是指通過一個線程同時監控多個文件描述符的就緒狀態。這樣，程序可以高效地處理多個 I/O 事件，而不需要為每個連接創建單獨的線程，從而節省系統資源。

多路復用主要有三種技術：select，poll，epoll。其中，epoll 是最新且性能最優的實現，能夠高效地處理大規模并發連接。

二、五種IO模型

[1]blockingIO - 阻塞IO
[2]nonblockingIO - 非阻塞IO
[3]signaldrivenIO - 信號驅動IO
[4]asynchronousIO - 異步IO
[5]IOmultiplexing - IO多路復用

1. 阻塞式I/O模型

在阻塞式 I/O 模型中，在I/O操作的兩個階段均會阻塞線程:

• 數據等待階段：當進程或線程發起IO請求（如：調用 recvfrom 系統調用）時，它會一直阻塞，直到內核確認數據已準備好（例：網卡接收數據、網絡數據到達內核緩沖區）。
• 數據復制階段：內核將數據從內核空間復制到用戶空間時，線程/進程仍處于阻塞狀態。此過程線程/進程在等待I/O完成期間無法執行其他任務（被掛起），CPU資源可能閑置。

主要特點：

• 阻塞掛起： 進程/線程在等待數據時會被掛起，不占用 CPU 資源。
• 及時響應： 每個操作都能得到及時處理，適合對實時性要求較高的場景。
• 實現簡單： 開發難度低，邏輯直觀，代碼按順序執行，無需處理多線程或異步回調的復雜性。
• 適用場景： 阻塞式 I/O 模型適合并發量較小、對實時性要求較高的應用。但在高并發場景中，其系統開銷和性能限制使其不再適用。
• 系統開銷大：由于每個請求都會阻塞進程/線程，因此需要為每個請求分配獨立的進程或線程來處理。在高并發場景下，這種模型會消耗大量系統資源（如內存和上下文切換開銷），導致性能瓶頸。

2. 非阻塞式I/O模型

在非阻塞式 I/O 模型中，當進程發起 I/O 系統調用（如 recvfrom）時：

• 如果內核緩沖區沒有數據，內核會立即返回一個錯誤（如 EWOULDBLOCK 或 EAGAIN），而不會阻塞進程。
• 如果內核緩沖區有數據，內核會將數據復制到用戶空間并返回成功。

主要特點：

• 非阻塞： 進程不會被掛起，無論是否有數據都會立即返回。
• 輪詢機制： 進程需要不斷發起系統調用（輪詢）來檢查數據是否就緒，這會消耗大量 CPU 資源。
• 實現難度較低： 相比阻塞式 I/O，開發復雜度稍高，但仍屬于較簡單的模型。
• 實時性差： 輪詢機制無法保證及時響應數據到達事件，可能導致延遲。
• 適用場景： 適合并發量較小、且對實時性要求不高的網絡應用開發。由于其 CPU 開銷較大，通常不適用于高并發或高性能場景。

3. 信號驅動IO

在信號驅動 I/O 模型中，進程發起一個 I/O 操作時，會向內核注冊一個信號處理函數（如 SIGIO），然后立即返回，不會被阻塞。當內核數據就緒時，會向進程發送一個信號，進程在信號處理函數中調用 I/O 操作（如 recvfrom）讀取數據。

主要特點：

• 非阻塞： 進程在等待數據時不會被阻塞，可以繼續執行其他任務。
• 回調機制： 通過信號通知的方式實現異步事件處理，數據就緒時內核主動通知進程。
• 實現難度大： 信號處理函數的編寫和調試較為復雜，開發難度較高。
• 信號處理復雜性： 信號處理函數需要處理異步事件，可能引入競態條件和不可預測的行為。

適用場景有限： 適合對實時性要求較高、但并發量較小的網絡應用開發。由于其實現復雜性和潛在問題，通常不適用于高并發或高性能場景。

4. 異步IO

在異步 I/O 模型中，當進程發起一個 I/O 操作時，會立即返回，不會被阻塞，也不會立即返回結果。內核會負責完成整個 I/O 操作（包括數據準備和復制到用戶空間），并在操作完成后通知進程。如果 I/O 操作成功，進程可以直接獲取到數據。

主要特點：

• 完全非阻塞： 進程在發起 I/O 操作后不會被阻塞，可以繼續執行其他任務。
• Proactor 模式： 內核負責完成 I/O 操作并通知進程，進程只需處理最終結果。
• 高性能： 適合高并發、高性能場景，能夠充分利用系統資源。
• 操作系統支持： 異步 I/O 需要操作系統的底層支持。在 Linux 中，異步 I/O 從 2.5 版本內核開始引入，并在 2.6 版本中成為標準特性。
• 實現難度大： 異步 I/O 的開發復雜度較高，需要處理回調、事件通知等機制。

適用場景： 異步 I/O 模型非常適合高性能、高并發的網絡應用開發，如大規模 Web 服務器、數據庫系統等。

5. IO復用模型

大多數文件系統的默認IO操作都是緩存IO。在Linux的緩存IO機制中，操作系統會將IO的數據緩存在文件系統的頁緩存（page cache）。也就是說，數據會先被拷貝到操作系統內核的緩沖區中，然后才會從操作系統內核的緩存區拷貝到應用程序的地址空間中。這種做法的缺點就是，需要在應用程序地址空間和內核進行多次拷貝，這些拷貝動作所帶來的CPU以及內存開銷是非常大的。 至于為什么不能直接讓磁盤控制器把數據送到應用程序的地址空間中呢？最簡單的一個原因就是應用程序不能直接操作底層硬件。 總的來說，IO分兩階段： 1)數據準備階段 2)內核空間復制回用戶進程緩沖區階段。如下圖：

三、I/O 多路復用之select、poll、epoll詳解

目前支持I/O多路復用的系統調用有select，pselect，poll，epoll。與多進程和多線程技術相比，I/O多路復用技術的最大優勢是系統開銷小，系統不必創建進程/線程，也不必維護這些進程/線程，從而大大減小了系統的開銷。 I/O多路復用就是通過一種機制，一個進程可以監視多個描述符，一旦某個描述符就緒（一般是讀就緒或者寫就緒），能夠通知程序進行相應的讀寫操作。但select，poll，epoll本質上都是同步I/O，因為他們都需要在讀寫事件就緒后自己負責進行讀寫，也就是說這個讀寫過程是阻塞的，而異步I/O則無需自己負責進行讀寫，異步I/O的實現會負責把數據從內核拷貝到用戶空間。

1. select

int select (int n, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout);

• readfds：內核檢測該集合中的IO是否可讀。如果想讓內核幫忙檢測某個IO是否可讀，需要手動把文件描述符加入該集合。
• writefds：內核檢測該集合中的IO是否可寫。同readfds，需要手動把文件描述符加入該集合。
• exceptfds：內核檢測該集合中的IO是否異常。同readfds，需要手動把文件描述符加入該集合。
• nfds：以上三個集合中最大的文件描述符數值 + 1，例如集合是{0,1,5,10}，那么 maxfd 就是 11
• timeout：用戶線程調用select的超時時長。
• 設置成NULL，表示如果沒有 I/O 事件發生，則 select 一直等待下去。
• 設置為非0的值，這個表示等待固定的一段時間后從 select 阻塞調用中返回。
• 設置成 0，表示根本不等待，檢測完畢立即返回。

函數返回值：

• 大于0：成功，返回集合中已就緒的IO總個數
• 等于-1：調用失敗
• 等于0：沒有就緒的IO

從上述的select函數聲明可以看出，fd_set本質是一個數組，為了方便我們操作該數組，操作系統提供了以下函數：

// 將文件描述符fd從set集合中刪除 
void FD_CLR(int fd, fd_set *set); 

// 判斷文件描述符fd是否在set集合中 
int  FD_ISSET(int fd, fd_set *set); 

// 將文件描述符fd添加到set集合中 
void FD_SET(int fd, fd_set *set); 

// 將set集合中, 所有文件描述符對應的標志位設置為0
void FD_ZERO(fd_set *set);

select 函數監視的文件描述符分3類，分別是writefds、readfds、和exceptfds，當用戶process調用select的時候，select會將需要監控的readfds集合拷貝到內核空間（假設監控的僅僅是socket可讀），然后遍歷自己監控的skb(SocketBuffer)，挨個調用skb的poll邏輯以便檢查該socket是否有可讀事件，遍歷完所有的skb后，如果沒有任何一個socket可讀，那么select會調用schedule_timeout進入schedule循環，使得process進入睡眠。如果在timeout時間內某個socket上有數據可讀了，或者等待timeout了，則調用select的process會被喚醒，接下來select就是遍歷監控的集合，挨個收集可讀事件并返回給用戶了，相應的偽碼如下：

int select(
    int nfds,
    fd_set *readfds,
    fd_set *writefds,
    fd_set *exceptfds,
    struct timeval *timeout);
// nfds:監控的文件描述符集里最大文件描述符加1
// readfds：監控有讀數據到達文件描述符集合，傳入傳出參數
// writefds：監控寫數據到達文件描述符集合，傳入傳出參數
// exceptfds：監控異常發生達文件描述符集合, 傳入傳出參數
// timeout：定時阻塞監控時間，3種情況
//  1.NULL，永遠等下去
//  2.設置timeval，等待固定時間
//  3.設置timeval里時間均為0，檢查描述字后立即返回，輪詢

/* 
* select服務端偽碼
* 首先一個線程不斷接受客戶端連接，并把socket文件描述符放到一個list里。
*/
while(1) {
  connfd = accept(listenfd);
  fcntl(connfd, F_SETFL, O_NONBLOCK);
  fdlist.add(connfd);
}
/*
* select函數還是返回剛剛提交的list，應用程序依然list所有的fd，只不過操作系統會將準備就緒的文件描述符做上標識，
* 用戶層將不會再有無意義的系統調用開銷。
*/
struct timeval timeout;
int max = 0;  // 用于記錄最大的fd，在輪詢中時刻更新即可
// 初始化比特位
FD_ZERO(&read_fd);
while (1) {
    // 阻塞獲取 每次需要把fd從用戶態拷貝到內核態
    nfds = select(max + 1, &read_fd, &write_fd, NULL, &timeout);
    // 每次需要遍歷所有fd，判斷有無讀寫事件發生
    for (int i = 0; i <= max && nfds; ++i) {
        // 只讀已就緒的文件描述符，不用過多遍歷
        if (i == listenfd) {
            // 這里處理accept事件
            FD_SET(i, &read_fd);//將客戶端socket加入到集合中
        }
        if (FD_ISSET(i, &read_fd)) {
            // 這里處理read事件
        }
    }
}

下面的動圖能更直觀的讓我們了解select：

通過上面的select邏輯過程分析，相信大家都意識到，select存在三個問題：

(1) 每次調用select，都需要把被監控的fds集合從用戶態空間拷貝到內核態空間，高并發場景下這樣的拷貝會使得消耗的資源是很大的。

(2) 能監聽端口的數量有限，單個進程所能打開的最大連接數由FD_SETSIZE宏定義，監聽上限就等于fds_bits位數組中所有元素的二進制位總數，其大小是32個整數的大小(在32位的機器上，大小就是3232，同理64位機器上為3264)，當然我們可以對宏FD_SETSIZE進行修改，然后重新編譯內核，但是性能可能會受到影響，一般該數和系統內存關系很大，具體數目可以cat /proc/sys/fs/file-max察看。32位機默認1024個，64位默認2048。

(3) 被監控的fds集合中，只要有一個有數據可讀，整個socket集合就會被遍歷一次調用sk的poll函數收集可讀事件：由于當初的需求是樸素，僅僅關心是否有數據可讀這樣一個事件，當事件通知來的時候，由于數據的到來是異步的，我們不知道事件來的時候，有多少個被監控的socket有數據可讀了，于是，只能挨個遍歷每個socket來收集可讀事件了。

2. poll

poll 的實現與 select 非常相似，都是通過監視多個文件描述符（fd）來實現 I/O 多路復用。兩者的主要區別在于描述 fd 集合的方式：select 使用 fd_set 結構，而 poll 使用 pollfd 結構。select 的 fd_set 結構限制了 fd 集合的大小（通常為 1024），而 poll 使用 pollfd 結構，理論上可以支持更多的 fd，解決了 select 的問題 (2)。

與 select 類似，poll 也存在性能瓶頸。當監視的 fd 數量較多時，poll 需要將整個 pollfd 數組在用戶態和內核態之間復制，無論這些 fd 是否就緒。這種復制的開銷會隨著 fd 數量的增加而線性增長，導致性能下降。

poll 適合需要監視較多 fd 的場景，但在高并發或 fd 數量非常大的情況下，性能仍然不如 epoll。

解決 fd 數量限制問題：

struct pollfd {
   int fd;           /*文件描述符*/
   short events;     /*監控的事件*/
   short revents;    /*監控事件中滿足條件返回的事件*/
};
int poll(struct pollfd *fds, unsigned long nfds, int timeout);

函數參數：

• fds：struct pollfd類型的數組, 存儲了待檢測的文件描述符，struct pollfd有三個成員：
• fd：委托內核檢測的文件描述符
• events：委托內核檢測的fd事件（輸入、輸出、錯誤），每一個事件有多個取值
• revents：這是一個傳出參數，數據由內核寫入，存儲內核檢測之后的結果
• nfds：描述的是數組 fds 的大小
• timeout: 指定poll函數的阻塞時長
• -1：一直阻塞，直到檢測的集合中有就緒的IO事件，然后解除阻塞函數返回
• 0：不阻塞，不管檢測集合中有沒有已就緒的IO事件，函數馬上返回
• 大于0：表示 poll 調用方等待指定的毫秒數后返回

函數返回值：

• -1：失敗
• 大于0：表示檢測的集合中已就緒的文件描述符的總個數

下面是poll的函數原型，poll改變了fds集合的描述方式，使用了pollfd結構而不是select的fd_set結構，使得poll支持的fds集合限制遠大于select的1024。poll雖然解決了fds集合大小1024的限制問題，從實現來看。很明顯它并沒優化大量描述符數組被整體復制于用戶態和內核態的地址空間之間，以及個別描述符就緒觸發整體描述符集合的遍歷的低效問題。poll隨著監控的socket集合的增加性能線性下降，使得poll也并不適合用于大并發場景。

poll服務端實現偽碼：
struct pollfd fds[POLL_LEN];
unsigned int nfds=0;
fds[0].fd=server_sockfd;
fds[0].events=POLLIN|POLLPRI;
nfds++;
while {
    res=poll(fds,nfds,-1);
    if(fds[0].revents&(POLLIN|POLLPRI)) {
        //執行accept并加入fds中，nfds++
        if(--res<=0) continue
    }
    //循環之后的fds
    if(fds[i].revents&(POLLIN|POLLERR )) {
        //讀操作或處理異常等
        if(--res<=0) continue
    }
}

3. epoll

在 Linux 網絡編程中，select 曾長期被用作事件觸發的機制。然而，隨著高并發場景的需求增加，select 的性能瓶頸逐漸顯現。為了解決這些問題，Linux 內核引入了 epoll 機制。相比于 select，epoll 的最大優勢在于其性能不會隨著監聽的文件描述符（fd）數量的增加而顯著下降。如前面我們所說，在內核中的select實現中，它是采用輪詢來處理的，輪詢的fd數目越多，自然耗時越多。并且，在linux/posix_types.h頭文件有這樣的聲明：

#define __FD_SETSIZE 1024 (select 最多只能同時監聽 1024 個 fd（由 __FD_SETSIZE 定義）。雖然可以通過修改內核頭文件并重新編譯內核來擴大這一限制，但這并不能從根本上解決問題。) 而epoll 使用基于事件回調的機制，而不是輪詢。它只會關注活躍的 fd，因此性能不會隨著 fd 數量的增加而顯著下降。

epoll 的使用： epoll的接口非常簡單，一共就三個函數：

• epoll_create創建句柄：使用 epoll_create 創建一個 epoll 句柄。參數 size 用于告訴內核監聽 fd 的數量（在較新的內核中，size 參數已被忽略，但仍需大于 0），這個參數不同于select()中的第一個參數，給出最大監聽的fd+1的值。需要注意的是，當創建好epoll句柄后，它就是會占用一個fd值，在linux下如果查看/proc/進程id/fd/，是能夠看到這個fd的，所以在使用完epoll后，必須調用close()關閉，否則可能導致fd被耗盡。
• epoll_ctl管理連接：使用 epoll_ctl 向 epoll 對象中添加、修改或刪除要管理的 fd。
• epoll_wait等待事件：使用 epoll_wait 等待其管理的 fd 上的 I/O 事件。

(1) epoll_create 函數

int epoll_create(int size);

• 功能：該函數生成一個 epoll 專用的文件描述符。
• 參數size: 用來告訴內核這個監聽的數目一共有多大，參數 size 并不是限制了 epoll 所能監聽的描述符最大個數，只是對內核初始分配內部數據結構的一個建議。自從 linux 2.6.8 之后，size 參數是被忽略的，也就是說可以填只有大于 0 的任意值。
• 返回值：如果成功，返回poll 專用的文件描述符，否者失敗，返回-1。

epoll_create的源碼實現：

SYSCALL_DEFINE1(epoll_create1, int, flags)
{
    struct eventpoll *ep = NULL;

    //創建一個 eventpoll 對象
    error = ep_alloc(&ep);
}

//struct eventpoll 的定義
// file：fs/eventpoll.c
struct eventpoll {

    //sys_epoll_wait用到的等待隊列
    wait_queue_head_t wq;

    //接收就緒的描述符都會放到這里
    struct list_head rdllist;

    //每個epoll對象中都有一顆紅黑樹
    struct rb_root rbr;

    ......
}
static int ep_alloc(struct eventpoll **pep)
{
    struct eventpoll *ep;

    //申請 epollevent 內存
    ep = kzalloc(sizeof(*ep), GFP_KERNEL);

    //初始化等待隊列頭
    init_waitqueue_head(&ep->wq);

    //初始化就緒列表
    INIT_LIST_HEAD(&ep->rdllist);

    //初始化紅黑樹指針
    ep->rbr = RB_ROOT;

    ......
}

其中eventpoll 這個結構體中的幾個成員的含義如下：

• wq： 等待隊列鏈表。軟中斷數據就緒的時候會通過 wq 來找到阻塞在 epoll 對象上的用戶進程。
• rbr： 紅黑樹。為了支持對海量連接的高效查找、插入和刪除，eventpoll 內部使用的就是紅黑樹。通過紅黑樹來管理用戶主進程accept添加進來的所有 socket 連接。
• rdllist： 就緒的描述符鏈表。當有連接就緒的時候，內核會把就緒的連接放到 rdllist 鏈表里。這樣應用進程只需要判斷鏈表就能找出就緒進程，而不用去遍歷紅黑樹的所有節點了。

(2) epoll_ctl 函數

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

• 功能：epoll 的事件注冊函數，它不同于 select() 是在監聽事件時告訴內核要監聽什么類型的事件，而是在這里先注冊要監聽的事件類型。
• 參數epfd: epoll 專用的文件描述符，epoll_create()的返回值
• 參數op: 表示動作，用三個宏來表示：

1. EPOLL_CTL_ADD：注冊新的 fd 到 epfd 中；
2. EPOLL_CTL_MOD：修改已經注冊的fd的監聽事件；
3. EPOLL_CTL_DEL：從 epfd 中刪除一個 fd；

- 參數fd: 需要監聽的文件描述符
- 參數event: 告訴內核要監聽什么事件，struct epoll_event 結構如:
- events 可以是以下幾個宏的集合：
- EPOLLIN ：表示對應的文件描述符可以讀（包括對端 SOCKET 正常關閉）；
- EPOLLOUT：表示對應的文件描述符可以寫；
- EPOLLPRI：表示對應的文件描述符有緊急的數據可讀（這里應該表示有帶外數據到來）；
- EPOLLERR：表示對應的文件描述符發生錯誤；
- EPOLLHUP：表示對應的文件描述符被掛斷；
- EPOLLET ：將 EPOLL 設為邊緣觸發(Edge Trigger)模式，這是相對于水平觸發(Level Trigger)來說的。
- EPOLLONESHOT：只監聽一次事件，當監聽完這次事件之后，如果還需要繼續監聽這個 socket 的話，需要再次把這個 socket 加入到 EPOLL 隊列里
- 返回值：0表示成功，-1表示失敗。

(3) epoll_wait函數

int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);

• 功能：等待事件的產生，收集在 epoll 監控的事件中已經發送的事件，類似于 select() 調用。
• 參數epfd: epoll 專用的文件描述符，epoll_create()的返回值
• 參數events: 分配好的 epoll_event 結構體數組，epoll 將會把發生的事件賦值到events 數組中（events 不可以是空指針，內核只負責把數據復制到這個 events 數組中，不會去幫助我們在用戶態中分配內存）。
• 參數maxevents: maxevents 告之內核這個 events 有多少個 。
• 參數timeout: 超時時間，單位為毫秒，為 -1 時，函數為阻塞。

返回值：

• 如果成功，表示返回需要處理的事件數目
• 如果返回0，表示已超時
• 如果返回-1，表示失敗

#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <stdio.h>
#include <unistd.h>
#include <errno.h>
#include <fcntl.h>
#include <stdlib.h>
#include <cassert>
#include <sys/epoll.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <string.h>
#include<iostream>
const int MAX_EVENT_NUMBER = 10000; //最大事件數
// 設置句柄非阻塞
int setnonblocking(int fd)
{
    int old_option = fcntl(fd, F_GETFL);
    int new_option = old_option | O_NONBLOCK;
    fcntl(fd, F_SETFL, new_option);
    return old_option;
}

int main(){

    // 創建套接字
    int nRet=0;
    int m_listenfd = socket(PF_INET, SOCK_STREAM, 0);
    if(m_listenfd<0)
    {
        printf("fail to socket!");
        return -1;
    }
    // 
    struct sockaddr_in address;
    bzero(&address, sizeof(address));
    address.sin_family = AF_INET;
    address.sin_addr.s_addr = htonl(INADDR_ANY);
    address.sin_port = htons(6666);

    int flag = 1;
    // 設置ip可重用
    setsockopt(m_listenfd, SOL_SOCKET, SO_REUSEADDR, &flag, sizeof(flag));
    // 綁定端口號
    int ret = bind(m_listenfd, (struct sockaddr *)&address, sizeof(address));
    if(ret<0)
    {
        printf("fail to bind!,errno :%d",errno);
        return ret;
    }

    // 監聽連接fd
    ret = listen(m_listenfd, 200);
    if(ret<0)
    {
        printf("fail to listen!,errno :%d",errno);
        return ret;
    }

    // 初始化紅黑樹和事件鏈表結構rdlist結構
    epoll_event events[MAX_EVENT_NUMBER];
    // 創建epoll實例
    int m_epollfd = epoll_create(5);
    if(m_epollfd==-1)
    {
        printf("fail to epoll create!");
        return m_epollfd;
    }


    // 創建節點結構體將監聽連接句柄
    epoll_event event;
    event.data.fd = m_listenfd;
    //設置該句柄為邊緣觸發（數據沒處理完后續不會再觸發事件，水平觸發是不管數據有沒有觸發都返回事件），
    event.events = EPOLLIN | EPOLLET | EPOLLRDHUP;
    // 添加監聽連接句柄作為初始節點進入紅黑樹結構中，該節點后續處理連接的句柄
    epoll_ctl(m_epollfd, EPOLL_CTL_ADD, m_listenfd, &event);

    //進入服務器循環
    while(1)
    {
        int number = epoll_wait(m_epollfd, events, MAX_EVENT_NUMBER, -1);
        if (number < 0 && errno != EINTR)
        {
            printf( "epoll failure");
            break;
        }
        for (int i = 0; i < number; i++)
        {
            int sockfd = events[i].data.fd;
            // 屬于處理新到的客戶連接
            if (sockfd == m_listenfd)
            {
                struct sockaddr_in client_address;
                socklen_t client_addrlength = sizeof(client_address);
                int connfd = accept(m_listenfd, (struct sockaddr *)&client_address, &client_addrlength);
                if (connfd < 0)
                {
                    printf("errno is:%d accept error", errno);
                    return false;
                }
                epoll_event event;
                event.data.fd = connfd;
                //設置該句柄為邊緣觸發（數據沒處理完后續不會再觸發事件，水平觸發是不管數據有沒有觸發都返回事件），
                event.events = EPOLLIN | EPOLLRDHUP;
                // 添加監聽連接句柄作為初始節點進入紅黑樹結構中，該節點后續處理連接的句柄
                epoll_ctl(m_epollfd, EPOLL_CTL_ADD, connfd, &event);
                setnonblocking(connfd);
            }
            else if (events[i].events & (EPOLLRDHUP | EPOLLHUP | EPOLLERR))
            {
                //服務器端關閉連接，
                epoll_ctl(m_epollfd, EPOLL_CTL_DEL, sockfd, 0);
                close(sockfd);
            }
            //處理客戶連接上接收到的數據
            else if (events[i].events & EPOLLIN)
            {
                char buf[1024]={0};
                read(sockfd,buf,1024);
                printf("from client :%s");

                // 將事件設置為寫事件返回數據給客戶端
                events[i].data.fd = sockfd;
                events[i].events = EPOLLOUT | EPOLLET | EPOLLONESHOT | EPOLLRDHUP;
                epoll_ctl(m_epollfd, EPOLL_CTL_MOD, sockfd, &events[i]);
            }
            else if (events[i].events & EPOLLOUT)
            {
                std::string response = "server response \n";
                write(sockfd,response.c_str(),response.length());

                // 將事件設置為讀事件，繼續監聽客戶端
                events[i].data.fd = sockfd;
                events[i].events = EPOLLIN | EPOLLRDHUP;
                epoll_ctl(m_epollfd, EPOLL_CTL_MOD, sockfd, &events[i]);
            }
            //else if 可以加管道，unix套接字等等數據
        }
    }


}

如下圖，可以幫助我們理解的更加絲滑(/手動狗頭)：

4. epoll的邊緣觸發與水平觸發

(1) 水平觸發（LT）

關注點是數據是否有無，只要讀緩沖區不為空，寫緩沖區不滿，那么epoll_wait就會一直返回就緒，水平觸發是epoll的默認工作方式。適合對事件處理邏輯要求不高的場景。

(2) 邊緣觸發（ET）

關注點是數據的變化。只有當緩沖區狀態發生變化時（例如從空變為非空，或從非空變為空），epoll_wait 才會返回就緒狀態。這里的數據變化并不單純指緩沖區從有數據變為沒有數據，或者從沒有數據變為有數據，還包括了數據變多或者變少。即當buffer長度有變化時，就會觸發。 假設epoll被設置為了邊緣觸發，當客戶端寫入了100個字符，由于緩沖區從0變為了100，于是服務端epoll_wait觸發一次就緒，服務端讀取了2個字節后不再讀取。這個時候再去調用epoll_wait會發現不會就緒，只有當客戶端再次寫入數據后，才會觸發就緒。 這就導致如果使用ET模式，那就必須保證要「一次性把數據讀取&寫入完」，否則會導致數據長期無法讀取/寫入。適合高性能場景，可以減少事件通知的次數，提高效率。

4. epoll 為什么比select，poll更高效？

從上圖可以看出，epoll使用紅黑樹管理文件描述符，紅黑樹插入和刪除的都是時間復雜度 O(logN)，不會隨著文件描述符數量增加而改變。 select、poll采用數組或者鏈表的形式管理文件描述符，那么在遍歷文件描述符時，時間復雜度會隨著文件描述的增加而增加，我們從以下幾點分析epoll的優勢：

(1) 事件驅動機制（基于回調，而非輪詢）

• select 和 poll 的輪詢機制： select 和 poll 采用輪詢的方式檢查所有被監視的文件描述符（fd），無論這些 fd 是否就緒。每次調用時，都需要將整個 fd 集合從用戶態復制到內核態，并在內核中遍歷所有 fd 來檢查其狀態。隨著 fd 數量的增加，輪詢的開銷會線性增長，導致性能顯著下降。
• epoll 的事件驅動機制：- epoll 使用基于事件回調的機制。內核會維護一個就緒隊列，只關注那些狀態發生變化的 fd（即活躍的 fd）。一旦檢測到epoll管理的socket描述符就緒時，內核會采用類似 callback 的回調機制，將其加入就緒隊列，epoll_wait 只需從隊列中獲取就緒的 fd，而不需要遍歷所有 fd。這種機制使得 epoll 的性能不會隨著 fd 數量的增加而顯著下降。

(2) 避免頻繁的用戶態與內核態數據拷貝

• select 和 poll 的數據拷貝問題： 每次調用 select 或 poll 時，都需要將整個 fd 集合從用戶態復制到內核態，調用結束后再將結果從內核態復制回用戶態。這種頻繁的數據拷貝在高并發場景下會帶來較大的性能開銷。
• epoll 的優化： epoll 使用了內存映射（ mmap ）技術，這樣便徹底省掉了這些socket描述符在系統調用時拷貝的開銷（因為從用戶空間到內核空間需要拷貝操作）。mmap將用戶空間的一塊地址和內核空間的一塊地址同時映射到相同的一塊物理內存地址（不管是用戶空間還是內核空間都是虛擬地址，最終要通過地址映射映射到物理地址），使得這塊物理內存對內核和對用戶均可見，減少用戶態和內核態之間的數據交換，不需要依賴拷貝，這樣子內核可以直接看到epoll監聽的socket描述符，效率極高。

(3) 支持更大的并發連接數

• select 的 fd 數量限制： select 使用 fd_set 結構，其大小通常被限制為 1024（由 __FD_SETSIZE 定義），這意味著它最多只能同時監視 1024 個 fd。雖然可以通過修改內核頭文件并重新編譯內核來擴大這一限制，但這并不能從根本上解決問題。
• poll 的改進與局限： poll 使用 pollfd 結構，理論上可以支持更多的 fd，但它仍然需要遍歷所有 fd，性能會隨著 fd 數量的增加而下降。
• epoll 的無限制支持： epoll 沒有 fd 數量的硬性限制，適合高并發場景，能夠輕松支持數萬甚至數十萬的并發連接。

四、總結

select，poll，epoll都是IO多路復用機制，即可以監視多個描述符，一旦某個描述符就緒（讀或寫就緒），能夠通知程序進行相應讀寫操作。 但select，poll，epoll本質上都是同步I/O，因為他們都需要在讀寫事件就緒后自己負責進行讀寫，也就是說這個讀寫過程是阻塞的，而異步I/O則無需自己負責進行讀寫，異步I/O的實現會負責把數據從內核拷貝到用戶空間。

• select，poll實現需要自己不斷輪詢所有fd集合，直到設備就緒，期間可能要睡眠和喚醒多次交替。而epoll其實也需要調用epoll_wait不斷輪詢就緒鏈表，期間也可能多次睡眠和喚醒交替，但是它是設備就緒時，調用回調函數，把就緒fd放入就緒鏈表中，并喚醒在epoll_wait中進入睡眠的進程。雖然都要睡眠和交替，但是select和poll在“醒著”的時候要遍歷整個fd集合，而epoll在“醒著”的時候只要判斷一下就緒鏈表是否為空就行了，這節省了大量的CPU時間。這就是回調機制帶來的性能提升。
• select，poll每次調用都要把fd集合從用戶態往內核態拷貝一次，并且要把current往設備等待隊列中掛一次，而epoll只要一次拷貝，而且把current往等待隊列上掛也只掛一次（在epoll_wait的開始，注意這里的等待隊列并不是設備等待隊列，只是一個epoll內部定義的等待隊列）。這也能節省不少的開銷。