深入淺出理解libevent—

概述

libevent,libev,libuv都是c實現的異步事件庫，注冊異步事件，檢測異步事件，根據事件的觸發先后順序，調用相對應回調函數處理事件。處理的事件包括：網絡 io 事件、定時事件以及信號事件。這三個事件驅動著服務器的運行。

網絡io事件：
linux：epoll、poll、select
mac：kqueue
window：iocp
定時事件：
紅黑樹
最小堆：二叉樹、四叉樹
跳表
時間輪
信號事件

libevent 和 libev 主要封裝了異步事件庫與操作系統的交互簡單的事件管理接口，讓用戶無需關注平臺檢測處理事件的機制的差異，只需關注事件的具體處理。

從設計理念出發，libev 是為了改進 libevent 中的一些架構決策；例如：全局變量的使用使得在多線程環境中很難安全地使用 libevent；event 的數據結構設計太大，它包含了 io、時間以及信號處理全封裝在一個結構體中，額外的組件如 http、dns、openssl 等實現質量差（容易產生安全問題），計時器不精確，不能很好地處理時間事件；

libev 通過完全去除全局變量的使用，而是通過回調傳參來傳遞上下文(后面libevent也這樣做了)；并且根據不同事件類型構建不同的數據結構，以此來減低事件耦合性；計時器使用最小四叉堆。libev 小而高效；只關注事件處理。

libevent 和 libev 對 window 支持比較差，由此產生了 libuv 庫；libuv 基于 libev，在window 平臺上更好的封裝了 iocp；node.js 基于 libuv；

libevent的優點

上來當然要先夸獎啦，Libevent 有幾個顯著的亮點：
? => 事件驅動（event-driven），高性能;
? => 輕量級，專注于網絡，不如ACE那么臃腫龐大；
? => 源代碼相當精煉、易讀；
? => 跨平臺，支持Windows、Linux、*BSD和Mac Os；
? => 支持多種I/O多路復用技術， epoll、poll、dev/poll、select和kqueue等；
? => 支持I/O，定時器和信號等事件；
? => 注冊事件優先級；

Libevent已經被廣泛的應用，作為底層的網絡庫；比如memcached、Vomit、Nylon、Netchat等等。

Libevent當前的最新穩定版是1.4.13；這也是本文參照的版本。

工作流程圖：

libevent的封裝層次

?如果不想自己操作IO事件，那么我們就將IO讀寫的操作交給libevent進行管理，讓其幫我們去處理邊界問題。從較高的封裝層次去使用libevent，我們只需要在libevent完成讀寫I/O的處理后自己僅需從緩沖區中讀數據來完成事件的邏輯處理，至于邊界的問題，我們不需要操心。下面會有更詳細的介紹
?

IO事件檢測的封裝與api介紹

libevent封裝了兩個層次，一個是事件檢測，一個是事件操作。事件檢測是低層次的封裝，由libevent完成事件的檢測，然后調用者自己完成IO操作，類似于將底層的epoll，select，poll的細節隱藏掉。該層次封裝了事件管理器的操作和事件本身的操作接口。

事件管理器event_base

構建事件管理器event_base_new

使用libevent 函數之前需要分配一個或者多個 event_base 結構體, 每個event_base結構體持有一個事件集合, 可以檢測以確定哪個事件是激活的, event_base結構相當于epoll紅黑樹的樹根節點, 每個event_base都有一種用于檢測某種事件已經就緒的 “方法”
?

struct event_base *event_base_new(void); 
函數說明: 獲得event_base結構，當于epoll紅黑樹的樹根節點
參數說明: 無
返回值: 成功返回event_base結構體指針;失敗返回NULL;

釋放事件管理器event_base_free

void event_base_free(struct event_base *);
函數說明: 釋放event_base指針

event_reinit

int event_reinit(struct event_base *base);
函數說明: 如果有子進程, 且子進程也要使用base, 則子進程需要對event_base重新初始化, 
此時需要調用event_reinit函數.
函數參數: 由event_base_new返回的執行event_base結構的指針
返回值: 成功返回0, 失敗返回-1
對于不同系統而言, event_base就是調用不同的多路IO接口去判斷事件是否已經被激活, 
對于linux系統而言, 核心調用的就是epoll, 同時支持poll和select.

event_get_supported_methods

const char **event_get_supported_methods(void);
函數說明: 獲得當前系統(或者稱為平臺)支持的方法有哪些
參數: 無
返回值: 返回二維數組, 類似與main函數的第二個參數**argv.

event_base_get_method

const char * event_base_get_method(const struct event_base *base);
函數說明: 獲得當前base節點使用的多路io方法
函數參數: event_base結構的base指針.
返回值: 獲得當前base節點使用的多路io方法的指針

?event_set()

void event_set(struct event *ev, int fd, short events, void (*callback)(int, short, void *), void *arg)
event_set 初始化事件event，設置回調函數和關注的事件。
參數說明：
ev：執行要初始化的event對象；
fd：該event綁定的“句柄”，對于信號事件，它就是關注的信號；
events：在該fd上關注的事件類型，它可以是EV_READ, EV_WRITE, EV_SIGNAL；
callback：這是一個函數指針，當fd上的事件event發生時，調用該函數執行處理，它有三個參數，調用時由event_base負責傳入，按順序，實際上就是event_set時的fd, event和arg；
arg：傳遞給callback函數指針的參數；定時事件說明：evtimer_set(&ev, timer_cb, NULL) = event_set(&ev, -1, 0, timer_cb, NULL)
由于定時事件不需要fd，并且定時事件是根據添加時（event_add）的超時值設定的，因此這里event也不需要設置。
這一步相當于初始化一個event handler，在libevent中事件類型保存在event結構體中。
注意：libevent并不會管理event事件集合，這需要應用程序自行管理；

#include <event.h>
#include <stdio.h>
#include <string.h>
int main()
{const char** p = event_get_supported_methods();//獲取當前系統支持的方法有哪些int i = 0;while(p[i] != NULL){printf("[%s] ",p[i]);}printf("\n");struct event_base* base = event_base_new();if(base == NULL) printf("event_base_new error\n");printf("[%s]\n",event_base_get_method(base));event_base_free(base);return 0;
}

?struct event結構體分析

struct event {TAILQ_ENTRY (event) ev_next;TAILQ_ENTRY (event) ev_active_next;TAILQ_ENTRY (event) ev_signal_next;unsigned int min_heap_idx;	/* for managing timeouts 用于管理超時默認是-1 */struct event_base *ev_base; //屬于哪個一event_baseint ev_fd; //設置事件監聽對象，也就是監聽句柄short ev_events; //設置監聽對象觸發的動作：EV_READ, EV_WRITE, EV_SIGNAL,EV_TIMEOUT,EV_PERSIST  short ev_ncalls; //事件被調用了幾次short *ev_pncalls;	/* Allows deletes in callback */struct timeval ev_timeout;int ev_pri;		/* smaller numbers are higher priority *///設置事件的回調函數void (*ev_callback)(int, short, void *arg);//設置事件的回調函數的參數void *ev_arg;int ev_res;		/* result passed to event callback */int ev_flags; //事件的狀態,EVLIST_INIT,EVLIST_INTERNAL,EVLIST_ACTIVE,EVLIST_SIGNAL,EVLIST_INSERTED,EVLIST_TIMEOUT
};

事件循環event_base_dispatch和event_base_loop

libevent在event_base_new好之后, 需要等待事件的產生, 也就是等待事件被激活, 所以程序不能退出, 對于epoll來說, 我們需要自己控制循環, 而在libevent中也給我們提供了API接口, 類似where(1)的功能.
?

//這個函數一般不用, 而大多數都調用libevent給我們提供的另外一個API：
int event_base_loop(struct event_base *base, int flags); 
函數說明: 進入循環等待事件
參數說明:base: 由event_base_new函數返回的指向event_base結構的指針flags的取值：#define EVLOOP_ONCE	0x01只觸發一次, 如果事件沒有被觸發, 阻塞等待#define EVLOOP_NONBLOCK	0x02非阻塞方式檢測事件是否被觸發, 不管事件觸發與否, 都會立即返回.

int event_base_dispatch(struct event_base *base);
函數說明: 進入循環等待事件
參數說明:由event_base_new函數返回的指向event_base結構的指針
調用該函數, 相當于沒有設置標志位的event_base_loop。程序將會一直運行, 
直到沒有需要檢測的事件了, 或者被結束循環的API終止。

?事件循環推出event_base_loopbreak和event_base_loopexit

int event_base_loopexit(struct event_base *base, const struct timeval *tv);
int event_base_loopbreak(struct event_base *base);
struct timeval {long    tv_sec;                    long    tv_usec;            
};

兩個函數的區別是如果正在執行激活事件的回調函數, 那么event_base_loopexit將在事件回調執行結束后終止循環（如果tv時間非NULL, 那么將等待tv設置的時間后立即結束循環）, 而event_base_loopbreak會立即終止循環。

event_process_active

主要是處理激活隊列中的數據

static void
event_process_active(struct event_base *base)
{struct event *ev;struct event_list *activeq = NULL;int i;short ncalls;獲得就緒鏈表中有就緒事件并且高優先級的表頭for (i = 0; i < base->nactivequeues; ++i) {if (TAILQ_FIRST(base->activequeues[i]) != NULL) {activeq = base->activequeues[i];break;}}assert(activeq != NULL);for (ev = TAILQ_FIRST(activeq); ev; ev = TAILQ_FIRST(activeq)) {if (ev->ev_events & EV_PERSIST)event_queue_remove(base, ev, EVLIST_ACTIVE);elseevent_del(ev);//如果不是永久事件則需要進行一系統的刪除工作，包括移除注冊在事件鏈表的事件等/* Allows deletes to work */ncalls = ev->ev_ncalls;ev->ev_pncalls = &ncalls;while (ncalls) {ncalls--;ev->ev_ncalls = ncalls;//根據回調次數調用回調函數(*ev->ev_callback)((int)ev->ev_fd, ev->ev_res, ev->ev_arg);if (event_gotsig || base->event_break)return;}}
}

事件對象

typedef void (*event_callback_fn)(evutil_socket_t fd, short events, void *arg);
struct event *event_new(struct event_base *base, evutil_socket_t fd, 
short events, event_callback_fn cb, void *arg);#define evsignal_new(b, x, cb, arg)   event_new((b), (x), EV_SIGNAL|EV_PERSIST, (cb), (arg))
函數說明: event_new負責創建event結構指針, 同時指定對應的base(epfd), 還有對應的文件描述符, 事件, 以及回調函數和回調函數的參數。
參數說明：base: 對應的根節點--epfdfd: 要監聽的文件描述符events:要監聽的事件#define  EV_TIMEOUT    0x01   //超時事件#define  EV_READ       0x02    //讀事件#define  EV_WRITE      0x04    //寫事件#define  EV_SIGNAL     0x08    //信號事件#define  EV_PERSIST     0x10    //周期性觸發#define  EV_ET         0x20    //邊緣觸發, 如果底層模型支持設置										 則有效, 若不支持則無效.若要想設置持續的讀事件則： EV_READ | EV_PERSISTcb: 回調函數, 原型如下：typedef void (*event_callback_fn)(evutil_socket_t fd, short events, void *arg);注意: 回調函數的參數就對應于event_new函數的fd, event和arg

銷毀事件對象event_free

void event_free(struct event *ev);
函數說明: 釋放由event_new申請的event節點。

注冊事件event_add(類似于epoll_ctl)

int event_add(struct event *ev, const struct timeval *timeout);
函數說明: 將非未決態事件轉為未決態, 相當于調用epoll_ctl函數(EPOLL_CTL_ADD), 
開始監聽事件是否產生, 相當于epoll的上樹操作.
參數說明：ev: 調用event_new創建的事件timeout: 限時等待事件的產生(定時事件使用), 也可以設置為NULL, 沒有限時。

注銷事件event_del(類似于epoll的del)

int event_del(struct event *ev);
函數說明: 將事件從未決態變為非未決態, 相當于epoll的下樹（epoll_ctl調用EPOLL_CTL_DEL操作）操作。
參數說明: ev指的是由event_new創建的事件.

事件驅動event介紹

事件驅動是libevent的核心思想

比較重要，但是學過epoll和reactor的話，學起來還是比較簡單

主要幾個狀態：

無效的指針: 此時僅僅是定義了 struct event *ptr；
非未決：相當于創建了事件, 但是事件還沒有處于被監聽狀態, 類似于我們使用epoll的時候定義了struct epoll_event ev并且對ev的兩個字段進行了賦值, 但是此時尚未調用epoll_ctl對事件上樹.
未決：就是對事件開始監聽, 暫時未有事件產生。相當于調用epoll_ctl對要監聽的事件上樹, 但是沒有事件產生.
激活：代表監聽的事件已經產生, 這時需要處理, 相當于調用epoll_wait函數有返回, 當事件被激活以后, libevent會調用該事件對應的回調函數.

只用libevent事件檢測，io操作自己來處理demo

像memcached它就是用這種層次(只使用libevent檢測，io操作自己寫)。我們從下面Demo中看到，使用libevent就像操作reactor一樣，只需要傳遞回調函數，在回調函數里面去寫io操作的邏輯。

#include <event.h>
#include <event2/listener.h>
#include <stdio.h>
#include <unistd.h>
#include <string.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <arpa/inet.h>
//為什么events是short是因為庫里定義過了
// #define  EV_TIMEOUT    0x01   //超時事件
// #define  EV_READ       0x02    //讀事件
// #define  EV_WRITE      0x04    //寫事件
// #define  EV_SIGNAL     0x08    //信號事件
// #define  EV_PERSIST     0x10    //周期性觸發
// #define  EV_ET         0x20    //邊緣觸發, 如果底層模型支持設置	// int event_assign(struct event *ev, struct event_base *base, evutil_socket_t fd, short events, event_callback_fn callback, void *arg);
// ev：要進行初始化的事件結構體指針。
// base：事件所關聯的事件基礎。
// fd：文件描述符或套接字，表示這個事件與哪個描述符相關聯。
// events：指定事件類型，比如讀取、寫入等。可以使用 EV_READ、EV_WRITE 等常量。
// callback：是事件觸發時調用的函數。
// arg：是傳遞給回調函數 callback 的可選參數。
void socket_read_cb(int fd,short events,void* arg);
void socket_accept_cb(int fd,short events,void* arg)
{sleep(1);struct sockaddr_in addr;socklen_t len = sizeof(addr);int clientfd = accept(fd, (struct sockaddr *) &addr, &len);evutil_make_socket_nonblocking(clientfd);//將該fd設置成非阻塞的printf("client_fd:%d",clientfd);struct event_base* base = (struct event_base*)arg;//這個是reactor對象struct event *ev = event_new(NULL, -1, 0, NULL, NULL);//建立一個event//注冊這個事件,寫到base里event_assign(ev,base,clientfd,EV_READ | EV_PERSIST,socket_read_cb,(void*)ev);//只是寫入了數據//這次的arg我們傳的是事件結構體event_add(ev,NULL);//跟epoll_ctl一樣,但是也不是很相同}void socket_read_cb(int fd,short events,void* arg)
{char msg[1024];struct event* ev = (struct event*)arg;int len = read(fd,msg,sizeof(msg) - 1);if (len <= 0) {printf("client fd:%d disconnect\n", fd);event_free(ev);close(fd);return;}msg[len] = '\0';printf("recv the client msg: %s",msg);char reply_msg[1024] = "recvieced msg: ";strcat(reply_msg + strlen(reply_msg), msg);write(fd, reply_msg, strlen(reply_msg));
}
int main()
{int listenfd = socket(AF_INET,SOCK_STREAM,0);if(listenfd  == -1) printf("socket error\n");struct sockaddr_in addr;memset(&addr, 0, sizeof(addr)); addr.sin_family = AF_INET;addr.sin_port = htons(8080);addr.sin_addr.s_addr = htonl(INADDR_ANY);if(bind(listenfd ,(struct sockaddr*)&addr,sizeof(addr)) == -1)printf("bind error\n");listen(listenfd ,3);struct event_base* base  = event_base_new();printf("create base\n");struct event* ev_listen = event_new(base,listenfd,EV_READ | EV_PERSIST,socket_accept_cb,base);//返回是struct event*得到這個事件結構體的指針/*event_new 等價于struct event ev_listen;event_set(&ev_listen, listenfd, EV_READ | EV_PERSIST, socket_accept_cb, base);event_base_set(base, &ev_listen);*/event_add(ev_listen,NULL);//相當于epoll_ctlevent_base_dispatch(base);//循環等待事件return 0;
}

IO事件操作的封裝與api介紹(主要是evconnlistener和bufferevent)

?自帶buffer的事件-bufferevent

bufferevent實際上也是一個event, 只不過比普通的event高級一些, 它的內部有兩個緩沖區, 以及一個文件描述符（網絡套接字）。一個網絡套接字有讀和寫兩個緩沖區, bufferevent同樣也帶有兩個緩沖區, 還有就是libevent事件驅動的核心回調函數, 那么四個緩沖區以及觸發回調的關系如下：

從圖中可以得知, 一個bufferevent對應兩個緩沖區, 三個回調函數, 分別是寫回調, 讀回調和事件回調

bufferevent有三個回調函數：

讀回調 – 當bufferevent將底層讀緩沖區的數據讀到自身的讀緩沖區時觸發讀事件回調.
寫回調 – 當bufferevent將自身寫緩沖的數據寫到底層寫緩沖區的時候觸發寫事件回調, 由于數據最終是寫入了內核的寫緩沖區中, 應用程序已經無法控制, 這個事件對于應用程序來說基本沒什么用, 只是通知功能.
事件回調 – 當bufferevent綁定的socket連接, 斷開或者異常的時候觸發事件回調.
?

構建bufferevent對象?

struct bufferevent *bufferevent_socket_new(struct event_base *base, evutil_socket_t fd, int options);
函數說明: bufferevent_socket_new 對已經存在socket創建bufferevent事件, 可用于
后面講到的連接監聽器的回調函數中.
參數說明：base :對應根節點fd   :文件描述符options : bufferevent的選項BEV_OPT_CLOSE_ON_FREE  -- 釋放bufferevent自動關閉底層接口(當bufferevent被釋放以后, 文件描述符也隨之被close)    BEV_OPT_THREADSAFE  -- 使bufferevent能夠在多線程下是安全的

銷毀bufferevent對象

void bufferevent_free(struct bufferevent *bufev);
函數說明: 釋放bufferevent

連接操作bufferevent_socket_connect

int bufferevent_socket_connect(struct bufferevent *bev, struct sockaddr *serv, int socklen);
函數說明: 該函數封裝了底層的socket與connect接口, 通過調用此函數, 可以將bufferevent事件與通信的socket進行綁定, 參數如下：bev – 需要提前初始化的bufferevent事件serv – 對端(一般指服務端)的ip地址, 端口, 協議的結構指針socklen – 描述serv的長度
說明: 調用此函數以后, 通信的socket與bufferevent緩沖區做了綁定, 后面調用了bufferevent_setcb函數以后, 會對bufferevent緩沖區的讀寫操作的事件設置回調函數, 當往緩沖區中寫數據的時候會觸發寫回調函數, 當數據從socket的內核緩沖區讀到bufferevent讀緩沖區中的時候會觸發讀回調函數.

設置bufferevent回調與bufferevent_setcb

void bufferevent_setcb(struct bufferevent *bufev,bufferevent_data_cb readcb,bufferevent_data_cb writecb,bufferevent_event_cb eventcb,void *cbarg
);
函數說明: bufferevent_setcb用于設置bufferevent的回調函數, 
readcb, writecb,eventcb分別對應了讀回調, 寫回調, 事件回調, 
cbarg代表回調函數的參數。

回調函數的原型：

typedef void (*bufferevent_data_cb)(struct bufferevent *bev, void *ctx);
typedef void (*bufferevent_event_cb)(struct bufferevent *bev, short what, void *ctx);
What 代表 對應的事件
BEV_EVENT_EOF--遇到文件結束指示
BEV_EVENT_ERROR--發生錯誤
BEV_EVENT_TIMEOUT--發生超時
BEV_EVENT_CONNECTED--請求的過程中連接已經完成

寫數據到寫緩沖區bufferevent_write

int bufferevent_write(struct bufferevent *bufev, const void *data, size_t size);
int bufferevent_write_buffer(struct bufferevent *bufev, struct evbuffer *buf);
bufferevent_write是將data的數據寫到bufferevent的寫緩沖區，bufferevent_write_buffer 
是將數據寫到寫緩沖區另外一個寫法, 實際上bufferevent的內部的兩個緩沖區結構就是struct evbuffer。

從讀緩沖區讀數據bufferevent_read

size_t bufferevent_read(struct bufferevent *bufev, void *data, size_t size);
int bufferevent_read_buffer(struct bufferevent *bufev, struct evbuffer *buf);
bufferevent_read 是將bufferevent的讀緩沖區數據讀到data中, 同時將讀到的數據從
bufferevent的讀緩沖清除。
bufferevent_read_buffer 將bufferevent讀緩沖數據讀到buf中, 接口的另外一種。

注冊與注銷事件類型bufferevent_enable/disable

int bufferevent_enable(struct bufferevent *bufev, short event);
int bufferevent_disable(struct bufferevent *bufev, short event);
bufferevent_enable與bufferevent_disable是設置事件是否生效, 如果設置為disable, 
事件回調將不會被觸發。

獲取讀寫緩沖區bufferevent_get_input和bufferevent_get_oupput

struct evbuffer *bufferevent_get_input(struct bufferevent *bufev)
struct evbuffer *bufferevent_get_output(struct bufferevent *bufev)
獲取bufferevent的讀緩沖區和寫緩沖區

分割字符讀evbuffer_readln與固定長度讀evbuffer_remove

char *evbuffer_readln(struct evbuffer *buffer, size_t *n_read_out, enum evbuffer_eol_style eol_style);
int evbuffer_remove(struct evbuffer *buf, void *data, size_t datlen);
分割字符讀evbuffer_readln
固定長度讀evbuffer_remove

bufferevent總結

對于bufferevent來說, 一個文件描述符, 2個緩沖區, 3個回調函數。文件描述符是用于和客戶端進行通信的通信文件描述符, 并不是監聽的文件描述符。

2個緩沖區是指: 一個bufferevent包括讀緩沖區和寫緩沖區。

3個回調函數指: 讀回調函數寫回調函數和事件回調函數

讀回調函數的觸發時機:

當socket的內核socket讀緩沖區中有數據的時候, bufferevent會將內核緩沖區中的數據讀到自身的讀緩沖區, 會觸發bufferevent的讀操作, 此時會調用bufferevent的讀回調函數.

寫回調函數的觸發時機:

當往bufferevent的寫緩沖區寫入數據的時候, bufferevent底層會把緩沖區中的數據寫入到內核的socket的寫緩沖區中, 此時會觸發bufferevent的寫回調函數, 最后由內核的驅動程序將數據發送出去.

事件(異常)回調函數的觸發時機:

客戶端關閉連接或者是被信號終止進程會觸發事件回調函數

連接監聽器-evconnlistener

鏈接監聽器封裝了底層的socket通信相關函數, 比如socket, bind, listen, accept這幾個函數。鏈接監聽器創建后實際上相當于調用了socket, bind, listen, 此時等待新的客戶端連接到來, 如果有新的客戶端連接, 那么內部先進行調用accept處理, 然后調用用戶指定的回調函數。可以先看看函數原型, 了解一下它是怎么運作的：
?

構建連接監聽器evconnlistener_new_bind

struct evconnlistener *evconnlistener_new_bind(struct event_base *base,evconnlistener_cb cb, void *ptr, unsigned flags, int backlog,const struct sockaddr *sa, int socklen
);
函數說明: 
是在當前沒有套接字的情況下對鏈接監聽器進行初始化, 看最后2個參數實際上就是bind使用的關鍵參數, 
backlog是listen函數的關鍵參數（略有不同的是, 如果backlog是-1, 那么監聽器會自動選擇一個合適的值, 
如果填0, 那么監聽器會認為listen函數已經被調用過了）, ptr是回調函數的參數, cb是有新連接之后的回調函數, 
但是注意這個回調函數觸發的時候, 鏈接器已經處理好新連接了, 并將與新連接通信的描述符交給回調函數。flags 需要參考幾個值：LEV_OPT_LEAVE_SOCKETS_BLOCKING   文件描述符為阻塞的LEV_OPT_CLOSE_ON_FREE            關閉時自動釋放LEV_OPT_REUSEABLE                端口復用LEV_OPT_THREADSAFE               分配鎖, 線程安全

struct evconnlistener *evconnlistener_new(struct event_base *base,evconnlistener_cb cb, void *ptr, unsigned flags, int backlog,evutil_socket_t fd
);

evconnlistener_new函數與前一個函數不同的地方在與后2個參數, 使用本函數時, 認為socket已經初始化好, 并且bind完成, 甚至也可以做完listen, 所以大多數時候, 我們都可以使用第一個函數。

accept的回調函數evconnlistener_cb

typedef void (*evconnlistener_cb)(struct evconnlistener *evl, evutil_socket_t fd, struct sockaddr *cliaddr, int socklen, void *ptr);

回調函數fd參數是與客戶端通信的描述符, 并非是等待連接的監聽的那個描述符, 所以cliaddr對應的也是新連接的對端地址信息, 已經是accept處理好的。

銷毀連接監聽器evconnlistener_free

void evconnlistener_free(struct evconnlistener *lev);

使用libevent的事件檢測與事件操作demo?

#include <netinet/in.h>
#include <sys/socket.h>
#include <stdio.h>
#include <signal.h>
#include <event.h>
#include <event2/buffer.h>
#include <time.h>
#include <string.h>
#include <stdlib.h>
#include <event2/bufferevent.h>
#include <event2/listener.h>
void socket_read_callback(struct bufferevent* bev,void* arg)
{//操作讀緩沖當中過的數據，通過該函數得到讀緩沖的地址struct evbuffer* evbuf = bufferevent_get_input(bev);char* msg = evbuffer_readln(evbuf,NULL,EVBUFFER_EOL_LF);//我們的bufferevent里肯定有讀緩沖// 也可以直接用 bufferevent_read 讀數據// bufferevent_read(struct bufferevent *bufev, void *data, size_t size)if(!msg) return;printf("server read the data: %s\n",msg);char reply[1024] = {0};sprintf(reply, "recvieced msg: %s\n", msg);//echo//需要自己釋放資源，很重要free(msg);bufferevent_write(bev,reply,strlen(reply));
}
void socket_event_callback(struct bufferevent *bev, short events, void *arg)
{if(events & BEV_EVENT_EOF) printf("connection closed\n");else if (events & BEV_EVENT_ERROR)printf("some other error\n");else if (events & BEV_EVENT_TIMEOUT)printf("timeout\n");bufferevent_free(bev);}
void listener_callback(struct evconnlistener *listener,evutil_socket_t fd,struct sockaddr *sock,int socklen, void *arg)
{char ip[32] = {0};evutil_inet_ntop(AF_INET,sock,ip,sizeof(ip) - 1);//該函數的作用是將網絡字節序表示的 IPv4 地址轉換為可讀的字符串格式，并將結果存儲在提供的緩沖區中。//這樣可以方便地將 IP 地址以人可讀的形式輸出，比如在日志中記錄連接的來源。printf("accept a client fd:%d ip:%s\n",fd,ip);//也就是說，監聽到之后，觸發回調，然后會自動把新連接的fd傳進來嗎//也就是相當于我們前面設置了一個監聽套接字的bufferevent，當內核中的socket有鏈接到來的時候//也就是內核中的讀緩沖有數據的時候，那么就會觸發bufferevent回調，寫到bufferevent的讀緩沖區//然后把數據傳到我們用戶層，就不需要我們自己去事件處理了struct event_base* base = (struct event_base*)arg;//把reactor對象傳進來了//創建一個bufferevent，構建bufferevent對象struct bufferevent* bev = bufferevent_socket_new(base,fd,BEV_OPT_CLOSE_ON_FREE);//函數說明: bufferevent_socket_new 對已經存在socket創建bufferevent事件, 可用于//后面講到的連接監聽器的回調函數中.我這里的fd是accept是用于和客戶端交互的fd//所以這個fd是我們設置到bufferevent中，讓bufferevent對象幫我們處理//選了這個選項會當bufferevent釋放后，里面的fd什么的也會自動關閉和釋放// 設置讀、寫、以及異常時的回調函數bufferevent_setcb(bev,socket_read_callback,NULL,socket_event_callback,NULL);bufferevent_enable(bev,EV_READ | EV_PERSIST);//注冊事件 
}
void stdin_callback(struct bufferevent* bev,void* arg)
{struct evbuffer* evbuf = bufferevent_get_input(bev);struct event_base* base =  (struct event_base*)arg;//這個就是上下文char* msg = evbuffer_readln(evbuf,NULL,EVBUFFER_EOL_LF);if(!msg) return;if (strcmp(msg, "quit") == 0) {printf("safe exit!!!\n");event_base_loopbreak(arg);//中斷事件循環}printf("stdio read the data: %s\n", msg);
}
void do_timer(int fd,short events,void* arg)
{struct event *timer = (struct event *) arg;time_t now = time(NULL);printf("do_timer %s", (char *) ctime(&now));
}
void do_sig_int(int fd,short events,void* arg)
{struct event *si = (struct event *)arg;event_del(si);printf("do_sig_int SIGINT\n");//CTRL + C
}
int main()
{struct sockaddr_in sin;memset(&sin,0,sizeof(sin));sin.sin_family = AF_INET;sin.sin_port = htons(8088);sin.sin_addr.s_addr = htonl(INADDR_ANY);struct event_base* base = event_base_new();//鏈接監聽器封裝了底層的socket通信相關函數, 比如socket, bind, listen, accept這幾個函數。struct evconnlistener* listener = evconnlistener_new_bind(base,listener_callback,base,LEV_OPT_REUSEABLE | LEV_OPT_CLOSE_ON_FREE,10,(struct sockaddr*)&sin,sizeof(sin));//: 創建一個監聽器對象，該監聽器將監聽指定的地址和端口，//并在有新連接時調用 listener_callback 函數進行處理。參數說明如下：// base: 事件基礎結構，將監聽器與此關聯。// listener_callback: 當有新連接建立時將調用的回調函數。// base: 將傳遞給回調函數的參數，這里是事件基礎結構。// LEV_OPT_REUSEABLE | LEV_OPT_CLOSE_ON_FREE: 一組標志，指定監聽器的行為，包括啟用地址重用和在釋放時關閉底層文件描述符。// 10: 允許在套接字上排隊等待連接的最大數量。// (struct sockaddr*)&sin: 要監聽的地址和端口信息。// sizeof(sin): 提供地址結構體的大小。//對stdin的io事件進行處理//stdin的文件描述符是0struct bufferevent* ioev = bufferevent_socket_new(base,0,BEV_OPT_CLOSE_ON_FREE);bufferevent_setcb(ioev,stdin_callback,NULL,NULL,base);//設置讀寫事件回調bufferevent_enable(ioev,EV_READ | EV_PERSIST);//設置事件屬性并開啟//定時事件struct event evtimer;struct timeval tv = {1,0};event_set(&evtimer,-1,EV_PERSIST,do_timer,&evtimer);//初始化// &evtimer: 是一個指向 struct event 結構體的指針，表示你要設置的事件對象。// -1: 是事件的文件描述符，定時器事件不需要一個真實的文件描述符，因此通常設置為 -1。// EV_PERSIST: 是事件的標志，表示這是一個持久性事件，即它會在每次觸發后自動重新添加到事件循環中，使得它可以周期性地觸發。// do_timer: 是事件觸發時執行的回調函數。// &evtimer: 是傳遞給回調函數的用戶數據。event_base_set(base,&evtimer);//設置event從屬的event_base,這一步相當于指明event要注冊到哪個event_base實例上。event_add(&evtimer,&tv);//信號事件struct event ev_sig_int;event_set(&ev_sig_int,-1,EV_PERSIST,do_sig_int,&ev_sig_int);event_base_set(base,&ev_sig_int);event_add(&ev_sig_int,NULL);//開啟事件主循環event_base_dispatch(base);/* 結束釋放資源 */evconnlistener_free(listener);event_base_free(base);return 0;
}

libevent事件原理剖析?

信號事件剖析(這個我們先不看了)

定時事件和網絡事件剖析

Timer小根堆

libevent定時器的機制是最小堆+epoll_wait的機制,event_base_dispatch內部調用的是event_base_loop，我們進入主循環看看，發現它先是去最小堆找timeout參數，然后執行epoll_wait。之后再將所有的超時任務取出timeout_process放到就緒隊列，我們發現現在網絡事件和定時事件都被加入到就緒隊列中了，然后按照優先級進行處理，調用對應的回調函數。

while (!done) {......tv_p = &tv;if (!N_ACTIVE_CALLBACKS(base) && !(flags & EVLOOP_NONBLOCK)) {timeout_next(base, &tv_p);	// 返回的 tv_p 即是 最小堆實現的定時器中第一個事件的剩余等待時間} ......clear_time_cache(base);res = evsel->dispatch(base, tv_p);	// 以tv_p作為 epoll_wait 的超時時間。這里相當于epoll_wait()，收集網絡事件......update_time_cache(base);	// 更新 time_cache，time_cache的作用在于不必每次都從系統調用獲取時間值......timeout_process(base);	// 將所有已超時的任務從最小堆中取出，插入到就緒隊列（有優先級）收集定時事件if (N_ACTIVE_CALLBACKS(base)) {int n = event_process_active(base);	// 處理這些就緒的任務，調用其回調函數......} 
}/* Activate every event whose timeout has elapsed. */
static void timeout_process(struct event_base *base)
{/* Caller must hold lock. */struct timeval now;struct event *ev;if (min_heap_empty_(&base->timeheap)) {return;}gettime(base, &now);while ((ev = min_heap_top_(&base->timeheap))) {if (evutil_timercmp(&ev->ev_timeout, &now, >))	// 從堆中取出所有 ev_timeout 已達到 now 的事件break;/* delete this event from the I/O queues */event_del_nolock_(ev, EVENT_DEL_NOBLOCK);	// 從所在的 event_base 中刪除該事件event_active_nolock_(ev, EV_TIMEOUT, 1);	//  激活該事件，即 插入到就緒隊列}
}

?event_active_nolock_()底層將調用event_queue_insert_active()將事件插入到event_base下的就緒隊列activequeues中，這個就緒隊列實際上是有nactivequeues個元素的隊列數組，數組下標越小的隊列優先級越高，每次我們新建一個event時默認的優先級ev_pri都是nactivequeues / 2（by default, we put new events into the middle priority），而注冊事件到event_base前可以通過該函數來手動設置優先級：
?

/* Set's the priority of an event - if an event is already scheduled
?* changing the priority is going to fail. */
int event_priority_set(struct event *ev, int pri)
?

讀寫緩沖區evbuffer的實現（重點理解）

我們在讀寫網絡IO的時候，我們是不能確保一次讀取，就是一個完整的數據包。比如我們寫入size，但是實際寫入n<size，數據沒有全部寫出去，那剩下的數據怎么辦呢？我們需要緩存起來等待下次寫數據觸發，讀數據同理。所以因為這個原因，我們需要設置緩沖區來解決這個問題。常用的解決方案有三種

fix buffer ：char rbuf[16 * 1024 * 1024];char wbuf[16 * 1024 * 1024] ，但是這樣會造成兩個新的問題，1. 存在空間浪費 2. 數據移動頻繁
ringbuffer：環形緩沖區，解決了數據移動頻繁的問題，但是數據空間浪費的問題沒有解決
libevent中的evbuffer。下面開始介紹evbuffer。
evbuffer 是 libevent 底層實現的一種鏈式緩沖區，當我們使用bufferevent來管理事件時，就會從每個事件的 evbuffer 中讀寫數據。每個 evbuffer 實質是一個緩沖區鏈表，其中的每個元素為 struct evbuffer_chain。一個struct evbuffer中的關鍵成員定義如下：
?

struct evbuffer {/** The first chain in this buffer's linked list of chains. */struct evbuffer_chain *first;/** The last chain in this buffer's linked list of chains. */struct evbuffer_chain *last;/** Pointer to the next pointer pointing at the 'last_with_data' chain. */struct evbuffer_chain **last_with_datap;	// 指針指向最后一個可寫的 chain/** Total amount of bytes stored in all chains.*/size_t total_len;...... // 以上為關鍵成員
}

每個evbuffer_chain的定義又如下所示：

/** A single item in an evbuffer. */
struct evbuffer_chain {/** points to next buffer in the chain */struct evbuffer_chain *next;	// 指向下一個 evbuffer_chain /** total allocation available in the buffer field. */size_t buffer_len;				// buffer 的長度/** unused space at the beginning of buffer or an offset into a file for sendfile buffers. */ev_misalign_t misalign;			// 實際數據在 buffer 中的偏移/** Offset into buffer + misalign at which to start writing.* In other words, the total number of bytes actually stored in buffer. */size_t off;						// buffer 中有效數據的末尾，接下來的數據從這個位置開始填入（該位置即 buffer + misalign + off）/** number of references to this chain */int refcnt;						// 這個 buffer的引用計數/** Usually points to the read-write memory belonging to this buffer allocated as part of the evbuffer_chain allocation.* For mmap, this can be a read-only buffer and EVBUFFER_IMMUTABLE will be set in flags.  For sendfile, it may point to NULL. */unsigned char *buffer;			// 指向實際數據存儲的位置，這是真正的 buffer
};

misaligin是什么意思呢？是已經被讀取的數據，下一段有效數據是從【buffer+misaligin ， buffer+misaligin +off】這一段off的長，是我們待取的有效數據。而【buffer，buffer+misaligin 】這一段是之前就已經被讀取過了，所以這里是失效的數據。所以misaligin 就解決了數據移動頻繁的問題。而我們的evbuffer_chain是鏈表形式，所以又解決了數據空間浪費的問題。所以說evbuffer的設計是非常巧妙的。

bufferevent_write
當我們調用bufferevent_write往寫緩沖區寫數據時，實際上是調用了evbuffer_add,在寫入后libevent自動幫我們寫到內核緩沖區，之后會觸發寫回調函數。

若這個evbuffer中沒有一個 chain 可以寫入數據，則需要根據寫入的數據大小新申請一個 chain 掛到鏈表末尾，然后往這個chain中寫數據，所以每個 chain 的 buffer 大小是不定的。還有更多細節內容我寫到注釋里面了，讀者自行閱讀。

int evbuffer_add(struct evbuffer *buf, const void *data_in, size_t datlen) {//...//如果大于限定的容量if (datlen > EV_SIZE_MAX - buf->total_len) {goto done;}//使chain指向之后一個鏈表if (*buf->last_with_datap == NULL) {chain = buf->last;}else {chain = *buf->last_with_datap;}//...//如果沒有chain，那么創建一個datlen大小的返回即可if (chain == NULL) {chain = evbuffer_chain_insert_new(buf, datlen);if (!chain)goto done;}if ((chain->flags & EVBUFFER_IMMUTABLE) == 0) {//...//remain為當前可用剩余空間還有多少remain = chain->buffer_len - (size_t) chain->misalign - chain->off;//如果剩余空間大于需求，那么直接分配即可if (remain >= datlen) {/* there's enough space to hold all the data in the* current last chain */memcpy(chain->buffer + chain->misalign + chain->off,data, datlen);chain->off += datlen;buf->total_len += datlen;buf->n_add_for_cb += datlen;goto out;}//否則看一看剩余空間+misalign是否大于需求，大于則移動off數據else if (!CHAIN_PINNED(chain) &&//里面涉及別的一些細節，這里不展開evbuffer_chain_should_realign(chain, datlen)) {/* we can fit the data into the misalignment */evbuffer_chain_align(chain);memcpy(chain->buffer + chain->off, data, datlen);chain->off += datlen;buf->total_len += datlen;buf->n_add_for_cb += datlen;goto out;}}else {/* we cannot write any data to the last chain */remain = 0;}//走到這里代表一個chain不能滿足datlen，那么預分配一個tmp chain/* we need to add another chain */to_alloc = chain->buffer_len;if (to_alloc <= EVBUFFER_CHAIN_MAX_AUTO_SIZE / 2)to_alloc <<= 1;if (datlen > to_alloc)to_alloc = datlen;tmp = evbuffer_chain_new_membuf(to_alloc);if (tmp == NULL)goto done;//把當前chain給分配完if (remain) {memcpy(chain->buffer + chain->misalign + chain->off,data, remain);chain->off += remain;buf->total_len += remain;buf->n_add_for_cb += remain;}//還需要多少大小從新的tmp里面分配data += remain;datlen -= remain;memcpy(tmp->buffer, data, datlen);tmp->off = datlen;evbuffer_chain_insert(buf, tmp);buf->n_add_for_cb += datlen;out:evbuffer_invoke_callbacks_(buf);result = 0;done:EVBUFFER_UNLOCK(buf);return result;
}

bufferevent_read

bufferevent_read()底層調用evbuffer_remove，這代表它按照指定長度去讀，其又調用了evbuffer_copyout_from，具體細節就不展開了，我們知道了怎么寫，那么怎么讀我們也就知道了。

/* Reads data from an event buffer and drains the bytes read */
int evbuffer_remove(struct evbuffer *buf, void *data_out, size_t datlen)
{ev_ssize_t n;EVBUFFER_LOCK(buf);n = evbuffer_copyout_from(buf, NULL, data_out, datlen);	// 拷貝數據if (n > 0) {if (evbuffer_drain(buf, n)<0)	// drain 就是丟棄已讀走的數據，即 調整當前 chain 的 misalign 或 直接釋放數據已全部讀走的 chainn = -1;}EVBUFFER_UNLOCK(buf);return (int)n;
}

evbuffer的缺點

上面我們說了evbuffer的優點，那么evbuffer的缺點呢？其實也很明顯，即我們的數據是存儲在不連續的內存上面（例如我們讀20B，結果著20B分別在兩個chain里面），內存不連續會帶來多次io，我們可能需要多次io才能把數據讀完整。對于內存不連續的問題，Linux內核提供了一個接口，readv和writev，解決內存不連續的讀寫問題

readv：將讀緩沖區的數據讀到不連續的內存中
writev：將不連續的內存數據寫到寫緩沖區
?

man 2 readv
# 第二個參數是數組，第三個參數是數組的長度
ssize_t readv(int fd, const struct iovec *iov, int iovcnt);
ssize_t writev(int fd, const struct iovec *iov, int iovcnt);struct iovec {void  *iov_base;    /* Starting address 起始地址*/size_t iov_len;     /* Number of bytes to transfer 長度*/
};