libevent高并发网络编程 - 02_libevent缓冲IO之bufferevent

1. 为什么需要缓冲区？

• libevent提供了一些高级别的网络编程抽象，其中包括缓冲式I/O的接口。在网络编程中，缓冲区是用来保存从网络中读取或向网络中写入数据的临时存储空间。因为在网络通信中，传输速度和数据处理速度往往不匹配，所以使用缓冲区可以帮助应用程序更好地控制数据流量，避免阻塞和死锁等问题。

• libevent的 bufferevent 接口就提供了对缓冲区操作的封装，使得应用程序可以更轻松地管理和操作缓冲区。具体而言，bufferevent 会自动管理底层文件描述符（socket）上的读写缓冲区，并提供事件驱动的回调机制来实现异步的缓冲区读写操作。此外，bufferevent 还支持多种协议栈，如 TCP、UDP 和 SSL 等，以适应不同的网络应用场景。

2. 水位

bufferevent 中的水位是指缓冲区中数据的数量达到或超过一定阈值时触发相应的事件。

具体而言，bufferevent 会维护两个水位，一个是读取的低水位（read low watermark），另一个是写入的高水位（write high watermark）。当缓冲区中的数据量高于读取低水位时，bufferevent 会触发读取事件；而当缓冲区中的数据量达到或超过写入高水位时，bufferevent 会暂停读取并触发写入事件，直到缓冲区中的数据量降到一定程度。

• 假设有一个 TCP 连接上的 bufferevent，读取低水位设置为 1024 字节。当接收到数据时，如果缓冲区中可读数据量达到或超过 1024 字节，则 bufferevent 会处于可读状态并触发读取事件；如果缓冲区中的可读数据量低于 1024 字节，则 bufferevent 不会触发读取事件，直到缓冲区中的可读数据量达到或超过 1024 字节时才开始触发读取事件。

• 假设有一个 TCP 连接上的 bufferevent，写入高水位设置为 4096 字节。当发送数据时，如果缓冲区中的可写数据量达到或超过 4096 字节，则 bufferevent 会处于不可写状态并触发写入事件；如果缓冲区中的可写数据量低于 4096 字节，则 bufferevent 不会触发写入事件，直到缓冲区中的可写数据量达到或超过 4096 字节时才开始触发写入事件。

触发写入事件不意味着立即将缓冲区中的数据全部写入网络中。实际上，bufferevent 会尽可能地将缓冲区中的数据写入网络中，但也要考虑到网络带宽、延迟和拥塞等因素，以确保数据传输的效率和可靠性。

3. bufferevent常用API

3.1 evconnlistener_new_bind()

evconnlistener_new_bind() 函数用于创建监听套接字，并将其添加到 libevent 的事件处理器中，等待客户端连接。

具体来说，它可以用于实现以下功能：

1. TCP 服务器：通过调用 evconnlistener_new_bind() 函数创建一个监听套接字，然后使用 listen() 函数开始监听客户端连接请求。当客户端连接到达时，libevent 将会自动触发相应的事件回调函数，可以在回调函数中进行新连接的处理和数据读写等操作。
2. UDP 服务器：虽然 UDP 是无连接的协议，但同样可以使用 evconnlistener_new_bind() 函数创建一个本地套接字并绑定到指定地址和端口上，以接收远程主机发送的 UDP 数据包。当有数据包到达时，libevent 将会触发相应的事件回调函数，开发者可以在回调函数中进行数据包的处理和响应等操作。
3. 其他网络应用：除了常见的 TCP/UDP 服务器外，evconnlistener_new_bind() 还可以用于创建各种类型的网络应用，例如 HTTP 服务器、WebSocket 服务器、DNS 服务器等。

总之，evconnlistener_new_bind() 函数是 libevent 库中实现高效异步网络编程所必需的重要接口之一，可以帮助更方便地创建、管理和处理网络连接。

函数原型：

struct evconnlistener *
evconnlistener_new_bind(struct event_base *base, evconnlistener_cb cb,void *ptr, unsigned flags, int backlog, const struct sockaddr *sa, int socklen)参数说明：base：事件处理器所使用的 event_base 对象；cb：新连接到来时调用的回调函数；ptr：传递给回调函数的参数指针；flags：监听套接字的标志位，可以使用 LEV_OPT_CLOSE_ON_FREE、LEV_OPT_REUSEABLE 和 LEV_OPT_THREADSAFE 进行设置；backlog：侦听套接字的挂起连接队列的最大长度；sa：sockaddr 结构体指针，表示要绑定的地址和端口号；socklen：地址结构体的长度。返回evconnlistener对象flags 参数用于设置监听套接字的属性，具体取值如下：LEV_OPT_CLOSE_ON_FREE：在 evconnlistener_free() 函数调用时自动关闭监听套接字，默认情况下不会关闭。LEV_OPT_REUSEABLE：设置监听套接字为可重用状态，即使在套接字被关闭后仍然可以重复使用同一个地址。LEV_OPT_THREADSAFE：开启线程安全模式，支持多个线程同时访问同一个 evconnlistener 对象。
例如，要同时设置可重用和自动关闭属性，可以将 flags 参数设置为 LEV_OPT_CLOSE_ON_FREE | LEV_OPT_REUSEABLE。需要注意的是，在使用多个 evconnlistener 对象进行端口监听时，如果想要复用同一个地址和端口号，必须要将所有的监听套接字都设置为可重用状态。evconnlistener_cb函数指针是libevent网络库中的一个回调函数指针类型，用于指向监听器接受新连接时的回调函数。该回调函数需要有如下形式：
void evconnlistener_cb(struct evconnlistener *listener, evutil_socket_t fd, struct sockaddr *addr, int socklen, void *ctx);

evconnlistener_cb函数指针是evconnlistener_new_bind()回调函数指针类型，用于指向监听器接受新连接时的回调函数。该回调函数需要有如下形式：

void evconnlistener_cb(struct evconnlistener *listener, evutil_socket_t fd, struct sockaddr *addr, int socklen, void *ctx);
参数说明：listener：指向evconnlistener监听器对象的指针。fd：表示新收到连接的socket文件描述符。addr：指向客户端地址信息的结构体指针。socklen：客户端地址结构体的长度。ctx：监听器上下文参数，在创建监听器时传入。
在evconnlistener_cb回调函数中，可以对新收到的连接进行处理，例如创建一个bufferevent缓冲事件或者直接进行数据读写等操作。

调用 evconnlistener_new_bind() 函数会创建一个新的 evconnlistener 对象，并将其添加到事件处理器中。该函数会自动创建并绑定一个服务器套接字，等待客户端连接。当有新的连接请求到达后，libevent 将会调用 cb 函数并传递连接套接字描述符作为参数，同时 ptr 指向 evconnlistener_new_bind() 中传递的参数。

3.2 evconnlistener_free()

evconnlistener_free() 函数用于释放一个已经创建的evconnlistener监听器对象，释放该对象所占用的内存空间。

void evconnlistener_free(struct evconnlistener *lev)

3.3 bufferevent_socket_new()

bufferevent_socket_new()用于创建一个基于socket的bufferevent缓冲事件对象的函数。将该缓冲事件对象与指定的socket文件描述符关联。通过该函数创建的缓冲事件对象来进行数据的读写操作，并且自动管理内部缓存区数据的处理。

• 该函数还支持设置多种缓冲策略和水位参数，以适应不同业务需求。例如，可以设置缓冲区大小、读写水位线等参数，来实现流控制和提高网络传输的效率。

• bufferevent_socket_new()函数还能够设置回调函数，处理缓冲区数据读写事件或者错误事件等。

函数的原型如下：

struct bufferevent *
bufferevent_socket_new(struct event_base *base, evutil_socket_t fd, int options)参数说明：base：指向event_base事件处理器对象的指针。fd：表示要关联到bufferevent的已经打开的socket文件描述符。options：表示缓冲事件的配置选项。返回值bufferevent对象options参数来设置缓冲事件的配置选项。常用的一些选项：BEV_OPT_CLOSE_ON_FREE：释放缓冲事件时自动关闭关联的socket文件描述符。BEV_OPT_THREADSAFE：将缓冲事件对象设为线程安全模式。BEV_OPT_DEFER_CALLBACKS：延迟回调机制，可以提高多个事件的处理效率。BEV_OPT_UNLOCK_CALLBACKS：解锁回调机制，可以支持多线程同时运行回调函数。BEV_EVENT_TIMEOUTBEV_EVENT_READING

3.4 bufferevent_enable()

bufferevent_enable()用于启用和关闭缓冲事件对象。将缓冲事件对象中开启对应的读、写模式，以便在有数据可读或可写时能够及时响应。

函数的原型如下：

int bufferevent_enable(struct bufferevent *bufev, short event)参数说明：bufev：指向要启用的bufferevent缓冲事件对象的指针。event：表示要启用的事件类型，可以是EV_READ、EV_WRITE或者EV_EVENT组合。 常用的event类型包括：EV_TIMEOUT：超时事件，用于实现定时器功能。EV_READ：读事件，表示该事件关心要读取的文件描述符是否可读。EV_WRITE：写事件，表示该事件关心要写入的文件描述符是否可写。EV_SIGNAL：信号事件，表示该事件与系统信号相关联。EV_PERSIST：持久性事件，表示注册的事件触发后不会自动被删除，而是需要手动删除。EV_ET：边缘触发事件，表示只有当事件状态发生变化时才通知监听者。EV_FINALIZE：最终化事件，表示事件处理器即将销毁之前，在内存释放时调用一次。EV_CLOSED：连接关闭事件，用于TCP连接管理。

bufferevent_enable()函数也可以关闭指定的缓冲事件对象，只需将event参数设置为0即可，例如：

bufferevent_enable(bufev, 0); // 关闭缓冲事件对象

3.5 bufferevent_set_timeouts()

bufferevent_set_timeouts()函数的作用是设置指定bufferevent缓冲事件对象的读写超时时间。该函数可以让在网络传输过程中，对数据读取或写入的响应时间进行限制，从而实现流控制和避免网络拥塞等问题。

该函数的原型如下：

int
bufferevent_set_timeouts(struct bufferevent *bufev,const struct timeval *tv_read,const struct timeval *tv_write)参数说明：bufev：指向要设置超时时间的bufferevent缓冲事件对象的指针。timeout_read：表示读超时时间的timeval结构体指针，如果为NULL，则不设置读超时时间。timeout_write：表示写超时时间的timeval结构体指针，如果为NULL，则不设置写超时时间。

3.6 bufferevent_setcb()

bufferevent_setcb()函数的作用是设置指定bufferevent缓冲事件对象的回调函数。该函数可以让在缓冲事件发生读写或者错误等事件时，及时进行处理和响应。

该函数的原型如下：

void
bufferevent_setcb(struct bufferevent *bufev,bufferevent_data_cb readcb, bufferevent_data_cb writecb,bufferevent_event_cb eventcb, void *cbarg)参数说明：bufev：指向要设置回调函数的bufferevent缓冲事件对象的指针。readcb：表示读取数据时的回调函数指针，当有数据可读时会调用该函数。writecb：表示写入数据时的回调函数指针，当缓冲区可写时会调用该函数。eventcb：表示异常事件（例如错误、连接关闭等）时的回调函数指针，当异常事件发生时会调用该函数。cbarg：表示回调函数的上下文参数，在调用回调函数时会传递给它。

readcb、writecb和eventcb这三个回调函数形式，如下所示：

void readcb(struct bufferevent *bev, void *ctx) {/* 处理读取缓冲区中的数据 */
}void writecb(struct bufferevent *bev, void *ctx) {/* 处理写入缓冲区的数据 */
}void eventcb(struct bufferevent *bev, short what, void *ctx) {if (what & BEV_EVENT_ERROR) {/* 处理错误事件 */}if (what & BEV_EVENT_TIMEOUT) {/* 处理超时事件 */}if (what & BEV_EVENT_EOF) {/* 处理连接关闭事件 */}
}bufferevent_setcb(bufev, readcb, writecb, eventcb, NULL);其中，bev参数表示发生事件的缓冲事件对象本身，ctx参数表示回调函数的上下文参数。

3.7 bufferevent_setwatermark()

bufferevent_setwatermark()函数的作用是设置指定bufferevent缓冲事件对象的读写水位线。该函数可以让网络传输过程中，控制缓冲区数据的读取和写入，以实现流控制和避免网络拥塞等问题。

该函数的原型如下：

void bufferevent_setwatermark(struct bufferevent *bufev,short events,size_t lowmark,size_t highmark);
参数说明：bufev：指向要设置水位线的bufferevent缓冲事件对象的指针。events：表示要设置水位线的事件类型，可以是BEV_READ、BEV_WRITE或者BEV_EVENT组合。lowmark：表示低水位线的大小，当缓冲区中数据量低于该值时会触发读取事件或可写事件。highmark：表示高水位线的大小，当缓冲区中数据量超过该值时会暂停读取事件或可写事件。

如果不想使用水位线机制，可以将lowmark和highmark参数都设为0。另外，bufferevent_setwatermark()函数只对BEV_READ和BEV_WRITE事件有效，对其他事件类型无效。

3.8 bufferevent_write()

bufferevent_write()函数的作用是向指定的bufferevent缓冲事件对象中写入数据。

该函数的原型如下：

int bufferevent_write(struct bufferevent *bufev, const void *data, size_t size)数说明：bufev：指向要写入数据的bufferevent缓冲事件对象的指针。data：表示要写入数据的缓冲区指针。size：表示要写入数据的长度。

需要注意的是，想要保证写入数据的可靠性，可以在调用bufferevent_write()函数之前，先调用bufferevent_flush()函数将写入的数据从内存缓冲中刷到socket文件描述符中去。

如果希望对写入的数据进行批量处理，可以使用bufferevent_write_buffer()函数，它可以将一个EVBUFFER缓冲区中的数据全部或部分地写入到指定的bufferevent缓冲事件对象中去，例如：

struct evbuffer *output = bufferevent_get_output(bufev);
evbuffer_add(output, "Hello, world!", 13);
bufferevent_write_buffer(bufev, output);

3.9 bufferevent_read()

bufferevent_read()函数的作用是从指定的bufferevent缓冲事件对象中读取数据。

该函数的原型如下：

size_t bufferevent_read(struct bufferevent *bufev, void *data, size_t size)参数说明：bufev：指向要读取数据的bufferevent缓冲事件对象的指针。data：表示接收数据的缓冲区指针。size：表示要读取数据的长度。

如果希望对读取的数据进行批量处理，可以使用bufferevent_read_buffer()函数，它可以将一个EVBUFFER缓冲区中的数据全部或部分地读取到指定的bufferevent缓冲事件对象中去，例如：

struct evbuffer *input = bufferevent_get_input(bufev);
size_t len = evbuffer_get_length(input);
char *data = (char*) malloc(len);if (bufferevent_read_buffer(bufev, input) == -1) {/* Handle error */
}evbuffer_remove(input, data, len);
/* 处理读取到的数据 */
free(data);

3.10 bufferevent_free()

bufferevent_free()函数的作用是释放指定的bufferevent缓冲事件对象。

该函数的原型如下：

void bufferevent_free(struct bufferevent *bufev);
参数说明：bufev：指向要释放的bufferevent缓冲事件对象的指针。

使用bufferevent_free()函数释放bufferevent缓冲事件对象后，相关的定时器、EV_READ、EV_WRITE、EV_SIGNAL等事件也会被取消，并且与之关联的EVBUFFER对象也会被销毁。因此，在调用bufferevent_free()函数之前，需要确保已经处理完相关事件和缓冲区中的数据。

3.11 bufferevent_trigger()

bufferevent_trigger()函数的作用是触发指定bufferevent缓冲事件对象的读写事件或异常事件。

该函数的原型如下：

void
bufferevent_trigger(struct bufferevent *bufev, short iotype, int options)参数说明：bufev：指向要触发事件的bufferevent缓冲事件对象的指针。events：表示要触发的事件类型，可以是EV_READ、EV_WRITE或者EV_EVENT组合。options：可选项，为0表示仅触发一次事件，为BEV_TRIG_DEFERRED表示延迟触发事件。

需要注意的是，bufferevent_trigger()函数并不会真正地读取或写入数据，而只是模拟触发相应的事件，从而让事件回调函数被调用。

例如，以下代码可以手动触发一个读取事件：

bufferevent_trigger(bufev, BEV_EVENT_READ, 0);

3.12 bufferevent_socket_connect()

bufferevent_socket_connect()函数的作用是创建一个新的sockaddr结构体，并将其与指定的socket文件描述符绑定，然后通过该文件描述符连接到指定的服务器。

该函数的原型如下：

int
bufferevent_socket_connect(struct bufferevent *bev,const struct sockaddr *sa, int socklen)
参数说明：bev：bufferevent缓冲事件对象的指针。sa：表示要连接的服务器地址的sockaddr结构体指针。socklen：表示服务器地址结构体的长度。

3.13 evutil_inet_pton()

evutil_inet_pton()函数的作用是将一个IPv4或IPv6地址字符串转换为网络字节序格式的二进制地址。该函数们在网络编程中，方便地进行IP地址的转换和处理。

该函数的原型如下：

int evutil_inet_pton(int af, const char *src, void *dst);参数说明：af：表示要转换的地址族类型，可以是AF_INET（IPv4）或AF_INET6（IPv6）。src：表示要转换的地址字符串。dst：指向存储转换结果的缓冲区指针。返回值：如果转换成功，返回1；如果出现错误，返回0。

evutil_inet_pton()函数会自动根据af参数选择相应的地址族类型，并将转换结果存储到dst指向的缓冲区中去。如果转换失败，返回0，并置errno为EINVAL等错误码。

例如，以下代码可以将一个IPv4地址字符串转换为网络字节序格式的二进制地址：

struct sockaddr_in sin;
memset(&sin, 0, sizeof(sin));if (evutil_inet_pton(AF_INET, "127.0.0.1", &sin.sin_addr) <= 0) {/* Handle error */
}

4 bufferevent服务器接例子

#include <iostream>
#include <event2/event.h>
#include <event2/listener.h>
#include <event2/bufferevent.h>
#include <string.h>#ifndef _WIN32
#include <signal.h>
#else
#endif#include <errno.h>
#include <string.h>using namespace std;//错误，超时 （连接断开会进入）
void event_cb(bufferevent *be,short events,void *arg)
{cout<<"[E]"<<flush;	//读取超时时间发生后，数据读取停止if(events & BEV_EVENT_TIMEOUT && events & BEV_EVENT_READING){cout<<"BEV_EVENT_READING BEV_EVENT_TIMEOUT"<<endl;//bufferevent_enable(be,EV_READ);bufferevent_free(be);}else if(events & BEV_EVENT_ERROR){bufferevent_free(be);}else{cout<<"OTHERS"<<endl;}
}
void write_cb(bufferevent *be,void *arg)
{cout<<"[W]"<<flush;	
}
void read_cb(bufferevent *be,void *arg)
{cout<<"[R]"<<flush;char data[1024] = {0};//读取输入缓冲数据int len = bufferevent_read(be,data,sizeof(data)-1);cout<<"["<<data<<"]"<<endl;if(len<=0)return;if(strstr(data,"quit") !=NULL){cout<<"quit";//退出并关闭socket BEV_OPT_CLOSE_ON_FREEbufferevent_free(be);}//发送数据 写入到输出缓冲bufferevent_write(be,"OK",3);
}void listen_cb(evconnlistener *ev,evutil_socket_t s,sockaddr*sin,int slen,void *arg)
{cout<<"listen_cb"<<endl;event_base *base = (event_base *)arg;//创建bufferevent上下文 BEV_OPT_CLOSE_ON_FREE清理bufferevent时关闭socketbufferevent *bev = bufferevent_socket_new(base,	//指向event_base事件处理器对象的指针。s,	//表示要关联到bufferevent的已经打开的socket文件描述符。BEV_OPT_CLOSE_ON_FREE);	//表示缓冲事件的配置选项。//添加监控事件bufferevent_enable(bev,EV_READ|EV_WRITE);bufferevent_setwatermark(bev,EV_READ,5,	//低水位 0就是无限制 默认是00	//高水位 0就是无限制 默认是0);bufferevent_setwatermark(bev,EV_WRITE,5,	//低水位 0就是无限制 默认是0 缓冲数据低于5 写入回调被调用0	//高水位无效);//超时时间的设置timeval t1 = {6,0};bufferevent_set_timeouts(bev,&t1,0);//设置回调函数bufferevent_setcb(bev,read_cb,write_cb,event_cb,base);
}int main(int argc,char *argv[])
{
#ifdef _WIN32 //初始化socket库WSADATA wsa;WSAStartup(MAKEWORD(2,2),&wsa);
#else//忽略管道信号，发送数据给已关闭的socketif (signal(SIGPIPE, SIG_IGN) == SIG_ERR)return 1;
#endifevent_base *base = event_base_new();//创建网络服务器//设定监听的端口和地址sockaddr_in sin;memset(&sin,0,sizeof(sin));sin.sin_family = AF_INET;sin.sin_port = htons(5001);evconnlistener *ev = evconnlistener_new_bind(base,	//事件处理器所使用的 event_base 对象listen_cb,		//新连接到来时调用的回调函数；base,			//回调函数的参数argLEV_OPT_REUSEABLE | LEV_OPT_CLOSE_ON_FREE, //传递给回调函数的参数指针；10,				//侦听套接字的挂起连接队列的最大长度(sockaddr*)&sin,	//结构体指针，表示要绑定的地址和端口号；sizeof(sin)	//地址结构体的长度。);//进入事件主循环event_base_dispatch(base);evconnlistener_free(ev);event_base_free(base);return 0;
}

5 bufferevent客户端接例子

#include <iostream>
#include <event2/event.h>
#include <event2/listener.h>
#include <event2/bufferevent.h>
#include <string.h>
#ifndef _WIN32
#include <signal.h>
#else
#endif
#include <string>
using namespace std;static int sendCount = 0;//错误，超时 （连接断开会进入）
void client_event_cb(bufferevent *be,short events,void *arg)
{cout<<"[client_E]"<<flush;	//读取超时时间发生后，数据读取停止if(events & BEV_EVENT_TIMEOUT && events & BEV_EVENT_READING){cout<<"BEV_EVENT_READING BEV_EVENT_TIMEOUT"<<endl;//bufferevent_enable(be,EV_READ);bufferevent_free(be);return;}else if(events & BEV_EVENT_ERROR){bufferevent_free(be);return;}//服务端的关闭事件if(events & BEV_EVENT_EOF){cout<<"BEV_EVENT_EOF"<<endl;bufferevent_free(be);}if(events & BEV_EVENT_CONNECTED ){cout<<"BEV_EVENT_CONNECTED"<<endl;//触发writebufferevent_trigger(be,EV_WRITE,0);}}
void client_write_cb(bufferevent *be,void *arg)
{cout<<"[client_W]"<<flush;FILE *fp = (FILE *)arg;char data[4096] = {0};int len = fread(data,1,sizeof(data)-1,fp);if(len<=0){//读到结尾或者文件出错fclose(fp);//立刻清理，可能会造成缓冲数据没有发送结束//bufferevent_free(be);bufferevent_disable(be,EV_WRITE);return;}sendCount += len;//写入bufferbufferevent_write(be,data,len);}void client_read_cb(bufferevent *be,void *arg)
{cout<<"[client_R]"<<flush;
}
int main(int argc,char *argv[])
{#ifdef _WIN32 //初始化socket库WSADATA wsa;WSAStartup(MAKEWORD(2,2),&wsa);
#else//忽略管道信号，发送数据给已关闭的socketif (signal(SIGPIPE, SIG_IGN) == SIG_ERR)return 1;
#endifevent_base *base = event_base_new();//创建网络服务器{//调用客户端代码//-1内部创建socket bufferevent *bev = bufferevent_socket_new(base,-1,BEV_OPT_CLOSE_ON_FREE);sockaddr_in sin;memset(&sin,0,sizeof(sin));sin.sin_family = AF_INET;sin.sin_port = htons(5001);evutil_inet_pton(AF_INET,"127.0.0.1",&sin.sin_addr.s_addr);FILE *fp = fopen("test_buffer_client.cpp","rb");//设置回调函数bufferevent_setcb(bev,client_read_cb,client_write_cb,client_event_cb,fp);bufferevent_enable(bev,EV_READ|EV_WRITE);int re = bufferevent_socket_connect(bev,(sockaddr*)&sin,sizeof(sin));if(re == 0){cout<<"connected"<<endl;}
}//进入事件主循环event_base_dispatch(base);event_base_free(base);return 0;
}