libevent粘包分包解决方案：bufferevent + evbuffer

转自：http://blog.sina.com.cn/s/blog_9f1496990102vshz.html

 

原文：http://www.lvtao.net/c/631.html

Libevent介绍

libevent是一个事件触发的网络库，适用于windows、linux、bsd等多种平台，内部使用select、epoll、kqueue等系统调用管理事件机制。著名分布式缓存软件memcached也是libevent based，并且libevent在使用上能够作到跨平台，并且根据libevent官方网站上公布的数据统计，彷佛也有着非凡的性能。

libevent官方网站 http://libevent.org/

英文文档 http://www.wangafu.net/~nickm/libevent-book/

中文文档 http://www.cppblog.com/mysileng/category/20374.html

Libevent是基于事件的网络库。说的通俗点，例如个人客户端链接到服务端属于一个链接的事件，当这个事件触发的时候就会去处理。

该文章阅读过程当中，请结合下面的socket例子，可能会更加清晰的理解每个接口的用法。

event_base

1. 建立event_base

event_base是event（事件，后面会讲event）的一个集合。event_base中存放你是监听是否就绪的event。通常状况下一个线程一个event_base，多个线程的状况下须要开多个event_base。

event_base主要是用来管理和实现事件的监听循环。

通常状况下直接new一个event_base就能够知足大部分需求了，若是须要配置参数的，能够参见libevent官网。

建立方法：

struct event_base *event_base_new(void);

销毁方法：

void event_base_free(struct event_base *base);

从新初始化：

int event_reinit(struct event_base *base);

2. 查看IO模型

IO多路复用模型中，有多种方法能够供咱们选择，可是这些模型是在不一样的平台下面的： select  poll  epoll  kqueue  devpoll  evport  win32

当咱们建立一个event_base的时候，libevent会自动为咱们选择最快的IO多路复用模型，Linux下通常会用epoll模型。

下面这个方法主要是用来获取IO模型的名称。

const char *event_base_get_method(const struct event_base *base);

3. 销毁event_base

void event_base_free(struct event_base *base);

4. 事件循环 event loop

咱们上面说到 event_base是一组event的集合，咱们也能够将event事件注册到这个集合中。当须要事件监听的时候，咱们就须要对这个event_base进行循环。

下面这个函数很是重要，会在内部不断的循环监听注册上来的事件。

int event_base_dispatch(struct event_base *base);

返回值：0 表示成功退出  -1 表示存在错误信息。

还能够用这个方法：

#define EVLOOP_ONCE             0x01  #define EVLOOP_NONBLOCK         0x02  #define EVLOOP_NO_EXIT_ON_EMPTY 0x04  
  int event_base_loop(struct event_base *base, int flags);

event_base_loop这个方法会比event_base_dispatch这个方法更加灵活一些。

EVLOOP_ONCE： 阻塞直到有一个活跃的event，而后执行完活跃事件的回调就退出。

EVLOOP_NONBLOCK : 不阻塞，检查哪一个事件准备好，调用优先级最高的那一个，而后退出。

0:若是参数填了0，则只有事件进来的时候才会调用一次事件的回调函数，比较经常使用

事件循环中止的状况：

1. event_base中没有事件event

2. 调用event_base_loopbreak()，那么事件循环将中止

3. 调用event_base_loopexit()，那么事件循环将中止

4. 程序错误，异常退出

两个退出的方法：

// 这两个函数成功返回 0 失败返回 -1  // 指定在 tv 时间后中止事件循环  // 若是 tv == NULL 那么将无延时的中止事件循环  int event_base_loopexit(struct event_base *base,const struct timeval *tv);  
// 当即中止事件循环（而不是无延时的中止）  int event_base_loopbreak(struct event_base *base);

两个方法区别：

1. event_base_loopexit(base, NULL) 若是当前正在为多个活跃事件调用回调函数，那么不会当即退出，而是等到全部的活跃事件的回调函数都执行完成后才退出事件循环

2. event_base_loopbreak(base) 若是当前正在为多个活跃事件调用回调函数，那么当前正在调用的回调函数会被执行，而后立刻退出事件循环，而并不处理其余的活跃事件了

5. event_base的例子：

#include <<span>stdio.h>  #include <<span>stdlib.h>  #include <<span>unistd.h>  #include <<span>sys/types.h>   #include <<span>sys/socket.h>          #include <<span>netinet/in.h>          #include <<span>arpa/inet.h>  #include <<span>string.h>  #include <<span>fcntl.h>   #include <<span>event2/event.h>  #include <<span>event2/bufferevent.h>  int main() {  
puts("init a event_base!");  
struct event_base *base; //定义一个event_base  
base = event_base_new(); //初始化一个event_base  
const char *x =  event_base_get_method(base); //查看用了哪一个IO多路复用模型，linux一下用epoll  
printf("METHOD:%s\n", x);  
int y = event_base_dispatch(base); //事件循环。由于咱们这边没有注册事件，因此会直接退出  
event_base_free(base);  //销毁libevent  
return 1;  
}

返回：

event 事件

event_base是事件的集合，负责事件的循环，以及集合的销毁。而event就是event_base中的基本单元：事件。

咱们举一个简单的例子来理解事件。例如咱们的socket来进行网络开发的时候，都会使用accept这个方法来阻塞监听是否有客户端socket链接上来，若是客户端链接上来，则会建立一个线程用于服务端与客户端进行数据的交互操做，而服务端会继续阻塞等待下一个客户端socket链接上来。客户端链接到服务端实际就是一种事件。

1. 建立一个事件event

struct event *event_new(struct event_base *base, evutil_socket_t fd,short what, event_callback_fn cb,void *arg);

参数：

1. base：即event_base

2. fd：文件描述符。

3. what：event关心的各类条件。

4. cb：回调函数。

5. arg：用户自定义的数据，能够传递到回调函数中去。

libevent是基于事件的，也就是说只有在事件到来的这种条件下才会触发当前的事件。例如：

1. fd文件描述符已准备好可写或者可读

2. fd立刻就准备好可写和可读。

3. 超时的状况 timeout

4. 信号中断

5. 用户触发的事件

what参数 event各类条件：

// 超时  #define EV_TIMEOUT 0x01  // event 相关的文件描述符能够读了  #define EV_READ 0x02  // event 相关的文件描述符能够写了  #define EV_WRITE 0x04  // 被用于信号检测（详见下文）  #define EV_SIGNAL 0x08  // 用于指定 event 为 persistent 持久类型。当事件执行完毕后，不会被删除，继续保持pending等待状态;  // 若是是非持久类型，则回调函数执行完毕后，事件就会被删除，想要从新使用这个事件，就必须将这个事件继续添加event_add   #define EV_PERSIST 0x10  // 用于指定 event 会被边缘触发  #define EV_ET 0x20

2. 释放event_free

真正的释放event的内存。

void event_free(struct event *event);

event_del 清理event的内存。这个方法并非真正意义上的释放内存。

当函数会将事件转为 非pending和非activing的状态。

int event_del(struct event *event);

3. 注册event

该方法将用于向event_base注册事件。

参数：ev 为事件指针；tv 为时间指针。当tv = NULL的时候则无超时时间。

函数返回：0表示成功 -1 表示失败。

int event_add(struct event *ev, const struct timeval *tv);

tv时间结构例子：

struct timeval five_seconds = {5, 0};  
event_add(ev1, &five_seconds);

4.event_assign

event_new每次都会在堆上分配内存。有些场景下并非每次都须要在堆上分配内存的，这个时候咱们就能够用到event_assign方法。

已经初始化或者处于 pending 的 event，首先须要调用 event_del() 后再调用 event_assign()。这个时候就能够重用这个event了。

// 此函数用于初始化 event（包括能够初始化栈上和静态存储区中的 event）  // event_assign() 和 event_new() 除了 event 参数以外，使用了同样的参数  // event 参数用于指定一个未初始化的且须要初始化的 event  // 函数成功返回 0 失败返回 -1  int event_assign(struct event *event, struct event_base *base,evutil_socket_t fd, short what,void (*callback)(evutil_socket_t, short, void *), void *arg);  
       
// 相似上面的函数，此函数被信号 event 使用  event_assign(event, base, signum, EV_SIGNAL|EV_PERSIST, callback, arg)

5. 信号事件

信号事件也能够对信号进行事件的处理。用法和event_new相似。只不过处理的是信号而已

// base --- event_base  // signum --- 信号，例如 SIGHUP  // callback --- 信号出现时调用的回调函数  // arg --- 用户自定义数据  evsignal_new(base, signum, cb, arg)  
       
//将信号 event 注册到 event_base  evsignal_add(ev, tv)   
       
// 清理信号 event  evsignal_del(ev)

6. event细节

1. 每个事件event都须要经过event_new初始化生成。event_new生成的事件是在堆上分配的内存。

2. 当一个事件经过event_add被注册到event_base上的时候，这个事件处于pending（等待状态），当只有有事件进来的时候，event才会被激活active状态，相关的回调函数就会被调用。

3. persistent 若是event_new中的what参数选择了EV_PERSIST，则是持久的类型。持久的类型调用玩回调函数后，会继续转为pending状态，就会继续等待事件进来。大部分状况下会选择持久类型的事件。

4. 而非持久的类型的事件，调用玩一次以后，就会变成初始化的状态。这个时候须要调用event_add 继续将事件注册到event_base上以后才能使用。

Socket实例

#include   #include   #include   #include       #include       #include       #include      #include   #include    
  #include   #include   
  //读取客户端  void do_read(evutil_socket_t fd, short event, void *arg) {  
    //继续等待接收数据    
    char buf[1024];  //数据传送的缓冲区      
    int len;    
    if ((len = recv(fd, buf, 1024, 0)) > 0)  {    
        buf[len] = '\0';      
        printf("%s\n", buf);      
        if (send(fd, buf, len, 0) < 0) {    //将接受到的数据写回客户端  
            perror("write");      
        }  
    }   
}  
  
  
//回调函数，用于监听链接进来的客户端socket  void do_accept(evutil_socket_t fd, short event, void *arg) {  
    int client_socketfd;//客户端套接字      
    struct sockaddr_in client_addr; //客户端网络地址结构体     
    int in_size = sizeof(struct sockaddr_in);    
    //客户端socket    
    client_socketfd = accept(fd, (struct sockaddr *) &client_addr, ∈_size); //等待接受请求，这边是阻塞式的    
    if (client_socketfd < 0) {    
        puts("accpet error");    
        exit(1);  
    }    
  
    //类型转换  
    struct event_base *base_ev = (struct event_base *) arg;  
  
    //socket发送欢迎信息    
    char * msg = "Welcome to My socket";    
    int size = send(client_socketfd, msg, strlen(msg), 0);    
  
    //建立一个事件，这个事件主要用于监听和读取客户端传递过来的数据  
    //持久类型，而且将base_ev传递到do_read回调函数中去  
    struct event *ev;  
    ev = event_new(base_ev, client_socketfd, EV_TIMEOUT|EV_READ|EV_PERSIST, do_read, base_ev);  
    event_add(ev, NULL);  
}  
  
  
//入口主函数  int main() {  
  
    int server_socketfd; //服务端socket    
    struct sockaddr_in server_addr;   //服务器网络地址结构体      
    memset(&server_addr,0,sizeof(server_addr)); //数据初始化--清零      
    server_addr.sin_family = AF_INET; //设置为IP通讯      
    server_addr.sin_addr.s_addr = INADDR_ANY;//服务器IP地址--容许链接到全部本地地址上      
    server_addr.sin_port = htons(8001); //服务器端口号      
    
    //建立服务端套接字    
    server_socketfd = socket(PF_INET,SOCK_STREAM,0);    
    if (server_socketfd < 0) {    
        puts("socket error");    
        return 0;    
    }    
  
    evutil_make_listen_socket_reuseable(server_socketfd); //设置端口重用  
    evutil_make_socket_nonblocking(server_socketfd); //设置无阻赛  
    
    //绑定IP    
    if (bind(server_socketfd, (struct sockaddr *)&server_addr, sizeof(struct sockaddr))<<span>0) {    
        puts("bind error");    
        return 0;    
    }    
  
    //监听,监听队列长度 5    
    listen(server_socketfd, 10);    
      
    //建立event_base 事件的集合，多线程的话 每一个线程都要初始化一个event_base  
    struct event_base *base_ev;  
    base_ev = event_base_new();   
    const char *x =  event_base_get_method(base_ev); //获取IO多路复用的模型，linux通常为epoll  
    printf("METHOD:%s\n", x);  
  
    //建立一个事件，类型为持久性EV_PERSIST，回调函数为do_accept（主要用于监听链接进来的客户端）  
    //将base_ev传递到do_accept中的arg参数  
    struct event *ev;  
    ev = event_new(base_ev, server_socketfd, EV_TIMEOUT|EV_READ|EV_PERSIST, do_accept, base_ev);  
  
    //注册事件，使事件处于 pending的等待状态  
    event_add(ev, NULL);  
  
    //事件循环  
    event_base_dispatch(base_ev);  
  
    //销毁event_base  
    event_base_free(base_ev);    
    return 1;  
}

说明：

1. 必须设置socket为非阻塞模式，不然就会阻塞在那边，影响整个程序运行

evutil_make_listen_socket_reuseable(server_socketfd); //设置端口重用  vutil_make_socket_nonblocking(server_socketfd); //设置无阻塞

2. 咱们首选创建的事件主要用于监听客户端的连入。当客户端有socket链接到服务器端的时候，回调函数do_accept就会去执行；当空闲的时候，这个事件就会是一个pending等待状态，等待有新的链接进来，新的链接进来了以后又会继续执行。

struct event *ev;  
ev = event_new(base_ev, server_socketfd, EV_TIMEOUT|EV_READ|EV_PERSIST, do_accept, base_ev);

3. 在do_accept事件中咱们建立了一个新的事件，这个事件的回调函数是do_read。主要用来循环监听客户端上传的数据。do_read这个方法会一直循环执行，接收到客户端数据就会进行处理。

//建立一个事件，这个事件主要用于监听和读取客户端传递过来的数据  //持久类型，而且将base_ev传递到do_read回调函数中去  struct event *ev;  
ev = event_new(base_ev, client_socketfd, EV_TIMEOUT|EV_READ|EV_PERSIST, do_read, base_ev);  
event_add(ev, NULL);

Bufferevent

上面的socket例子估计通过测试估计你们就会有不少疑问：

1. do_read方法做为一个事件会一直被循环

2. 当客户端链接断开的时候，do_read方法仍是在循环，根本不知道客户端已经断开socket的链接。

3. 须要解决各类粘包和拆包（相关粘包拆包文章）问题

若是要解决这个问题，咱们可能要作大量的工做来维护这些socket的链接状态，读取状态。而Libevent的Bufferevent帮咱们解决了这些问题。

Bufferevent主要是用来管理和调度IO事件；而Evbuffer（下面一节会讲到）主要用来缓冲网络IO数据。

Bufferevent目前支持TCP协议，而不知道UDP协议。咱们这边也只讲TCP协议下的Bufferevent的使用。

咱们先看下下面的接口（而后结合下面改进socket的例子，本身动手去实验一下）：

1. 建立Bufferevent API

//建立一个Bufferevent  struct bufferevent *bufferevent_socket_new(struct event_base *base, evutil_socket_t fd, enum bufferevent_options options);

参数：

base：即event_base

fd：文件描述符。若是是socket的方法，则socket须要设置为非阻塞的模式。

options：行为选项，下面是行为选项内容

1. BEV_OPT_CLOSE_ON_FREE ：当 bufferevent 被释放同时关闭底层（socket 被关闭等） 通常用这个选项

2. BEV_OPT_THREADSAFE ：为 bufferevent 自动分配锁，这样可以在多线程环境中安全使用

3. BEV_OPT_DEFER_CALLBACKS ： 当设置了此标志，bufferevent 会延迟它的全部回调（参考前面说的延时回调）

4. BEV_OPT_UNLOCK_CALLBACKS ： 若是 bufferevent 被设置为线程安全的，用户提供的回调被调用时 bufferevent 的锁会被持有。若是设置了此选项，Libevent 将在调用你的回调时释放 bufferevent 的锁

2. 释放Bufferevent

void bufferevent_free(struct bufferevent *bev);

若是设置了延时回调BEV_OPT_DEFER_CALLBACKS，则释放会在延时回调调用了回调函数以后，才会真正释放。

3. 设置Bufferevent的回调函数和相关设置

前面咱们说过了，使用了Bufferevent以后，Libevent会帮咱们托管三种事件：1. 读取事件  2. 写入事件  3. 处理事件

咱们先看一下回调函数结构：

1. 读取和写入的回调函数结构，其中 ctx为通用传递的参数

typedef void (*bufferevent_data_cb)(struct bufferevent *bev, void *ctx);

2. 事件回调，即链接断开、错误处理等回调。其中ctx为通用传递的参数。

events参数为事件，用户能够在回调函数中拿到这个事件来进行事务处理的判断：

1. BEV_EVENT_READING   在 bufferevent 上进行读取操做时出现了一个事件

2. BEV_EVENT_WRITING  在 bufferevent 上进行写入操做时出现了一个事件

3. BEV_EVENT_ERROR  进行 bufferevent 操做时出错

4. BEV_EVENT_TIMEOUT  在 bufferevent 上出现了超时

5. BEV_EVENT_EOF  在 bufferevent 上遇到了文件结束符，链接断开

6. BEV_EVENT_CONNECTED 在 bufferevent 上请求链接完成了

typedef void (*bufferevent_event_cb)(struct bufferevent *bev, short events, void *ctx);

3. 在bufferevent上设置回调函数。

bufev：bufferevent_socket_new建立的bufferevent

readcb：读取事件的回调函数，没有则能够为NULL

writecb：写入事件的回调函数，没有则能够为NULL

eventcb：事件函数的回调函数，没有则能够为NULL，通常咱们能够在这里面判断链接断开等。

cbarg：公用传输的传递

经过这个函数，咱们就能够设置咱们须要的一些回调函数信息。

void bufferevent_setcb(struct bufferevent *bufev,bufferevent_data_cb readcb,  
bufferevent_data_cb writecb,bufferevent_event_cb eventcb,void *cbarg);

取回回调函数:

void bufferevent_getcb(struct bufferevent *bufev,bufferevent_data_cb *readcb_ptr,bufferevent_data_cb *writecb_ptr,  
bufferevent_event_cb *eventcb_ptr,void **cbarg_ptr)

4. 设置Bufferevent事件的类型

bufferevent_enable(bev, EV_READ|EV_WRITE|EV_PERSIST);

5. 水位设置。

水位设置能够这么理解，bufferevent至关于一个水位容器，其中参数：

events：EV_READ 则为设置读取事件；EV_WRITE 则为写入事件。EV_READ |  EV_WRITE 为设置二者的水位。

lowmark：最低水位，默认为0。这个参数很是重要，例如lowmark设置为10，则当bufferevent容器中有10个字符的时候才会去调用readcb这个回调函数。

void bufferevent_setwatermark(struct bufferevent *bufev, short events,size_t lowmark, size_t highmark);

6. 下面能够看一个设置和回调函数例子：

//建立一个bufferevent  struct bufferevent *bev = bufferevent_socket_new(base_ev, client_socketfd, BEV_OPT_CLOSE_ON_FREE);  
//设置读取方法和error时候的方法  bufferevent_setcb(bev, read_cb, NULL, error_cb, base_ev);     
//设置类型  bufferevent_enable(bev, EV_READ|EV_WRITE|EV_PERSIST);  
//设置水位  bufferevent_setwatermark(bev, EV_READ, 0, 0);  
//读取事件回调函数  void read_cb(struct bufferevent *bev, void *arg) {  
#define MAX_LINE     256  char line[MAX_LINE+1];  
int n;  
evutil_socket_t fd = bufferevent_getfd(bev);  
while (n = bufferevent_read(bev, line, MAX_LINE), n > 0) {  
line[n] = '\0';  
printf("fd=%u, read line: %s\n", fd, line);  
bufferevent_write(bev, line, n);  
}  
puts("haha");  
}  
//写入事件回调函数  void write_cb(struct bufferevent *bev, void *arg) {}  
//事件回调  void error_cb(struct bufferevent *bev, short event, void *arg) {  
evutil_socket_t fd = bufferevent_getfd(bev);  
printf("fd = %u, ", fd);  
if (event & BEV_EVENT_TIMEOUT) {  
printf("Timed out\n");  
} else if (event & BEV_EVENT_EOF) {  
printf("connection closed\n");  
} else if (event & BEV_EVENT_ERROR) {  
printf("some other error\n");  
}  
bufferevent_free(bev);  
}

4. 输入输出相关函数

1. 获取buffer：

// 获取到输入 buffer  struct evbuffer *bufferevent_get_input(struct bufferevent *bufev);  
// 获取到输出 buffer  struct evbuffer *bufferevent_get_output(struct bufferevent *bufev);

2. 写入和输出函数，成功返回0，失败返回-1：

bufev：bufferevent

data：写入的字符串数据

size：字符长度

//写入  int bufferevent_write(struct bufferevent *bufev, const void *data, size_t size);  
//输出  size_t bufferevent_read(struct bufferevent *bufev, void *data, size_t size);

3. 写入输出函数2：

bufev：bufferevent

buf：buffer块  下面会讲到evbuffer的使用

int bufferevent_write_buffer(struct bufferevent *bufev, struct evbuffer *buf);  
int bufferevent_read_buffer(struct bufferevent *bufev,struct evbuffer *buf);

使用Bufferevent后的Socket例子：

上面咱们已经介绍完了Bufferevent的相关API，能够看下具体例子。

#include   #include   #include   #include       #include       #include       #include      #include   #include    
  #include   #include   
  void read_cb(struct bufferevent *bev, void *arg) {  
    #define MAX_LINE    256  
    char line[MAX_LINE+1];  
    int n;  
    evutil_socket_t fd = bufferevent_getfd(bev);  
    while (n = bufferevent_read(bev, line, MAX_LINE), n > 0) {  
        line[n] = '\0';  
        printf("fd=%u, read line: %s\n", fd, line);  
        bufferevent_write(bev, line, n);  
    }  
    puts("haha");  
}  
void write_cb(struct bufferevent *bev, void *arg) {}  
void error_cb(struct bufferevent *bev, short event, void *arg) {  
    evutil_socket_t fd = bufferevent_getfd(bev);  
    printf("fd = %u, ", fd);  
    if (event & BEV_EVENT_TIMEOUT) {  
        printf("Timed out\n");  
    } else if (event & BEV_EVENT_EOF) {  
        printf("connection closed\n");  
    } else if (event & BEV_EVENT_ERROR) {  
        printf("some other error\n");  
    }  
    bufferevent_free(bev);  
}  
  
//回调函数，用于监听链接进来的客户端socket  void do_accept(evutil_socket_t fd, short event, void *arg) {  
    int client_socketfd;//客户端套接字      
    struct sockaddr_in client_addr; //客户端网络地址结构体     
    int in_size = sizeof(struct sockaddr_in);    
    //客户端socket    
    client_socketfd = accept(fd, (struct sockaddr *) &client_addr, ∈_size); //等待接受请求，这边是阻塞式的    
    if (client_socketfd < 0) {    
        puts("accpet error");    
        exit(1);  
    }    
  
    //类型转换  
    struct event_base *base_ev = (struct event_base *) arg;  
  
    //socket发送欢迎信息    
    char * msg = "Welcome to My socket";    
    int size = send(client_socketfd, msg, strlen(msg), 0);    
  
    //建立一个事件，这个事件主要用于监听和读取客户端传递过来的数据  
    //持久类型，而且将base_ev传递到do_read回调函数中去  
    //struct event *ev;  
    //ev = event_new(base_ev, client_socketfd, EV_TIMEOUT|EV_READ|EV_PERSIST, do_read, base_ev);  
    //event_add(ev, NULL);  
  
    //建立一个bufferevent  
    struct bufferevent *bev = bufferevent_socket_new(base_ev, client_socketfd, BEV_OPT_CLOSE_ON_FREE);  
    //设置读取方法和error时候的方法  
    bufferevent_setcb(bev, read_cb, NULL, error_cb, base_ev);    
    //设置类型  
    bufferevent_enable(bev, EV_READ|EV_WRITE|EV_PERSIST);  
    //设置水位，每次接受10个字符  
    bufferevent_setwatermark(bev, EV_READ, 0, 10);  
}  
  
  
//入口主函数  int main() {  
  
    int server_socketfd; //服务端socket    
    struct sockaddr_in server_addr;   //服务器网络地址结构体      
    memset(&server_addr,0,sizeof(server_addr)); //数据初始化--清零      
    server_addr.sin_family = AF_INET; //设置为IP通讯      
    server_addr.sin_addr.s_addr = INADDR_ANY;//服务器IP地址--容许链接到全部本地地址上      
    server_addr.sin_port = htons(8001); //服务器端口号      
    
    //建立服务端套接字    
    server_socketfd = socket(PF_INET,SOCK_STREAM,0);    
    if (server_socketfd < 0) {    
        puts("socket error");    
        return 0;    
    }    
  
    evutil_make_listen_socket_reuseable(server_socketfd); //设置端口重用  
    evutil_make_socket_nonblocking(server_socketfd); //设置无阻赛  
    
    //绑定IP    
    if (bind(server_socketfd, (struct sockaddr *)&server_addr, sizeof(struct sockaddr))<<span>0) {    
        puts("bind error");    
        return 0;    
    }    
  
    //监听,监听队列长度 5    
    listen(server_socketfd, 10);    
      
    //建立event_base 事件的集合，多线程的话 每一个线程都要初始化一个event_base  
    struct event_base *base_ev;  
    base_ev = event_base_new();   
    const char *x =  event_base_get_method(base_ev); //获取IO多路复用的模型，linux通常为epoll  
    printf("METHOD:%s\n", x);  
  
    //建立一个事件，类型为持久性EV_PERSIST，回调函数为do_accept（主要用于监听链接进来的客户端）  
    //将base_ev传递到do_accept中的arg参数  
    struct event *ev;  
    ev = event_new(base_ev, server_socketfd, EV_TIMEOUT|EV_READ|EV_PERSIST, do_accept, base_ev);  
  
    //注册事件，使事件处于 pending的等待状态  
    event_add(ev, NULL);  
  
    //事件循环  
    event_base_dispatch(base_ev);  
  
    //销毁event_base  
    event_base_free(base_ev);    
    return 1;  
}

Evbuffer IO缓冲

上面讲了Bufferevent主要用于事件的管理和调度IO。而Evbuffer给咱们提供了很是实用的IO缓存工具。

上一个例子中，虽然解决了断开链接、读取事件等IO管理的工做，可是也是存在缺陷的。

1. 由于TCP粘包拆包的缘由，咱们不知道一次接收到的数据是不是完整的。

2. 咱们没法根据客户端传递过来的数据来分析客户端的请求信息。根据上面的问题，咱们可能会考虑设计一个缓冲容器，这个容器主要用来不停得接收客户端传递过来的数据信息，而且要等到信息量接收到必定的程度的时候，咱们对客户端的信息进行分析处理，最后才能知道客户端的请求内容。若是本身作这个缓冲容器，恐怕是须要花费不少的时间，而Libevent已经给咱们设计了Evbuffer，咱们能够直接使用Evbuffer缓冲容器来知足咱们的业务需求。

evbuffer结构：

struct evbuffer{  
  // 当前有效缓冲区的内存起始地址  
 u_char *buffer;   
  // 整个分配(realloc)用来缓冲的内存起始地址  
  u_char *orig_buffer;   
  // origin_buffer和buffer之间的字节数  
 size_t misalign;   
  // 整个分配用来缓冲的内存字节数  
 size_t totallen;   
  // 当前有效缓冲区的长度(字节数)  
 size_t off;   
  //回到函数，当缓冲区有变化的时候会被调用  
 void (*cb)(struct evbuffer *, size_t, size_t, void *);  
  //回调函数的参数  
 void *cbarg;   
};

libevent的缓冲是一个连续的内存区域，其处理数据的方式(写数据和读数据)更像一个队列操做方式：从后写入，从前

读出。evbuffer分别设置相关指针(一个指标)用于指示读出位置和写入位置。其大体结构如图：

orig_buffer指向由realloc分配的连续内存区域，buffer指向有效数据的内存区域，totallen表示orig_buffer指向的内存

区域的大小，misalign表示buffer相对于orig_buffer的偏移，off表示有效数据的长度。

下面是一些基础的和最经常使用的API，详细的API设计，仍是请翻看官方网站：

1.  建立和销毁Evbuffer

struct evbuffer *evbuffer_new(void);  
void evbuffer_free(struct evbuffer *buf);

2. 线程锁

int evbuffer_enable_locking(struct evbuffer *buf, void *lock);  
void evbuffer_lock(struct evbuffer *buf);  
void evbuffer_unlock(struct evbuffer *buf);

3. 检查buffer长度，比较经常使用

size_t evbuffer_get_length(const struct evbuffer *buf);

返回的是buffer中的字节数。

4. 向buffer中添加数据，经常使用

int evbuffer_add(struct evbuffer *buf, const void *data, size_t datlen);

这个函数添加data处的datalen字节到buf的末尾，成功时返回0，失败时返回-1。

int evbuffer_expand(struct evbuffer *buf, size_t datlen);

这个函数修改缓冲区的最后一块，或者添加一个新的块，使得缓冲区足以容纳datlen字节，而不须要更多的内存分配。

int evbuffer_prepend(struct evbuffer *buf, const void *data, size_t size);  
int evbuffer_prepend_buffer(struct evbuffer *dst, struct evbuffer* src);

除了将数据移动到目标缓冲区前面以外，这两个函数的行为分别与evbuffer_add()和evbuffer_add_buffer()相同。

使用这些函数时要小心，永远不要对与bufferevent共享的evbuffer使用。这些函数是2.0.1-alpha版本新添加的。

5. 删除和移动buffer中的内容

int evbuffer_remove(struct evbuffer *buf, void *data, size_t datlen);

evbuffer_remove（）函数从buf前面复制和移除datlen字节到data处的内存中。若是可用字节少于datlen，函数复制全部字节。失败时返回-1，不然返回复制了的字节数。

int evbuffer_add_buffer(struct evbuffer *dst, struct evbuffer *src);  
int evbuffer_remove_buffer(struct evbuffer *src, struct evbuffer *dst,  
    size_t datlen);

evbuffer_add_buffer()将src中的全部数据移动到dst末尾，成功时返回0，失败时返回-1。

evbuffer_remove_buffer()函数从src中移动datlen字节到dst末尾，尽可能少进行复制。若是字节数小于datlen，全部字节被移动。函数返回移动的字节数。

evbuffer_add_buffer()在0.8版本引入；evbuffer_remove_buffer()是2.0.1-alpha版本新增长的。

6. 搜索buffer中的内容，经常使用

struct evbuffer_ptr {  
ev_ssize_t pos;  
struct {  
  } _internal;  
};  
struct evbuffer_ptr evbuffer_search(struct evbuffer *buffer,  
const char *what, size_t len, const struct evbuffer_ptr *start);  
struct evbuffer_ptr evbuffer_search_range(struct evbuffer *buffer,  
const char *what, size_t len, const struct evbuffer_ptr *start,  
const struct evbuffer_ptr *end);  
struct evbuffer_ptr evbuffer_search_eol(struct evbuffer *buffer,  
struct evbuffer_ptr *start, size_t *eol_len_out,  
enum evbuffer_eol_style eol_style);

结构evbuffer_ptr中的pos为偏移量，若是为-1则没查询到，大于-1，则搜索到了匹配的位置。

1. evbuffer_search()函数在缓冲区中查找含有len个字符的字符串what。函数返回包含字符串位置，或者在没有找到字符串时包含-1的evbuffer_ptr结构体。若是提供了start参数，则从指定的位置开始搜索；不然，从开始处进行搜索。

2. evbuffer_search_range()函数和evbuffer_search行为相同，只是它只考虑在end以前出现的what。

3. evbuffer_search_eol()函数像evbuffer_readln()同样检测行结束，可是不复制行，而是返回指向行结束符的evbuffer_ptr。若是eol_len_out非空，则它被设置为EOL字符串长度。

7. 面向行的读取

不少互联网协议都是基于行的。evbuffer_readln()函数从evbuffer前面取出一行，用一个新分配的空字符结束的字符串返回这一行。若是n_read_out不是NULL，则它被设置为返回的字符串的字节数。若是没有整行供读取，函数返回空。返回的字符串不包括行结束符。

enum evbuffer_eol_style {  
EVBUFFER_EOL_ANY,  
EVBUFFER_EOL_CRLF,  
EVBUFFER_EOL_CRLF_STRICT,  
EVBUFFER_EOL_LF,  
EVBUFFER_EOL_NUL  
};  
char *evbuffer_readln(struct evbuffer *buffer, size_t *n_read_out,  
    enum evbuffer_eol_style eol_style);

1. EVBUFFER_EOL_LF：行尾是单个换行符（也就是\n，ASCII值是0x0A）

2. EVBUFFER_EOL_CRLF_STRICT：行尾是一个回车符，后随一个换行符（也就是\r\n，ASCII值是0x0D 0x0A）

3. EVBUFFER_EOL_CRLF：行尾是一个可选的回车，后随一个换行符（也就是说，能够是\r\n或者\n）。这种格式对于解析基于文本的互联网协议颇有用，由于标准一般要求\r\n的行结束符，而不遵循标准的客户端有时候只使用\n。

4. EVBUFFER_EOL_ANY：行尾是任意数量、任意次序的回车和换行符。这种格式不是特别有用。它的存在主要是为了向后兼容。

例子：

//readline  char * rline;  
size_t len;  
rline = evbuffer_readln(buf, &len, EVBUFFER_EOL_CRLF);  
puts("Hello");  
if (rline != NULL) {  
    bufferevent_write_buffer(bev, buf); //使用buffer的方式输出结果  }

8. 复制数据

ev_ssize_t evbuffer_copyout(struct evbuffer *buf, void *data, size_t datlen);  
ev_ssize_t evbuffer_copyout_from(struct evbuffer *buf,  
     const struct evbuffer_ptr *pos,  
     void *data_out, size_t datlen);

evbuffer_copyout（）的行为与evbuffer_remove（）相同，可是它不从缓冲区移除任何数据。也就是说，它从buf前面复制datlen字节到data处的内存中。若是可用字节少于datlen，函数会复制全部字节。失败时返回-1，不然返回复制的字节数。

若是从缓冲区复制数据太慢，可使用evbuffer_peek（）。

使用Evbuffer优化后的例子

#include   #include   #include   #include       #include       #include       #include      #include   #include    
  #include   #include   #include   #define MAX_LINE    256  
  void read_cb(struct bufferevent *bev, void *arg) {  
    struct evbuffer *buf = (struct evbuffer *)arg;  
    char line[MAX_LINE+1];  
    int n;  
    evutil_socket_t fd = bufferevent_getfd(bev);  
    while (n = bufferevent_read(bev, line, MAX_LINE), n > 0) {  
        line[n] = '\0';  
          
        //将读取到的内容放进缓冲区  
        evbuffer_add(buf, line, n);  
  
        //搜索匹配缓冲区中是否有==，==号来分隔每次客户端的请求  
        const char *x = "==";  
        struct evbuffer_ptr ptr = evbuffer_search(buf, x, strlen(x), 0);     
        if (ptr.pos != -1) {  
            bufferevent_write_buffer(bev, buf); //使用buffer的方式输出结果  
        }  
    }  
}  
void write_cb(struct bufferevent *bev, void *arg) {}  
void error_cb(struct bufferevent *bev, short event, void *arg) {  
    evutil_socket_t fd = bufferevent_getfd(bev);  
    printf("fd = %u, ", fd);  
    if (event & BEV_EVENT_TIMEOUT) {  
        printf("Timed out\n");  
    } else if (event & BEV_EVENT_EOF) {  
        printf("connection closed\n");  
    } else if (event & BEV_EVENT_ERROR) {  
        printf("some other error\n");  
    }  
    //清空缓冲区  
    struct evbuffer *buf = (struct evbuffer *)arg;  
    evbuffer_free(buf);  
    bufferevent_free(bev);  
}  
  
//回调函数，用于监听链接进来的客户端socket  void do_accept(evutil_socket_t fd, short event, void *arg) {  
    int client_socketfd;//客户端套接字      
    struct sockaddr_in client_addr; //客户端网络地址结构体     
    int in_size = sizeof(struct sockaddr_in);    
    //客户端socket    
    client_socketfd = accept(fd, (struct sockaddr *) &client_addr, ∈_size); //等待接受请求，这边是阻塞式的    
    if (client_socketfd < 0) {    
        puts("accpet error");    
        exit(1);  
    }    
  
    //类型转换  
    struct event_base *base_ev = (struct event_base *) arg;  
  
    //socket发送欢迎信息    
    char * msg = "Welcome to My socket";    
    int size = send(client_socketfd, msg, strlen(msg), 0);    
  
    //建立一个事件，这个事件主要用于监听和读取客户端传递过来的数据  
    //持久类型，而且将base_ev传递到do_read回调函数中去  
    //struct event *ev;  
    //ev = event_new(base_ev, client_socketfd, EV_TIMEOUT|EV_READ|EV_PERSIST, do_read, base_ev);  
    //event_add(ev, NULL);  
  
    //建立一个evbuffer，用来缓冲客户端传递过来的数据  
    struct evbuffer *buf = evbuffer_new();  
    //建立一个bufferevent  
    struct bufferevent *bev = bufferevent_socket_new(base_ev, client_socketfd, BEV_OPT_CLOSE_ON_FREE);  
    //设置读取方法和error时候的方法，将buf缓冲区当参数传递  
    bufferevent_setcb(bev, read_cb, NULL, error_cb, buf);    
    //设置类型  
    bufferevent_enable(bev, EV_READ|EV_WRITE|EV_PERSIST);  
    //设置水位  
    bufferevent_setwatermark(bev, EV_READ, 0, 0);  
}  
  
  
//入口主函数  int main() {  
  
    int server_socketfd; //服务端socket    
    struct sockaddr_in server_addr;   //服务器网络地址结构体      
    memset(&server_addr,0,sizeof(server_addr)); //数据初始化--清零      
    server_addr.sin_family = AF_INET; //设置为IP通讯      
    server_addr.sin_addr.s_addr = INADDR_ANY;//服务器IP地址--容许链接到全部本地地址上      
    server_addr.sin_port = htons(8001); //服务器端口号      
    
    //建立服务端套接字    
    server_socketfd = socket(PF_INET,SOCK_STREAM,0);    
    if (server_socketfd < 0) {    
        puts("socket error");    
        return 0;    
    }    
  
    evutil_make_listen_socket_reuseable(server_socketfd); //设置端口重用  
    evutil_make_socket_nonblocking(server_socketfd); //设置无阻赛  
    
    //绑定IP    
    if (bind(server_socketfd, (struct sockaddr *)&server_addr, sizeof(struct sockaddr))<<span>0) {    
        puts("bind error");    
        return 0;    
    }    
  
    //监听,监听队列长度 5    
    listen(server_socketfd, 10);    
      
    //建立event_base 事件的集合，多线程的话 每一个线程都要初始化一个event_base  
    struct event_base *base_ev;  
    base_ev = event_base_new();   
    const char *x =  event_base_get_method(base_ev); //获取IO多路复用的模型，linux通常为epoll  
    printf("METHOD:%s\n", x);  
  
    //建立一个事件，类型为持久性EV_PERSIST，回调函数为do_accept（主要用于监听链接进来的客户端）  
    //将base_ev传递到do_accept中的arg参数  
    struct event *ev;  
    ev = event_new(base_ev, server_socketfd, EV_TIMEOUT|EV_READ|EV_PERSIST, do_accept, base_ev);  
  
    //注册事件，使事件处于 pending的等待状态  
    event_add(ev, NULL);  
  
    //事件循环  
    event_base_dispatch(base_ev);  
  
    //销毁event_base  
    event_base_free(base_ev);    
    return 1;  
}

Util工具

Libevent还提供一些工具方法。这些方法能够简化咱们的开发。

1. 时间处理函数

//建立一个事件，类型为持久性EV_PERSIST，回调函数为do_accept（主要用于监听链接进来的客户端）  //将base_ev传递到do_accept中的arg参数  // 用于加或者减前两个参数，结果被保存在第三个参数中  #define evutil_timeradd(tvp, uvp, vvp)   #define evutil_timersub(tvp, uvp, vvp)   // 清除 timeval 将其值设置为 0  #define evutil_timerclear(tvp)   // 判断 timeval 是否为 0，若是是 0 返回 false，不然返回 true  #define evutil_timerisset(tvp)   // 比较两个 timeval  // 使用的时候这样用：  // evutil_timercmp(t1, t2, <=) 含义为判断 t1 <= t2 是否成立  // cmp 为全部的 C 关系操做符  #define evutil_timercmp(tvp, uvp, cmp)  // 获取当前时间并保存到 tv  // tz 目前无用  int evutil_gettimeofday(struct timeval *tv, struct timezone *tz);

2. Socket API

// 用于关闭一个 socket  int evutil_closesocket(evutil_socket_t s);  
#define EVUTIL_CLOSESOCKET(s) evutil_closesocket(s)  // 返回当前线程的最后一次 socket 操做的错误码  #define EVUTIL_SOCKET_ERROR()  // 改变当前 socket 的错误码  #define EVUTIL_SET_SOCKET_ERROR(errcode)  // 返回特定的 sock 的错误码  #define evutil_socket_geterror(sock)  // 经过 socket 错误码获取到一个字符串描述  #define evutil_socket_error_to_string(errcode)  // 设置 sock 为非阻塞的 socket  int evutil_make_socket_nonblocking(evutil_socket_t sock);  
// 设置 sock 的地址可重用  int evutil_make_listen_socket_reuseable(evutil_socket_t sock);

3. 字符串

// 它们对应于标准的 snprintf 和 vsnprintf  int evutil_snprintf(char *buf, size_t buflen, const char *format, ...);  
int evutil_vsnprintf(char *buf, size_t buflen, const char *format, va_list ap);

4. 安全的随机函数

// 此函数将使用随机的数据填充 n 个字节的 buf  void evutil_secure_rng_get_bytes(void *buf, size_t n);

附：一个客户端例子

#include     #include         #include         #include        #include     #include     #include     #include         int main() {      
int client_fd; //定义一个客户端的SOCKET      struct sockaddr_in server_addr; //服务器端        memset(&server_addr,0,sizeof(server_addr)); //数据初始化--清零    server_addr.sin_family=AF_INET; //设置为IP通讯          server_addr.sin_addr.s_addr = inet_addr("127.0.0.1");//服务器IP地址        server_addr.sin_port = htons(8001); //服务器端口号          client_fd = socket(PF_INET, SOCK_STREAM, 0);      
if (client_fd < 1) {  
puts("client socket error");       
return 0;     
}       
     int ret = connect(client_fd, (struct sockaddr *)&server_addr, sizeof(struct sockaddr));   
if (ret < 0) {        
puts("client connect error!");     
return 0;     
}       
char buf[1024];       
int len = recv(client_fd, buf, 1024, 0); //等待接收服务器端的数据     buf[len] = '\0';    
puts(buf);      
char *x = "Hello World,saodsadoosadosa==sadsad==";    
send(client_fd, x, strlen(x), 0); //发送数据     memset(buf, 0, 1024);  
int len2 = recv(client_fd, buf, 1024, 0); //继续接收服务端返回的数据   buf[len2] = '\0';   
puts(buf);      
shutdown(client_fd,2); //关闭socket    }