libevent和基於libevent的網絡編程

時間 2019-11-10

原文原文鏈接

1 libevent介紹和安裝

介紹

libevent是一個輕量級的基於事件驅動的高性能的開源網絡庫，而且支持多個平臺，對多個平臺的I/O複用技術進行了封裝，當咱們編譯庫的代碼時，編譯的腳本將會根據OS支持的處理事件機制，來編譯相應的代碼，從而在libevent接口上保持一致。html

在當前的服務器上，面對的主要問題就是要能處理大量的鏈接。而經過libevent這個網絡庫，咱們就能夠調用它的API來很好的解決上面的問題。首先，能夠來回顧一下，對這個問題的傳統解決方法。linux

問題： 如何處理多個客戶端鏈接git

解決方案1：I/O複用技術github

這幾種方式都是同步I/O，即當讀寫事件就緒，他們本身須要負責進行讀寫，這個讀寫過程是阻塞的，而異步I/O則不須要本身負責讀寫，只須要通知負責讀寫的程序就能夠了。數據庫

循環
假設當前我服務器有多個網絡鏈接須要看管，那麼我就循環遍歷打開的網絡鏈接的列表，來判斷是否有要讀取的數據。這種方法的缺點很明顯，那就是 1.速度緩慢（必須遍歷全部的網絡鏈接） 2.效率低（處理一個鏈接時可能發生阻塞，妨礙其餘網絡鏈接的檢查和處理）編程
select方式
select對應於內核中的sys_select調用，sys_select首先將第二三四個參數指向的fd_set拷貝到內核，而後對每一個被SET的描述符調用進行poll，並記錄在臨時結果中（fdset），若是有事件發生，select會將臨時結果寫到用戶空間並返回；當輪詢一遍後沒有任何事件發生時，若是指定了超時時間，則select會睡眠到超時，睡眠結束後再進行一次輪詢，並將臨時結果寫到用戶空間，而後返回。
select返回後，須要逐一檢查關注的描述符是否被SET（事件是否發生）。（select支持的文件描述符數量過小了，默認是1024）。api
poll方式
poll與select不一樣，經過一個pollfd數組向內核傳遞須要關注的事件，故沒有描述符個數的限制，pollfd中的events字段和revents分別用於標示關注的事件和發生的事件，故pollfd數組只須要被初始化一次。
poll的實現機制與select相似，其對應內核中的sys_poll，只不過poll向內核傳遞pollfd數組，而後對pollfd中的每一個描述符進行poll，相比處理fdset來講，poll效率更高。
poll返回後，須要對pollfd中的每一個元素檢查其revents值，來得指事件是否發生。數組
epoll方式
epoll經過epoll_create建立一個用於epoll輪詢的描述符，經過epoll_ctl添加/修改/刪除事件，經過epoll_wait檢查事件，epoll_wait的第二個參數用於存放結果。
epoll與select、poll不一樣，首先，其不用每次調用都向內核拷貝事件描述信息，在第一次調用後，事件信息就會與對應的epoll描述符關聯起來。其次，epoll不是經過輪詢，而是經過在等待的描述符上註冊回調函數，當事件發生時，回調函數負責把發生的事件存儲在就緒事件鏈表中，最後寫到用戶空間。
epoll返回後，該參數指向的緩衝區中即爲發生的事件，對緩衝區中每一個元素進行處理便可，而不須要像poll、select那樣進行輪詢檢查。服務器

解決方案2：多線程技術或多進程技術網絡

多線程技術和多進程技術也能夠處理高併發的數據鏈接，由於在服務器中能夠產生大量的進程和線程和處理咱們須要監視的鏈接。可是，這兩種方式也是有很大的侷限性的，好比多進程模型就不適合大量的短鏈接，由於進程的產生和關閉須要消耗較大的系統性能，一樣，還要進程進程間的通訊，在CPU性能不足的狀況下不太適合。而多線程技術則不太適合處理長鏈接，由於當咱們創建一個進程時，linux中會消耗8G的棧空間，若是咱們的每一個鏈接都杵着不斷開，那麼大量鏈接長鏈接後，致使的結果就是內存的大量消耗。

解決方案3：經常使用的上述兩者複合使用
上述的兩種方法各具備優缺點，所以，咱們能夠將上述的方法結合起來，這也是目前使用較多的處理高併發的方法。多進程+I/O複用或者多線程+I/O複用。而在具體的實現上，又能夠分爲不少的方式。好比多線程+I/O複用技術，咱們使用使用一個主線程負責監聽一個端口和接受的描述符是否有讀寫事件產生，若是有，則將事件分發給其餘的工做進程去完成，這也是進程池的理念。

在說完上述的高併發的處理方法以後，咱們能夠來介紹一個libevent的主要特點了。

一樣，lievent也是採用的上述系統提供的select，poll和epoll方法來進行I/O複用，可是針對於多個系統平臺上的不一樣的I/O複用實現方式，libevent進行了從新的封裝，並提供了統一的API接口。libevent在實現上使用了事件驅動這種機制，其本質上是一種Reactor模式。

Reactor模式，是一種事件驅動機制。應用程序須要提供相應的接口並註冊到Reactor上，若是相應的事件發生，Reactor將主動調用應用程序註冊的接口，這些接口又稱爲「回調函數」。

在Libevent中也是同樣，向Libevent框架註冊相應的事件和回調函數；當這些事件發生時，Libevent會調用這些回調函數處理相應的事件。

lbevent的事件支持三種，分別是網絡IO、定時器和信號。定時器的數據結構使用最小堆（Min Heap），以提升效率。網絡IO和信號的數據結構採用了雙向鏈表(TAILQ)。

安裝

libevent的安裝很簡單，我是直接從github上clone下一個源碼，而後進行編譯安裝的。

具體的命令是（假設你已經安裝了git）：

# git clone https://github.com/nmathewson/Libevent.git
  # cd Libevent
  # sh autogen.sh
  # ./configure && make
  # make install
  # make verify  //驗證安裝

2 Linux下libevent主要API介紹

如今的libevent版本已經到達libevent2了，其增長了多線程的支持，API函數也發生了一些微小的變化。

建立事件集

struct event_base *event_base_new(void)
建立事件

struct event event_new(struct event_base ,evutil_socket_t ,short ,event_callback_fn,void*)

參數一：事件所在的事件集。
參數二：socket的描述符。
參數三：事件類型，其中EV_READ表示等待讀事件發生，EV_WRITE表示寫事件發生，或者它倆的組合，EV_SIGNAL表示須要等待事件的號碼，如果不包含上述的標誌，就是超時事件或者手動激活的事件。
參數四：事件發生時須要調用的回調函數。
參數五：回調函數的參數值。
添加事件和刪除事件

int event_add(struct event * ev,const struct timeval* timeout)

參數一：須要添加的事件
參數二：事件的最大等待事件，若是是NULL的話，就是永久等待

int event_del(struct event *)
參數一：須要刪除的事件
分配監聽事件

int event_base_dispatch(struct event_base * )

參數一：須要監視的事件集
I/O buffer事件

struct bufferevent* bufferevent_socket_new
(struct event_base * base,evutil_socket_t fd,int options)

參數一：須要添加到的時間集
參數二：相關的文件描述符
參數三：0或者是相應的BEV_OPT_*可選標誌

int bufferevent_enable(struct bufferevent * bev,short event)

參數一：須要啓用的bufferevent
參數二：any combination of EV|READ | EV_WRITE

int bufferevent_disable(struct bufferevent * bev,short event)

參數說明：同上

size_t bufferevent_read(struct bufferevent bev,void data,size_t size)

參數一：讀取的buffer_event事件
參數二：存儲數據的指針
參數三：數據buffer的大小

返回值：讀取數據的字節數

int bufferevent_write(struct bufferevent bev,const void data,size_t size)

參數一：讀取的buffer_event事件
參數二：存儲數據的指針
參數三：要寫入的數據的大小，字節數

若是你想知道更多的API使用狀況，請點擊這裏。

3.1 編程實例之聊天室服務器

下面，就基於libevent2編寫一個聊天室服務器。

設計思想：首先建立一個套接字，進而建立一個事件對此端口進行監聽，將所請求的用戶組成一個隊列，並監聽全部的用戶事件，當某個用戶說話了，產生了讀事件，就將該用戶的發言發送給隊列中的其餘用戶。

程序分析

須要包含的libevent函數頭：

#include <event2/event.h>
#include <event2/event_struct.h>
#include <event2/bufferevent.h>
#include <event2/buffer.h>

建立一個client結構體，接受鏈接後存放數據：

struct client {
/* The clients socket. */
    int fd;

/* The bufferedevent for this client. */
struct bufferevent *buf_ev;
    struct bufferevent *buf_ev;
/*
 * This holds the pointers to the next and previous entries in
 * the tail queue.
 */
    TAILQ_ENTRY(client) entries;
};

先來看下mian函數的處理：

int
main(int argc, char **argv)
{
    int listen_fd;
    struct sockaddr_in listen_addr;
    struct event ev_accept;
    int reuseaddr_on;

    /* Initialize libevent. */
    evbase = event_base_new();

    /* Initialize the tailq. */
    TAILQ_INIT(&client_tailq_head);

    /* Create our listening socket. */
    listen_fd = socket(AF_INET, SOCK_STREAM, 0);
    if (listen_fd < 0)
        err(1, "listen failed");
    memset(&listen_addr, 0, sizeof(listen_addr));
    listen_addr.sin_family = AF_INET;
    listen_addr.sin_addr.s_addr = INADDR_ANY;
    listen_addr.sin_port = htons(SERVER_PORT);
    
    if (bind(listen_fd, (struct sockaddr *)&listen_addr,
    sizeof(listen_addr)) < 0)
    err(1, "bind failed");
    if (listen(listen_fd, 5) < 0)
    err(1, "listen failed");
    reuseaddr_on = 1;
    setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &reuseaddr_on, 
    sizeof(reuseaddr_on));

    /* Set the socket to non-blocking, this is essential in event
     * based programming with libevent. */
    if (setnonblock(listen_fd) < 0)
    err(1, "failed to set server socket to non-blocking");

    /* We now have a listening socket, we create a read event to
    * be notified when a client connects. */
    event_assign(&ev_accept, evbase, listen_fd, EV_READ|EV_PERSIST, 
    on_accept, NULL);
    event_add(&ev_accept, NULL);

    /* Start the event loop. */
    event_base_dispatch(evbase);

    return 0;
}

首先，函數初始化了一個用戶隊列tailq，接着建立了一個socket套接字，並將套接字設定爲非阻塞模式，接着對一個全局的evbase事件集合，註冊了事件，事件源是listen_fd，回調函數是on_accept，事件發生的狀況是EV_READ,並且標誌EV_PESIST代表該事件一直存在，然後開啓事件掃描循環event_base_dispatch(evbase)。

再看一下回調函數on_accpet實現：

void
on_accept(int fd, short ev, void *arg)
{
    int client_fd;
    struct sockaddr_in client_addr;
    socklen_t client_len = sizeof(client_addr);
    struct client *client;

    client_fd = accept(fd, (struct sockaddr *)&client_addr, &client_len);
    if (client_fd < 0) {
        warn("accept failed");
        return;
    }

    /* Set the client socket to non-blocking mode. */
    if (setnonblock(client_fd) < 0)
        warn("failed to set client socket non-blocking");

    /* We've accepted a new client, create a client object. */
    client = calloc(1, sizeof(*client));
    if (client == NULL)
        err(1, "malloc failed");
    client->fd = client_fd;

    client->buf_ev = bufferevent_socket_new(evbase, client_fd, 0);
    bufferevent_setcb(client->buf_ev, buffered_on_read, NULL,
        buffered_on_error, client);

    /* We have to enable it before our callbacks will be
    * called. */
    bufferevent_enable(client->buf_ev, EV_READ);

    /* Add the new client to the tailq. */
    TAILQ_INSERT_TAIL(&client_tailq_head, client, entries);

    printf("Accepted connection from %s\n", 
        inet_ntoa(client_addr.sin_addr));
}

這個回調函數的做用很顯然，就是接受了一個客戶端的請求，並申請好了一個client信息，將須要的內容填寫好，在填寫中須要注意的是，又向上述的事件集evbase中註冊了一個bufferevent事件client->buf_ev,並註冊了回調函數buffered_on_read,buffered_on_error，這三個函數分別是當接受後的鏈接發生了讀或者錯誤事件後的執行函數。接着，將用戶的client結構放入了用戶的隊列tailq中去。

用戶的buffer可讀後的執行函數：

void
buffered_on_read(struct bufferevent *bev, void *arg)
{
    struct client *this_client = arg;
    struct client *client;
    uint8_t data[8192];
    size_t n;

    /* Read 8k at a time and send it to all connected clients. */
    for (;;) {
        n = bufferevent_read(bev, data, sizeof(data));
        if (n <= 0) {
            /* Done. */
            break;
        }
    
        /* Send data to all connected clients except for the
         * client that sent the data. */

        TAILQ_FOREACH(client, &client_tailq_head, entries) {
            if (client != this_client) {
                bufferevent_write(client->buf_ev, data,  n);
            }
        }

    }

}

執行函數的做用很明顯，將libevent管理中的buffer數據讀取出，存入本地的data數組內，而後對隊列中的client進行檢索，若是不是發數據的client,則將數據寫入該client的buffer中，發送給該用戶。這裏注意的是須要反覆讀取buffer中的數據，防止一個讀取並無讀取乾淨，直到讀取不到數據爲止。

buffer出錯處理函數和上述函數差很少，功能就是出錯後，結束掉保存的client結構，詳細就不說了。

程序源碼： chat-server.c Makefile

編譯的時候記得修改Makefile中Libevent文件夾的位置

3.2 編程實例之回顯服務器（純異步IO）

設計思想：所謂回顯服務器就是將客戶端發過來的數據再發回去，這裏主要也就是說明libevent的純IO複用實現。實現方法和上面的差很少，甚至能夠說更加簡單。

程序和上面的聊天服務器差很少，只是在buffer可讀的事件函數中，不是將用戶的數據發送給其餘用戶，而是直接發送給用戶自己。

程序源碼： libevent_echosrv_buffered.c Makefile

3.3 編程實例之回顯服務器（多線程--per connection per thread）

設計思想：上面的方法單純使用libevent的簡單函數來實現服務，可是這裏，咱們假設咱們須要處理的客戶端不多，因而咱們可使用對於每一個鏈接咱們分配一個線程這樣的方式來實現對用戶的服務。這種方式簡單有效，一對一服務，就算業務邏輯出現阻塞也不怕。

程序分析

首先定義了一些數據結構，worker數據結構定義的是一個工做者，它包含有一個工做線程，和結束標誌，須要獲取的工做隊列，和創建鏈表須要的指針。job數據結構定義的是操做一個job的方法和對象，這回到程序中，實際上就是指的是事件發生後，封裝好的client結構體和處理這個結構體的方法。workqueue數據結構指的是當前的工做隊列中的工做者，以及工做隊列中的待完成的工做，以及互斥鎖和條件變量（由於多個工做進程須要訪問這些資源）。

具體的流程就是，用一個主線程監聽一個套接字，並將套接字接受到的鏈接accept，並建立一個client數據結構保存該鏈接的信息，在這個client結構中註冊一個bufferevent事件，註冊到client->evbase上（這時候這是向client中的evbase註冊了一個事件尚未進行循環這個事件集）。

接着，當監聽到某個client有bufferevent事件發生，主線程就把該client結構體和須要進行的工做方法包裝成一個job結構，而後把這個job扔到workqueue上去，並通知各個工做者。然後，各個工做者開着的線程就被激活了，瘋狂地去workqueue上去搶工做作，某個worker拿到工做後，就能夠解包job，根據job的工做說明書（job_function）操做工做對象（client）了。這裏，job的工做說明有是循環client中的client->evbase，因而這樣線程就會一直去監視這個鏈接的狀態，若是有數據就這會調用回調函數進行處理。同時，這個線程也就是阻塞在這裏，這對這一個鏈接負責。

創建workqueue須要的結構體和函數有：

typedef struct worker {
    pthread_t thread;
    int terminate;
    struct workqueue *workqueue;
    struct worker *prev;
    struct worker *next;
} worker_t;

typedef struct job {
    void (*job_function)(struct job *job);
    void *user_data;
    struct job *prev;
    struct job *next;
} job_t;

typedef struct workqueue {
    struct worker *workers;
    struct job *waiting_jobs;
    pthread_mutex_t jobs_mutex;
    pthread_cond_t jobs_cond;
} workqueue_t;

int workqueue_init(workqueue_t *workqueue, int numWorkers);

void workqueue_shutdown(workqueue_t *workqueue);

void workqueue_add_job(workqueue_t *workqueue, job_t *job);

主線程的on_accept函數爲：

void on_accept(evutil_socket_t fd, short ev, void *arg) {
    int client_fd;
    struct sockaddr_in client_addr;
    socklen_t client_len = sizeof(client_addr);
    workqueue_t *workqueue = (workqueue_t *)arg;
    client_t *client;
    job_t *job;

    client_fd = accept(fd, (struct sockaddr *)&client_addr, &client_len);
    if (client_fd < 0) {
        warn("accept failed");
        return;
    }

    /* Set the client socket to non-blocking mode. */
    if (evutil_make_socket_nonblocking(client_fd) < 0) 
    {
        warn("failed to set client socket to non-blocking");
        close(client_fd);
        return;
    }

    /* Create a client object. */
    if ((client = malloc(sizeof(*client))) == NULL) 
    {
        warn("failed to allocate memory for client state");
        close(client_fd);
        return;
    }
    memset(client, 0, sizeof(*client));
    client->fd = client_fd;

    /* Add any custom code anywhere from here to the end of this function
    * to initialize your application-specific attributes in the client struct.
    */

    if ((client->output_buffer = evbuffer_new()) == NULL) 
    {
        warn("client output buffer allocation failed");
        closeAndFreeClient(client);
        return;
    }

    if ((client->evbase = event_base_new()) == NULL) 
    {
        warn("client event_base creation failed");
        closeAndFreeClient(client);
        return;
    }

    client->buf_ev = bufferevent_socket_new(client->evbase, client_fd, BEV_OPT_CLOSE_ON_FREE);
    if ((client->buf_ev) == NULL) {
        warn("client bufferevent creation failed");
        closeAndFreeClient(client);
        return;
    }
    bufferevent_setcb(client->buf_ev, buffered_on_read, buffered_on_write,
                  buffered_on_error, client);

    /* We have to enable it before our callbacks will be
     * called. */
    bufferevent_enable(client->buf_ev, EV_READ);

    /* Create a job object and add it to the work queue. */
    if ((job = malloc(sizeof(*job))) == NULL) {
        warn("failed to allocate memory for job state");
        closeAndFreeClient(client);
        return;
    }
    job->job_function = server_job_function;
    job->user_data = client;

    workqueue_add_job(workqueue, job);
}

job中的工做指南爲：

static void server_job_function(struct job *job) {
    client_t *client = (client_t *)job->user_data;
    //do my job
    event_base_dispatch(client->evbase);
    
    closeAndFreeClient(client);
    free(job);
}

程序源碼： echoserver_threaded.c workqueue.c workqueue.h Makefile

3.4 編程實例之回顯服務器（多線程--線程池+異步IO）

設計思想：假設咱們的用戶不少，高併發，長鏈接，那麼咱們仍是來用I/O複用和線程池實現吧，用一個控制線程經過I/O複用負責監聽和分發事件，用一組線程池來進行處理事件，這樣就能夠靈活地將控制邏輯和業務邏輯分開了，見下述講解。

程序分析
具體的流程和上面的差很少，用一個主線程監聽一個套接字，並將套接字接受到的鏈接accept，並建立一個client數據結構保存該鏈接的信息，在這個client結構中註冊一個bufferevent事件，可是這裏，將事件註冊到accept_evbase中，仍然用主線程進行監聽。

而面對監聽後出現的事件，將client和操做client的方法打包成一個job，放到上述的workqueue中去，讓工做進程來完成。這樣的操做和上述的差異在於上述方法將bufferevent註冊到client中的evbase中，用工做線程監聽，而本方法用主線程監聽，工做線程負責處理監聽產生的事件。

這要的差異在於兩個函數 on_accept函數:

void on_accept(evutil_socket_t fd, short ev, void *arg) {
    int client_fd;
    struct sockaddr_in client_addr;
    socklen_t client_len = sizeof(client_addr);
    client_t *client;

    client_fd = accept(fd, (struct sockaddr *)&client_addr, &client_len);
    if (client_fd < 0) {
        warn("accept failed");
        return;
    }

    /* Set the client socket to non-blocking mode. */
    if (evutil_make_socket_nonblocking(client_fd) < 0) {
        warn("failed to set client socket to non-blocking");
        close(client_fd);
        return;
    }

    /* Create a client object. */
    if ((client = malloc(sizeof(*client))) == NULL) {
        warn("failed to allocate memory for client state");
        close(client_fd);
        return;
    }
    memset(client, 0, sizeof(*client));
    client->fd = client_fd;

    /* Add any custom code anywhere from here to the end of this function
     * to initialize your application-specific attributes in the client struct.
    */

    if ((client->output_buffer = evbuffer_new()) == NULL) {
        warn("client output buffer allocation failed");
        closeAndFreeClient(client);
        return;
    }
    //須要注意的是，這裏註冊到evbase_accept
    client->buf_ev = bufferevent_socket_new(evbase_accept, client_fd,BEV_OPT_CLOSE_ON_FREE);
    if ((client->buf_ev) == NULL) {
        warn("client bufferevent creation failed");
        closeAndFreeClient(client);
        return;
    }
    bufferevent_setcb(client->buf_ev, buffered_on_read, buffered_on_write,
                  buffered_on_error, client);

    /* We have to enable it before our callbacks will be
    * called. */
    bufferevent_enable(client->buf_ev, EV_READ);
}

在buffered_on_read中，提交job。

void buffered_on_read(struct bufferevent *bev, void *arg) 
{
    client_t *client = (client_t *)arg;
    job_t *job;

    /* Create a job object and add it to the work queue. */
    if ((job = malloc(sizeof(*job))) == NULL) {
        warn("failed to allocate memory for job state");
        closeAndFreeClient(client);
        return;
    }
    job->job_function = server_job_function;
    job->user_data = client;

    workqueue_add_job(&workqueue, job);
}

在job工做指南server_job_function中就能夠作你工做該作的事兒了，根據發來的信息進行數據庫處理，http返回等等。

程序源碼： echoserver_threaded.c workqueue.c workqueue.h Makefile