Libevent库源码介绍及TCP后端服务器_libevent开发服务器

Libevent库介绍

Libevent 是一个轻量级的开源高性能事件通知库，主要有以下几个亮点：事件驱动，高性能;轻量级，专注于网络，源代码相当精炼、易读；跨平台，支持 Windows、 Linux、 *BSD 和 Mac Os；支持多种 I/O 多路复用技术， epoll、 poll、 dev/poll、 select 和 kqueue 等；支持 I/O，定时器和信号等事件；注册事件优先级.

Libevent模型

常用API

1.event_init()   --  event_base_new() 的封装

event_init()函数主要调用event_base_new()函数，返回event_base结构体；我们直接调用event_base_new()函数也是可以的；

struct event_base *
event_init(void)
{
struct event_base *base = event_base_new();
 
if (base != NULL)
current_base = base;
 
return (base);
}

2.event_base_new()  初始化event_base

event_base_new()主要就是对结构体event_base进行初始化，设置一些参数；

struct event_base *
event_base_new(void)
{
int i;
struct event_base *base;
 
//用calloc而不用malloc的原因?
//calloc动态分配完内存后，自动初始化该内存为零
if ((base = calloc(1, sizeof(struct event_base))) == NULL)
event_err(1, "%s: calloc", __func__);
 
event_sigcb = NULL;
event_gotsig = 0;
 
detect_monotonic();//设置use_monotonic
gettime(base, &base->event_tv);
//初始化定时事件的小根堆
min_heap_ctor(&base->timeheap);
//初始化事件链表，头 ==   尾
TAILQ_INIT(&base->eventqueue);
//初始化信号
base->sig.ev_signal_pair[0] = -1;
base->sig.ev_signal_pair[1] = -1;
//初始化IO多路复用机制
base->evbase = NULL;
//选取以NULL 结尾，初始化
for (i = 0; eventops[i] && !base->evbase; i++) {
base->evsel = eventops[i];
 
base->evbase = base->evsel->init(base);
}
//如果没有IO多路复用
 
if (base->evbase == NULL)
event_errx(1, "%s: no event mechanism available", __func__);
 
if (evutil_getenv("EVENT_SHOW_METHOD")) 
event_msgx("libevent using: %s\n",
   base->evsel->name);
 
/* allocate a single active event queue */
//设置优先级
//活跃事件链表中，优先级值越小，越优先
event_base_priority_init(base, 1);
 
return (base);
}

3.event_add()    将event添加到事件链表上，注册事件

根据时间类型添加到不同的列表中；

1.将event注册到event_base的I/O多路复用要监听的事件链表中
2.将event注册到event_base的已注册事件链表中
3.如果传入了超时时间，则删除旧的超时时间，重新设置，并将event添加到event_base的小根堆中；

int
event_add(struct event *ev, const struct timeval *tv)
{
struct event_base *base = ev->ev_base;
const struct eventop *evsel = base->evsel;
void *evbase = base->evbase;
int res = 0;
 
event_debug((
"event_add: event: %p, %s%s%scall %p",
ev,
ev->ev_events & EV_READ ? "EV_READ " : " ",
ev->ev_events & EV_WRITE ? "EV_WRITE " : " ",
tv ? "EV_TIMEOUT " : " ",
ev->ev_callback));
 
assert(!(ev->ev_flags & ~EVLIST_ALL));
 
/*
* prepare for timeout insertion further below, if we get a
* failure on any step, we should not change any state.
*/
if (tv != NULL && !(ev->ev_flags & EVLIST_TIMEOUT)) {
if (min_heap_reserve(&base->timeheap,
1 + min_heap_size(&base->timeheap)) == -1)
return (-1);  /* ENOMEM == errno */
}
 
if ((ev->ev_events & (EV_READ|EV_WRITE|EV_SIGNAL)) &&
    !(ev->ev_flags & (EVLIST_INSERTED|EVLIST_ACTIVE))) {
res = evsel->add(evbase, ev);
if (res != -1)
event_queue_insert(base, ev, EVLIST_INSERTED);
}
 
/* 
* we should change the timout state only if the previous event
* addition succeeded.
*/
if (res != -1 && tv != NULL) {
struct timeval now;
 
/* 
* we already reserved memory above for the case where we
* are not replacing an exisiting timeout.
*/
if (ev->ev_flags & EVLIST_TIMEOUT)
event_queue_remove(base, ev, EVLIST_TIMEOUT);
 
/* Check if it is active due to a timeout.  Rescheduling
* this timeout before the callback can be executed
* removes it from the active list. */
if ((ev->ev_flags & EVLIST_ACTIVE) &&
    (ev->ev_res & EV_TIMEOUT)) {
/* See if we are just active executing this
* event in a loop
*/
if (ev->ev_ncalls && ev->ev_pncalls) {
/* Abort loop */
*ev->ev_pncalls = 0;
}
event_queue_remove(base, ev, EVLIST_ACTIVE);
}
 
gettime(base, &now);
evutil_timeradd(&now, tv, &ev->ev_timeout);
 
event_debug((
"event_add: timeout in %ld seconds, call %p",
tv->tv_sec, ev->ev_callback));
 
event_queue_insert(base, ev, EVLIST_TIMEOUT);
}
 
return (res);
}

4.event_base_dispatch()   循环、检测、分发事件

event_base_dispatch仅调用了event_base_loop函数;

int
event_base_dispatch(struct event_base *event_base)
{
  return (event_base_loop(event_base, 0));
}

5.event_base_loop()

event_base_loop()主要就是循环、检测、分发事件

1.信号标记被设置，则调用信号的回调函数
2.根据定时器最小时间，设置I/O多路复用的最大等待时间，这样即使没有I/O事件发生，也能在最小定时器超时时返回。
3.调用I/O多路复用，监听事件，将活跃事件添加到活跃事件链表中
4.检查定时事件，将就绪的定时事件从小根堆中删除，插入到活跃事件链表中

libevent的核心就event_base_loop()；在这其中检测和分发通过I/O多路复用来完成，比如我们经常使用的poll和epoll，通过epoll.c就可以看到源码；其实原理与我们之前学习到的epoll编程是很类似的，只是多了一部分的处理方式，达到与整合系统互相呼应的效果；

int
event_base_loop(struct event_base *base, int flags)
{
//IO复用方式
const struct eventop *evsel = base->evsel;
void *evbase = base->evbase;
struct timeval tv;
struct timeval *tv_p;
int res, done;
 
/* clear time cache */
base->tv_cache.tv_sec = 0;
 
if (base->sig.ev_signal_added)
evsignal_base = base;
done = 0;
while (!done) {
/* Terminate the loop if we have been asked to */
if (base->event_gotterm) {
//设置中止循环
base->event_gotterm = 0;
break;
}
 
if (base->event_break) {
base->event_break = 0;
break;
}
 
/* You cannot use this interface for multi-threaded apps */
while (event_gotsig) {
event_gotsig = 0;
if (event_sigcb) {
res = (*event_sigcb)();
if (res == -1) {
errno = EINTR;
return (-1);
}
}
}
 
timeout_correct(base, &tv);
 
tv_p = &tv;
if (!base->event_count_active && !(flags & EVLOOP_NONBLOCK)) {
timeout_next(base, &tv_p);
} else {
/* 
* if we have active events, we just poll new events
* without waiting.
*/
evutil_timerclear(&tv);
}
/* If we have no events, we just exit */
if (!event_haveevents(base)) {
event_debug(("%s: no events registered.", __func__));
return (1);
}
 
/* update last old time */
gettime(base, &base->event_tv);
 
/* clear time cache */
base->tv_cache.tv_sec = 0;
 
res = evsel->dispatch(base, evbase, tv_p);
 
if (res == -1)
return (-1);
gettime(base, &base->tv_cache);
 
timeout_process(base);
//有就绪事件则调用事件注册的回调函数
if (base->event_count_active) {
event_process_active(base);
if (!base->event_count_active && (flags & EVLOOP_ONCE))
done = 1;
} else if (flags & EVLOOP_NONBLOCK)
done = 1;
}
 
/* clear time cache */
base->tv_cache.tv_sec = 0;
 
event_debug(("%s: asked to terminate loop.", __func__));
return (0);
}

TCP服务器模型

1.evconnlistener

基于event和event_base已经可以写一个C/S模型了。但是对于服务器端来说，仍然需要用户自行调用socket、bind、listen、accept等步骤。这个过程有点繁琐，并且一些细节可能考虑不全，为此Libevent推出了一些对应的封装函数，简化了整个监听的流程，用户仅仅需要在对应回调函数里面处理已完成连接的套接字即可。

常用API

evconnlistener_new_bind:
通过evconnlistener_new_bind传递回调函数，在accept成功后，在回调函数里面处理已连接的套接字。evconnlistener其实是对even_base和event的封装。

evconnlistener具体使用可参考:

Libevent之evconnlistener详解_evconnlistener_new_bind_阿卡基YUAN的博客-CSDN博客

evconnlistener有关介绍与使用_Xiezongyi的博客-CSDN博客

2.bufferevent

Bufferevent主要是用来管理和调度IO事件,负责数据的read和write, 因为do_read方法作为一个事件会一直被循环, 当客户端连接断开的时候，do_read方法还是在循环，根本不知道客户端已经断开socket的连接。如果要解决这个问题，我们可能要做大量的工作来维护这些socket的连接状态，读取状态。而Libevent的Bufferevent帮我们解决了这些问题。Bufferevent封装了recv和send函数，并且设置了水位，有两种水位：低水位和高水位。低水位为0是默认值，表示收到了就读了，当设置了低水位（下界）的值，收到了这么多的大小才会去处理，没有到达低水位的字节数的话就一直不处理。高水位的默认值也是0，设置高水位（上界）超过这个值的话就要分批处理。另外, Bufferevent还解决了各种粘包和拆包问题。

struct bufferevent* evbuf=bufferevent_socket_new(connectedbase,fd,BEV_OPT_CLOSE_ON_FREE|BEV_OPT_THREADSAFE);
 bufferevent_setcb(evbuf,receive_read_cb,NULL,event_error_cb,NULL);
 bufferevent_setwatermark(evbuf,EV_WRITE,0,SEND_BUFFERCACHESIZE);
 bufferevent_setwatermark(evbuf,EV_READ,0,RECEIVE_BUFFERCACHESIZE);
// 0就是默认值，收到了就读

bufferevent具体使用可参考:

libevent学习——buffevent事件及低水位高水位设置_bufferevent_setwatermark_dxgzg的博客-CSDN博客

基于Libevent实现的TCP服务器的部分代码: 

int TCPServer::initServer()
{
    cout<<" init tcp server"<<endl;
    int status=0;
    // event_enable_debug_mode();
    evthread_use_pthreads();//启动线程安全
    ebase= event_base_new();//初始化base
    if(ebase)
    {
        cout<<" event init success"<<endl;
        memset(&esin,0,sizeof(esin));
        esin.sin_family=AF_INET;
        esin.sin_addr.s_addr=htonl(0);
        esin.sin_port=htons(ServerMemoryCache::getInstance()->configinfor->port);
        elistener=evconnlistener_new_bind(ebase,accept_conn_cb,NULL,LEV_OPT_CLOSE_ON_FREE|LEV_OPT_REUSEABLE,-1,
        (struct sockaddr*)&esin,sizeof(esin));
        if(elistener)
        {
            cout<<" evconnlistener_new_bind  success"<<endl;
            cout<<" evconnlistener_set_error_cb ..."<<endl;
            evconnlistener_set_error_cb(elistener,accept_error_cb);
            cout<<" evconnlistener_set_error_cb  success"<<endl;
            status=0;
        }
        else
        {
            cout<<" evconnlistener_new_bind  failed"<<endl;
            status=2;
            puts("event lister 创建失败!");
        }
    }
    else
    {
        cout<<"event init failed"<<endl;
        puts("event 打开失败!");
        status=1;
    }
    return status;
}
 
void TCPServer::accept_conn_cb(struct evconnlistener *listener,evutil_socket_t fd,
    struct sockaddr* address,int socklen,void *ctx)
{
 
    cout<<" receivce connect..."<<endl;
    struct event_base* connectedbase=evconnlistener_get_base(listener);
    struct bufferevent* evbuf=bufferevent_socket_new(connectedbase,fd,BEV_OPT_CLOSE_ON_FREE|BEV_OPT_THREADSAFE);
    ClientConnectinfo* coninfo=new ClientConnectinfo();
    time_t curtime;
    time(&curtime);
    bufferevent_setcb(evbuf,receive_read_cb,NULL,event_error_cb,NULL);
    bufferevent_setwatermark(evbuf,EV_WRITE,0,SEND_BUFFERCACHESIZE);
    bufferevent_setwatermark(evbuf,EV_READ,0,RECEIVE_BUFFERCACHESIZE);
    bufferevent_enable(evbuf,EV_WRITE|EV_READ);
    struct timeval tv;
    tv.tv_sec=BUFFEROVERTIME;
    tv.tv_usec=0;
    bufferevent_set_timeouts(evbuf,&tv,&tv);
    coninfo->lastreceivetime=curtime;
    coninfo->clientbuffer=evbuf;
    coninfo->fd=fd;
    ServerMemoryCache::getInstance()->connectedbuffers.push_back(coninfo);
 
    unique_lock<mutex> lkhaveclient(ServerMemoryCache::getInstance()->ishaveconnectedclientmtx);
    lkhaveclient.unlock();
    ServerMemoryCache::getInstance()->ishaveconnectedclientcv.notify_one();
 
    cout<<" new connect deal with finish..."<<endl;
}