libevent高并发网络编程 - 01_libevent事件Event处理_libevent socket 多连接

链接: C/C++Linux服务器开发/后台架构师【零声教育】-学习视频教程-腾讯课堂

1. libevent事件驱动和事件处理简介

libevent是一个事件驱动框架，可以用于处理网络通信等I/O操作其中，事件处理(Event)和缓冲区( bufferevent)处理是两个不同的组件。

• 事件处理（event）主要负责监听文件描述符上的事件，并在事件发生时调用相应的回调函数进行处理例如，一个TCP连接建立成功后，事件处理器可以监听该连接上是否有可读、可写或异常事件发生，并在事件发生时触发相应的回调函数进行数据处理。

• 缓冲区处理（bufferevent）则是对事件处理的补充，它提供了对I/O数据的缓冲和处理功能当事件处理器检测到某个文件描述符可读时，缓冲区处理器会从该文件描述符读取一定量的数据并将其缓存在内存中，然后通过回调函数对缓存数据进行处理。类似地，当事件处理器检测到某个文件描述符可写时，缓冲区处理器会将已缓存的数据写入文件描述符中。

因此，事件处理和缓冲区处理是两个不同的组件，但常常会结合使用，以实现高效的网络通信。

下面以libevent事件处理（event）学习，相关的API接口，编写linux信号事件示例 、定时器事件示例、网络服务器事件示例。

2. 事件状态分析

在libevent中，状态的切换是由不同的API调用触发的：

1. initialized 已初始化：事件初始化通常使用event_new()或evtimer_new()等函数创建一个新的事件对象，并设置其回调函数和关注的事件类型。
2. pending 待决：将事件添加到事件处理器中，可以使用event_add()或event_assign()函数，将事件添加到指定的事件处理器中等待处理。
3. active 激活：当事件处理器检测到相应的I/O事件时，会激活相关的事件并执行其回调函数事件处理器可使用event_base_loop()或event_base_dispatch()等函数循环等待I/O事件的发生。
4. persistent 持久的：如果需要将事件设置为永久性事件，可以在事件处理器中将其标记为永久性例如，在调用event_add()函数之前，可以使用event_set()函数设置EV_PERSIST标志来使事件变为永久性事件。

3. 事件Event常用API

3.1 event_base_new()

event_base_new()**用于创建一个新的事件处理器对象（event_base）**它没有任何参数，返回一个指向新创建的event_base对象的指针。

在使用libevent时，需要先创建一个事件处理器对象，然后将需要处理的事件添加到该事件处理器中。event_base_new()函数就是用来创建这个事件处理器对象的。例如：

struct event_base *base = event_base_new();
1

3.2 event_base_free()

event_base_free()用于释放事件处理器对象（event_base）所占用的资源它有一个参数，即需要释放的event_base对象的指针。

在使用libevent时，当不再需要某个事件处理器对象时，需要使用event_base_free()函数释放相应的资源。例如：

struct event_base *base = event_base_new();
// 使用base处理事件...
event_base_free(base);
1
2
3

3.3 event_new()

event_new()用于创建一个新的事件对象（struct event）：

struct event * 
event_new(struct event_base *base, evutil_socket_t fd, short events, void (*cb)(evutil_socket_t, short, void *), void *arg)
    
参数
	base：指向事件处理器对象（event_base）的指针，表示新创建的事件对象将与该事件处理器相关联。
	fd：表示该事件对象关注的文件描述符。
	events：表示该事件对象关注的事件类型，可选值包括EV_READ、EV_WRITE、EV_SIGNAL等。
	callback：表示该事件对象在触发时需要执行的回调函数。
    arg：回调函数的传入参数。
    
返回一个指向新创建的struct event对象的指针
1
2
3
4
5
6
7
8
9
10
11

events参数可以用来指定一个事件（struct event）关注的事件类型。可选的事件类型包括：

• EV_TIMEOUT：超时事件，当计时器到达指定时间时触发。
• EV_READ：读事件，当文件描述符上有可读数据时触发。
• EV_WRITE：写事件，当文件描述符可写入数据时触发。
• EV_SIGNAL：信号事件，当指定的信号被触发时触发。
• EV_PERSIST：持久事件，使得事件对象在处理完毕后不会自动删除，可以用于周期性地执行某个任务。

这些事件类型可以通过按位或运算组合使用，表示事件对象同时关注多个事件类型。例如，将一个事件对象设置为关注读、写和超时事件，可以使用以下代码：

short events = EV_READ | EV_WRITE | EV_TIMEOUT;
1

关注不同的事件类型可以实现各种不同的功能，例如：

• 读事件：可以用来处理网络通信中的数据接收。
• 写事件：可以用来处理网络通信中的数据发送。
• 超时事件：可以用来实现定时任务。
• 信号事件：可以用来捕获操作系统发送的信号并进行处理。
• 永久事件：可以用来周期性地执行某些任务，例如定时检查某个状态或发送心跳包。

事件对象的回调函数（callback）

void (*cb)(evutil_socket_t, short, void *)

参数：
	fd：触发事件的文件描述符。
	events：触发的事件类型，可以是EV_READ、EV_WRITE、EV_TIMEOUT等。
	arg：回调函数所关联的参数，在创建事件对象时通过arg参数传递给事件对象。
1
2
3
4
5
6

例如，在以下代码中，my_callback()函数是一个事件对象的回调函数，其参数为fd、events和arg：

void my_callback(evutil_socket_t fd, short events, void *arg) {
    // 在此处编写事件回调函数的代码...
}

struct event_base *base = event_base_new();
int fd = ... ; // 一个文件描述符
short events = EV_READ | EV_PERSIST;
struct event *ev = event_new(base, fd, events, my_callback, NULL);
1
2
3
4
5
6
7
8

此代码会创建一个新的事件对象，并将其与上面创建的事件处理器对象相关联。事件对象关注的文件描述符为fd，关注的事件类型为EV_READ并且设置为永久事件（即EV_PERSIST标志），当该文件描述符上有读事件发生时，就会调用my_callback()函数进行处理。

3.4 event_add()

event_add()用于将事件对象（struct event）添加到指定的事件处理器（event_base）中等待触发。

int event_add(struct event *ev, const struct timeval *tv)
    
	ev：需要添加到事件处理器中的事件对象的指针。
	timeout：表示等待事件触发的超时时间，可以为NULL表示不进行超时等待。当timeout参数为0时，表示事件处理器在等待事件触发时不会进行任何超时等待。程序会一直阻塞在event_base_dispatch()函数调用中，直到有事件触发或者调用了event_base_loopbreak()函数或event_base_loopexit()函数来终止事件循环。
1
2
3
4

例如，以下代码会将一个事件对象添加到指定的事件处理器中等待触发：

struct event_base *base = event_base_new();
int fd = ... ; // 一个文件描述符
short events = EV_READ | EV_PERSIST;
struct event *ev = event_new(base, fd, events, my_callback, NULL);

// 将事件对象添加到事件处理器中等待触发
struct timeval timeout = {5, 0}; // 设定5秒超时等待
event_add(ev, &timeout);

// 使用事件处理器等待事件触发
event_base_dispatch(base);
1
2
3
4
5
6
7
8
9
10
11

先创建了一个新的事件对象ev，并将其添加到指定的事件处理器base中，并指定了5秒的超时等待时间。在调用event_base_dispatch()函数时，程序将进入阻塞状态，并等待事件处理器中的事件触发。如果在等待期间没有任何事件触发，程序将在超时后自动返回。

3.5 event_del()

event_del()用于将事件对象（struct event）从事件处理器（event_base）中删除，使其不再等待触发。

int event_del(struct event *ev)
    
ev：需要从事件处理器中删除的事件对象的指针。
1
2
3

3.6 event_free()

event_free()用于释放事件对象（struct event）所占用的资源。它有一个参数，即需要释放的struct event对象的指针。

在使用libevent时，当不再需要某个事件对象时，需要使用event_free()函数释放相应的资源。例如：

struct event_base *base = event_base_new();
int fd = ... ; // 一个文件描述符
short events = EV_READ | EV_PERSIST;
struct event *ev = event_new(base, fd, events, my_callback, NULL);

// 使用ev处理事件...
event_free(ev);
1
2
3
4
5
6
7

3.7 event_base_dispatch()

event_base_dispatch()是libevent库中的一个函数，用于启动事件处理器（event_base）并进入事件循环，等待事件对象触发。

int event_base_dispatch(struct event_base *event_base)
{
	return (event_base_loop(event_base, 0));
}
1
2
3
4

3.8 event_base_loopbreak()

event_base_loopbreak()用于终止事件处理器（event_base）的事件循环。

int event_base_loopbreak(struct event_base *event_base)
    
base：需要终止事件循环的事件处理器对象的指针。
1
2
3

3.9 evsignal_new()

evsignal_new()用于创建一个新的信号事件对象（struct event）。

#define evsignal_new(b, x, cb, arg)				\
	event_new((b), (x), EV_SIGNAL|EV_PERSIST, (cb), (arg))

隐藏的事件 EV_SIGNAL|EV_PERSIST（信号事件|持久事件）

参数
    b：指向事件处理器对象（event_base）的指针，表示新创建的事件对象将与该事件处理器相关联。
    x：表示需要关注的信号编号。
    cb：表示该事件对象在触发时需要执行的回调函数。
    arg：回调函数的传入参数。
    
返回一个指向新创建的`struct event`对象的指针。
    
每个信号都有一个唯一的编号（signal number），例如：
    SIGINT：中断信号，通常由CTRL-C键盘按键触发。
    SIGTERM：终止信号，通常用于请求终止进程。
    SIGKILL：强制终止信号，无法被阻塞、处理或忽略。
    SIGPIPE：管道破裂信号，通常由写入已关闭的管道或socket时触发。
    SIGALRM：定时器到期信号，通常用于实现计时器功能。
    SIGHUP：重启信号，当控制终端或SSH连接关闭时触发。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

void my_signal_callback(evutil_socket_t fd, short events, void *arg) {
    // 在此处编写信号事件回调函数的代码...
}

struct event_base *base = event_base_new();
int signum = SIGINT;
struct event *ev = evsignal_new(base, signum, my_signal_callback, NULL);
1
2
3
4
5
6
7

此代码会创建一个新的信号事件对象，并将其与上面创建的事件处理器对象相关联。信号事件对象关注的信号编号为SIGINT，当该信号被触发时，就会调用my_signal_callback()函数进行处理。

3.10 event_self_cbarg()

event_self_cbarg()，**获取当前正在执行回调函数的事件对象。它返回一个void指针，指向当前事件对象的起始地址。**如果在非回调函数中调用此函数，则返回NULL。

void * event_self_cbarg(void)
{
	return &event_self_cbarg_ptr_;
}
1
2
3
4

3.11 evsignal_pending()

evsignal_pending(), 用于检测指定信号是否处于非待决状态（non-pending state）的函数。如果指定信号处于非待决状态，则返回1；否则返回0。

#define evsignal_pending(ev, tv)	event_pending((ev), EV_SIGNAL, (tv))
1

3.12 evutil_make_socket_nonblocking()

evutil_make_socket_nonblocking()用于将 socket 设置为非阻塞模式。该函数的声明如下：

int evutil_make_socket_nonblocking(evutil_socket_t fd)
1

该函数的实现原理是通过调用 fcntl() 或者 ioctlsocket() 来将 socket 设置为非阻塞模式。如果操作成功，则返回 0；否则返回一个负数错误码。

3.13 evutil_make_listen_socket_reuseable()

evutil_make_listen_socket_reuseable() 用于设置 socket 地址重用选项。该函数的声明如下：

int evutil_make_listen_socket_reuseable(evutil_socket_t sock)
1

在 TCP 协议中，默认情况下，在端口被占用后，如果再次绑定该端口会返回错误。设置地址重用选项可以允许多个进程或线程同时监听同一端口号。

该函数的实现原理是通过调用 setsockopt() 函数来设置 SO_REUSEADDR 选项。如果操作成功，则返回 0；否则返回一个负数错误码。

4. linux信号事件示例

#include <iostream>
#include <event2/event.h>
#include <signal.h>
using namespace std;

//sock 文件描述符，which 事件类型 arg传递的参数
static void Ctrl_C(evutil_socket_t sock, short which, void *arg)
{
    cout<<"Ctrl_C"<<endl;
}

static void Kill(evutil_socket_t sock, short which, void *arg)
{
    cout<<"Kill"<<endl;
    event *ev = (event*)arg;

    //如果处于非待决
    if(!evsignal_pending(ev, NULL))
    {
        event_del(ev);
		event_add(ev,NULL);
    }
}

int main(int argc,char *argv[])
{
    //创建一个新的事件处理器对象
    event_base *base = event_base_new();

    //添加ctrl +C 信号事件，处于no pending
    //evsignal_new 隐藏的事件 EV_SIGNAL|EV_PERSIST（信号事件|持久事件）
    event* csig = evsignal_new(base, SIGINT, Ctrl_C, base);
    if(!csig)
    {
        cerr<<"SIGINT evsignal_new failed!"<<endl;
        return -1;
    }
    //添加事件到pending
    if(event_add(csig,0) != 0)
    {
        cerr<<"SIGINT event_add failed!"<<endl;
        return -1;
    }


    //添加kill信号
    //非持久事件，只进入一次 event_self_cbarg()传递当前的event
    event *ksig = event_new(base, SIGTERM, EV_SIGNAL, Kill, event_self_cbarg());
    if(!ksig)
    {
        cerr<<"SIGTERM evsignal_new failed!"<<endl;
        return -1;
    }
    //添加事件到pending
    if(event_add(ksig,0) != 0)
    {
        cerr<<"SIGTERM event_add failed!"<<endl;
        return -1;
    }

    //进入事件主循环
    event_base_dispatch(base);

    //释放事件对象
    event_free(csig);

    //释放事件处理器对象
    event_base_free(base);

    return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71

5. 定时器事件示例

#include <iostream>
#include <event2/event.h>
#ifndef _WIN32
#include <signal.h>
#else
#endif

using namespace std;

static timeval t1 = {1,0};

void timer1(int sock,short which,void *arg)
{
	cout<<"[timer1]"<<flush;
	event *ev = (event *)arg;
	//no pending
	if(!evtimer_pending(ev,&t1))
	{
		evtimer_del(ev);
		evtimer_add(ev,&t1);
	}
}

void timer2(int sock,short which,void *arg)
{
	cout<<"[timer2]"<<flush;
}

int main(int argc,char *argv[])
{
#ifdef _WIN32 
	//初始化socket库
	WSADATA wsa;
	WSAStartup(MAKEWORD(2,2),&wsa);
#else
	//忽略管道信号，发送数据给已关闭的socket
	if (signal(SIGPIPE, SIG_IGN) == SIG_ERR)
		return 1;
#endif

	event_base *base = event_base_new();
	
	//定时器
	cout<<"test timer"<<endl;
	
	//event_new
	/*
	#define evtimer_new(b, cb, arg)		event_new((b), -1, 0, (cb), (arg))
	#define evtimer_add(ev, tv)		event_add((ev), (tv))
	#define evtimer_del(ev)			event_del(ev)
	*/
	//定时器 非持久事件 
    event *ev1 = evtimer_new(base,timer1,event_self_cbarg());
	if(!ev1)
	{
		cout<<"evtimer_new timer1 failed!"<<endl;
		return -1;
	}
	evtimer_add(ev1,&t1); //插入性能 O(logn)

    static timeval t2;
	t2.tv_sec = 2;
	t2.tv_usec = 200000; //微秒
	event *ev2 = event_new(base,-1,EV_PERSIST,timer2,0);
	event_add(ev2,&t2);



	//进入事件主循环
	event_base_dispatch(base);
	event_base_free(base);
		
    return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

6. 网络服务器事件示例

#include <iostream>
#include <event2/event.h>

#ifndef _WIN32
#include <signal.h>
#else
#endif

#include <errno.h>
#include <string.h>

using namespace std;
#define SPORT 5001 

//正常断开连接也会进入，超时会进入
void client_cb(evutil_socket_t s,short w, void *arg)
{
	//水平触发LT 只有有数据没有处理，会一直进入
	//边缘触发ET 有数据时只进入一次
	//cout<<"."<<flush;return;
	
	event *ev = (event *)arg;
	//判断超时
	if(w&EV_TIMEOUT)
	{
		cout<<"timeout"<<endl;
		event_free(ev);
		evutil_closesocket(s);
		return;
	}
	
	//char buf[1024] = {0};
	char buf[1024] = {0};
	int len = recv(s,buf,sizeof(buf)-1,0);
	if(len>0)
	{
		cout<<buf;
		send(s,"ok",2,0);
	}
	else
	{
		//需要清理event
		cout<<"event_free"<<flush;
		event_free(ev);
		evutil_closesocket(s);
		
	}
	
}

void listen_cb(evutil_socket_t s,short w, void *arg)
{
	cout<<"listen_cb"<<endl;
	sockaddr_in sin;
	socklen_t size = sizeof(sin);
	//读取连接信息
	evutil_socket_t client = accept(s,(sockaddr*)&sin,&size);
	char ip[16] = {0};
	evutil_inet_ntop(AF_INET,&sin.sin_addr,ip,sizeof(ip)-1);
	cout<<"client ip is "<<ip<<endl;
	
	//客户端数据读取事件
	event_base *base = (event_base *)arg;
	event *ev = event_new(base,client,EV_READ|EV_PERSIST,client_cb,event_self_cbarg());
	//边缘触发
	//event *ev = event_new(base,client,EV_READ|EV_PERSIST|EV_ET,client_cb,event_self_cbarg());
	timeval t = {10,0};
	event_add(ev,&t);
}

int main(int argc,char *argv[])
{
#ifdef _WIN32 
	//初始化socket库
	WSADATA wsa;
	WSAStartup(MAKEWORD(2,2),&wsa);
#else
	//忽略管道信号，发送数据给已关闭的socket
	if (signal(SIGPIPE, SIG_IGN) == SIG_ERR)
		return 1;
#endif

	event_base *base = event_base_new();
	
	//event 服务器
	cout<<"test event server"<<endl;
	
	//创建socket
	evutil_socket_t sock = socket(AF_INET,SOCK_STREAM,0);
	if(sock<=0)
	{
		cerr<<"socket error："<<strerror(errno)<<endl;
		return -1;
	}

	//设置地址复用和非阻塞
	evutil_make_socket_nonblocking(sock);
	evutil_make_listen_socket_reuseable(sock);

	//绑定端口和地址
	sockaddr_in sin;
	memset(&sin,0,sizeof(sin));
	sin.sin_family = AF_INET;
	sin.sin_port = htons(SPORT);
	int re = ::bind(sock,(sockaddr*)&sin,sizeof(sin));  //使用全局bind,不使用std的bind
	if(re != 0)
	{
		cerr<<"bind error:"<<strerror(errno)<<endl;
		return -1;
	}
	//开始监听
	listen(sock,10);

	//开始接受连接事件 默认水平触发
	event *ev = event_new(base,sock,EV_READ|EV_PERSIST,listen_cb,base);
	event_add(ev,0);
	
	//进入事件主循环
	event_base_dispatch(base);
	evutil_closesocket(sock);
	event_base_free(base);
		
    return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124