libevent的简单应用

柔光的暖阳◎ 2022-06-17 14:27 190阅读 0赞

Libevent的应用主要围绕几大事件：超时事件、信号事件、读/写事件。

下面就一一简单介绍一下它们的使用。

## 超时事件 ##

示例：

/*
     * Compile with:
     * gcc  time-test time-test.c -o time-test time-test  -I/usr/local/include  -L/usr/local/lib  -levent
     */
    /*
     * XXX This sample code was once meant to show how to use the basic Libevent
     * interfaces, but it never worked on non-Unix platforms, and some of the
     * interfaces have changed since it was first written.  It should probably
     * be removed or replaced with something better.
     *
     * Compile with:
     * gcc -I/usr/local/include -o time-test time-test.c -L/usr/local/lib -levent
     */
    
    #include <sys/types.h>
    #include <sys/stat.h>
    #include <time.h>
    #ifdef _EVENT_HAVE_SYS_TIME_H
    #include <sys/time.h>
    #endif
    #include <stdio.h>
    #include <event2/event.h>
    #include <event2/event_struct.h>
    #include <event2/util.h>
    
    struct timeval lasttime;
    int event_is_persistent;
    
    static void timeout_cb(evutil_socket_t fd, short event, void *arg)
    {
    	struct timeval newtime, difference;
    	struct event *timeout = arg;
    	double elapsed;
    
    	evutil_gettimeofday(&newtime, NULL);
    	evutil_timersub(&newtime, &lasttime, &difference);
    	elapsed = difference.tv_sec +
    	    (difference.tv_usec / 1.0e6);
    
    	printf("timeout_cb called at %d': %.3f seconds elapsed.\n",
    	    (int)newtime.tv_sec, elapsed);
    	lasttime = newtime;
    //  /* 启动此测试程序时,不加-p参数,使用以下代码也可实现相同功能 */
    //	struct timeval tv;
    //	evutil_timerclear(&tv);
    //	tv.tv_sec = 1;
    //	event_add(timeout, &tv);	//再次添加定时事件
    }
    
    int main(int argc, char **argv)
    {
    	struct event timeout;		//创建事件
    	struct timeval tv;
    	struct event_base *base;	//创建事件"总管"的指针
    	int flags;					//事件标志,超时事件可不设EV_TIMEOUT,因为在添加事件时可设置
    
    	if (argc == 2 && !strcmp(argv[1], "-p")) {
    		event_is_persistent = 1;
    		flags = EV_PERSIST;		//使得事件具有持久性(否则事件只会调度一次)
    	} else {
    		event_is_persistent = 0;
    		flags = 0;
    	}
    	/* Initalize the event library */
    	base = event_base_new();			//创建事件"总管"
    	/* Initalize one event */
    	event_assign(&timeout, base, -1, flags, timeout_cb, (void*) &timeout);
    	evutil_timerclear(&tv);
    	tv.tv_sec = 1;
    	event_add(&timeout, &tv);		//添加事件,同时设置超时时间
    	evutil_gettimeofday(&lasttime, NULL);
    	event_base_dispatch(base);		//循环监视事件,事件标志的条件发生,就调用回调函数
    	return (0);
    }

/\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*/

启动进程：

\[lgh@localhost test\]$ ./libe\_timer\_test -p

结果：

timeout\_cb called at 1325693811': 1.000 seconds elapsed.

timeout\_cb called at 1325693812': 1.000 seconds elapsed.

timeout\_cb called at 1325693813': 1.001 seconds elapsed.

timeout\_cb called at 1325693814': 1.000 seconds elapsed.

timeout\_cb called at 1325693815': 1.000 seconds elapsed.

以一行/秒的速率打印以上信息，也就是每秒打印一行。因为超时为一秒。

==============================

启动进程：

\[lgh@localhost test\]$ ./libe\_timer\_test

结果：

timeout\_cb called at 1325693516: 1.000 seconds elapsed.

没有-p参数,事件只调度一次。如果不加EV\_PERSIST标志也想实现事件的持续性，还有一种办法，就是在回调函数的后面再添加该事件，即上面回调函数的批量注释代码。

## 信号事件 ##

示例：

/*
     * Compile with:
     *  gcc libe_signal_test.c -o libe_signal_test -I/usr/local/include -L/usr/local/lib -levent
     */
    #include <signal.h>
    #include <stdio.h>
    #include <event.h>
    #include <event2/event.h>
    #ifdef _EVENT___func__
    #define __func__ _EVENT___func__
    #endif
    
    //int called = 0;
    
    static void signal_cb(evutil_socket_t fd, short event, void *arg)
    {
    	struct event *signal = arg;
    	sleep(10);
    	printf("%s: got signal %d\n", __func__, EVENT_SIGNAL(signal));
    
    //	if (called >= 2)
    //		event_del(signal);
    //	called++;
    }
    
    int main(int argc, char **argv)
    {
    	struct event signal_usr;
    	struct event_base* base;
    
    	/* Initalize the event library */
    	base = event_base_new();
    	/* Initalize one event */
    	event_assign(&signal_usr, base, SIGUSR1, EV_SIGNAL|EV_PERSIST, signal_cb,
    	    &signal_usr);
    	event_add(&signal_usr, NULL);
    	event_base_dispatch(base);
    	event_base_free(base);
    	printf("end of main!\n");
    	return (0);
    }

启动进程：

\[lgh@localhost test\]$ ./libe\_signal\_test &

\[1\] 2998

用kill -10 2998命令给进程发送信号SIGUSR1，进程的的执行结果如下：

\[lgh@localhost test\]$ kill -10 2998

signal\_cb: got signal 10

\[lgh@localhost test\]$ kill -10 2998

signal\_cb: got signal 10

\[lgh@localhost test\]$ signal\_cb: got signal 10

给进程发送了3次SIGUSR1信号，信号回调函数执行了三次（其中最后一行隔了几秒才打印出来）。这说明libevent对linux中的不可靠信号也是支持排队的。

## 读/写事件 ##

文件描述符是否可读/写，这个不太好模拟（可能用文件的读/写锁可以实现模拟，鄙人目前还没有尝试过，有试过的朋友可以指点一下），有一种方法就是用socket连接来模拟，先建立一个服务端和客户端，当服务端的监听端口可读时说明有一个新的连接请求。

示例：

服务端——proc\_server.c

#include <sys/socket.h>
    #include <sys/types.h>
    #include <netinet/in.h>
    #include <stdio.h>
    #include <event.h>
    #include <event2/event.h>
    #include<errno.h>
    
    #define PORT        	6666			//----------需改灵活点
    #define BACKLOG     	10				//好像没起到作用，我设置为1，在同一机子下开两个连接，没弹出警告信息
    #define EV_BUFSIZE_T   	sizeof(pid_t)	//服务端与客户端传递数据的buffer大小
    
    /* 管理每一个连接的读写事件和数据 */
    typedef struct sock_event {
        struct event *read_ev;		//读事件
        struct event *write_ev;		//写事件
        pid_t *buffer;				//buffer仅存进程pid
    	int	gapbeats;				//定时间隔
    	int	maxbeats;				//在客户端不工作或退出的情况下，服务端的最大检测次数
    	int	pastbeats;				//没有收到数据的情况下，当前的心跳检测数
    }sock_ev;
    
    struct event_base* base;		//管理所有连接事件
    
    /* 释放堆分配的sock_ev结构体 */
    void release_sock_ev(sock_ev *ev)
    {
        event_del(ev->read_ev);
        free(ev->read_ev);
        event_del(ev->write_ev);	
        free(ev->write_ev);
    	free(ev->buffer);
        free(ev);
    }
    
    /* 功能:创建一个sock_ev结构体,并且将它初始化.
     * 参数:gapbeats,服务端两个检测心跳的间隔时间(单位:秒);
     *		maxbeats,没有收到客户端数据的最大检测次数.
     * 返回:sock_ev结构体指针.
     */
    sock_ev* create_sock_ev(int gapbeats, int maxbeats)
    {
    	sock_ev * se = (sock_ev *)malloc(sizeof(sock_ev));	
    	if (!se)
    		return NULL;
    	memset(se, 0, sizeof(sock_ev));
    	se->read_ev = (struct event*)malloc(sizeof(struct event));
        se->write_ev = (struct event*)malloc(sizeof(struct event));
        se->buffer = (pid_t *)malloc(EV_BUFSIZE_T);
        if (!se->read_ev || !se->write_ev || !se->buffer)
        	return NULL;
        memset(se->read_ev, 0, sizeof(struct event));
        memset(se->write_ev, 0, sizeof(struct event));
        memset(se->buffer, 0, EV_BUFSIZE_T);
        se->gapbeats = gapbeats;
        se->maxbeats = maxbeats;
        se->pastbeats = 0;
    
    	return se;	
    }
    
    /* 功能:写事件回调函数
     * 参数:libevent回调函数的三个典型参数
     * 		sock,文件描述符;event,事件类型(EV_WRITE);arg,传给函数的数据指针(buffer)
     * 返回: void (libevent回调函数的返回为void)
    */
    void socket_write(int sock, short event, void* arg)
    {
        pid_t* buffer;
        if (!arg)
        	return;
        	
        buffer = (pid_t*)arg;
        if (send(sock, buffer, sizeof(*buffer), 0) < 0) {
        	printf("server send msg error: errno %d--%s\n", errno, strerror(errno));    
    		return; 
        }
    	memset(buffer, 0, sizeof(*buffer));
    }
    
    /* 功能:读事件回调函数
     * 参数:libevent回调函数的三个典型参数
     * 		sock,文件描述符;event,事件类型(EV_READ);arg,传给函数的数据指针(sock_ev)
     * 返回: void.
    */
    void socket_read(int sock, short event, void* arg)
    {
        int size;
        sock_ev* sockev = (sock_ev*)arg;
        if(!sockev)
        	return;
        	
        memset(sockev->buffer, 0, EV_BUFSIZE_T);
        size = recv(sock, sockev->buffer, EV_BUFSIZE_T, 0);
        if (size <= 0) {			//接收数据失败
        	sockev->pastbeats++;				//全局变量
        	printf("pastbeats:\t%d\n", sockev->pastbeats);		//--debug
        	if (sockev->pastbeats >= sockev->maxbeats) {		
        		printf("---client error or exit:please restart\n");	//--debug
    	        release_sock_ev(sockev);
    	        close(sock);
            }
            return;
        }
        sockev->pastbeats = 0;
        printf("pastbeats:\t%d\n", sockev->pastbeats);			//--debug
        printf("receive data:\t%d	size:\t%d\n", *sockev->buffer, size);
        event_add(sockev->write_ev, NULL);			//添加端口写事件，将数据返回给客户端
    }
    
    /* 功能:接受新连接请求
     * 参数:libevent回调函数的三个典型参数
     * 		sock,文件描述符;event,事件类型(EV_READ,监听端口可读,表示有新连接请求);
     		arg,目前为空指针.
     * 返回: void.
    */
    void connect_accept(int sock, short event, void* arg)
    {
        struct sockaddr_in cli_addr;
        int connetfd, sin_size;
        struct timeval beat;				//定时读事件,来检测客户端发送了数据
        sock_ev* sockev;  					//为连接建立端口事件
        
        if ((sockev = create_sock_ev(1, 10)) == NULL)
        	return;
        sin_size = sizeof(struct sockaddr_in);    
        connetfd = accept(sock, (struct sockaddr*)&cli_addr, &sin_size);
        if (connetfd == -1) {
        	printf("server accept() error: errno %d--%s\n", errno, strerror(errno));    
    		return;
        }  
        event_assign(sockev->read_ev, base, connetfd, EV_PERSIST, socket_read, sockev);  
        //下面是老版接口	
    //    event_set(sockev->read_ev, connetfd, EV_PERSIST, socket_read, sockev);		//读事件  (若加上EV_READ|,则定时读会失效)
    //    event_base_set(base, sockev->read_ev);  
        evutil_timerclear(&beat);
        beat.tv_sec = sockev->gapbeats;			 //定期检查端口是否可读,来判断客户端是否存在
        event_add(sockev->read_ev, &beat);
        
        event_assign(sockev->write_ev, base, connetfd, EV_WRITE, socket_write, sockev->buffer);  																			//写事件
    //    event_set(sockev->write_ev, connetfd, EV_WRITE, socket_write, sockev->buffer);	
    //    event_base_set(base, sockev->write_ev);										  
    }
    
    int main(int argc, char* argv[])
    {
        struct sockaddr_in server_addr;
        int sock;
    //    struct event listen_ev;		//创建连接请求监听事件
    	struct event *listen_ev;
    	
        sock = socket(AF_INET, SOCK_STREAM, 0);
        setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, (int *)1, sizeof(int));
        memset(&server_addr, 0, sizeof(server_addr));
        server_addr.sin_family = AF_INET;
        server_addr.sin_port = htons(PORT);
        server_addr.sin_addr.s_addr = INADDR_ANY;
        if (bind(sock, (struct sockaddr*)&server_addr, sizeof(struct sockaddr)) == -1) {
        	printf("bind socket error: errno %d--%s\n", errno, strerror(errno));    
    		exit(0); 
        }
        if (listen(sock, BACKLOG) == -1) {
        	printf("listen socket error: errno %d--%s\n", errno, strerror(errno));    
    		exit(0);
        }
    
        base = event_base_new();	//base,全局变量.
        listen_ev = event_new(base, sock, EV_READ|EV_PERSIST, connect_accept, NULL);	//创建evnet对象并初始化
    	if (!listen_ev) {
    		printf("event_new() fail\n");
    		exit(0);
    	}
    //    event_set(&listen_ev, sock, EV_READ|EV_PERSIST, connect_accept, NULL); 这是老接口
    //    event_base_set(base, &listen_ev);
        event_add(listen_ev, NULL);		//添加到监视事件集中，event就变成的未决状态
        event_base_dispatch(base);		//轮询监视所有事件
        
    	if(event_del(listen_ev) == 0) {	//从监视事件集中删除
    		event_free(listen_ev);		//删除事件,释放空间
    	}
    	event_base_free(base);			//删除base对象
        exit(0);
    }

客户端：proc\_client.c

#include<stdio.h>
    #include<stdlib.h>
    #include<string.h>
    #include<errno.h>
    #include<sys/types.h>
    #include<sys/socket.h>
    #include<netinet/in.h>
    #include <event.h>
    #include <event2/event.h>
    #include <event2/util.h>
    
    #define MAXLINE 1024
    
    static void heartbit_cb(evutil_socket_t fd, short event, void *arg);
    
    int main(int argc, char** argv)
    {
    	int sockfd, n, received; 
    	int len, bytes;   
    	char recvline[MAXLINE], sendline[MAXLINE]; 
    	pid_t tests = getpid(); 
    	pid_t *pids = &tests; 
    	pid_t testr = 0; 
    	pid_t *pidr = &testr; 
    	struct sockaddr_in    servaddr;  
    	struct timeval tv; 
    	struct event_base* base;
    	struct event *client_ev;
    	
    	if( argc != 2){    
    		printf("usage: ./client <ip address>\n");    
    		exit(0);    
    	}    
    
    	memset(sendline, 0, MAXLINE);
    	if( (sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0){    
    		printf("create socket error: %s(errno: %d)\n", strerror(errno),errno);    
    		exit(0);    
    	}    
    	memset(&servaddr, 0, sizeof(servaddr));    
    	servaddr.sin_family = AF_INET;    
    	servaddr.sin_port = htons(6666);   //把16位值从主机字节序转换成网络字节序 
    	if( inet_pton(AF_INET, argv[1], &servaddr.sin_addr) <= 0){   // [将“点分十进制”ip－> 网络字节序“整数”ip] 
    		printf("inet_pton error for %s\n",argv[1]);    
    		exit(0);    
    	}    
    	if( connect(sockfd, (struct sockaddr*)&servaddr, sizeof(servaddr)) < 0){    
    		printf("connect error: %s(errno: %d)\n",strerror(errno),errno);    
    		exit(0);    
    	}    
    	printf("send msg to server: \n");  
    	evutil_timerclear(&tv);
    	tv.tv_sec = 2;
    	
    	base = event_base_new();
    	client_ev = event_new(base, sockfd, EV_PERSIST, heartbit_cb, pids);
    	if (!client_ev) {
    		printf("event_new() fail\n");
    		exit(0);
    	}
    //	event_set(&client_ev, sockfd, EV_PERSIST, heartbit_cb, pids); //若加上EV_WRITE|,sockfd可写,则会一直写(与定时事件是关系或) EV_PERSIST
    //	event_base_set(base, &client_ev);
        event_add(client_ev, &tv);
        event_base_dispatch(base);
        
        if(event_del(client_ev) == 0) {
    		event_free(client_ev);
    	}
    	event_base_free(base);
    	close(sockfd);    
    	exit(0);
    }
    
    /* 功能: 向服务端发送心跳包
     * 参数: 
     *
     */
    static void heartbit_cb(evutil_socket_t fd, short event, void *arg)
    {
    	pid_t *pid = (pid_t *)arg; 
    	pid_t testr = 0; 
    	pid_t *pid_recv = &testr; 
    	int len;
    
    	len = sizeof(pid_t);
    	if( send(fd, pid, len, 0) != len) {    
    		printf("send msg error: %s(errno: %d)\n", strerror(errno), errno);    
    		exit(0);    
    	}
    	//接收从服务端的返回数据  	
    	fputs("echo from server:\n", stdout);
    	if(recv(fd, pid_recv, len, 0) < 0) {
    		printf("Failed to receive bytes from client\n");
    		exit(-1);
    	} 
    	printf("%d\n", *pid_recv);
    	fputs("\n", stdout); 	
    	
    }

结果：

\[lgh@localhost proc\]$ ./proc\_client 192.168.1.107  
send msg to server:  
echo from server:  
7482

echo from server:  
7482

\--------------------------------------------------------------------------------------------

\[lgh@localhost proc\]$ ./proc\_server  
pastbeats: 0  
receive data:7482       size:4  
pastbeats: 0  
receive data:7482       size:4  
pastbeats: 0  
receive data:7482       size:4  
pastbeats: 1  
pastbeats: 2  
pastbeats: 3  
pastbeats: 4  
pastbeats: 5  
pastbeats: 6  
pastbeats: 7  
pastbeats: 8  
pastbeats: 9  
pastbeats: 10  
\---pastbeats > maxbeats  
=============================================================

测试设置：客户端每2秒发一次心跳,服务端每1秒去查看端口是否可读.  
                    服务端若经过10次还没收到心跳,则认为客户端已退出.  
 测试过程：先让客户端发3次心跳,再终止掉客户端.---用时6秒。  
                    此时，服务端已经对端口发起了6次检测，有3次接收到了数据，有3次没有收到数据。  
                     当终止掉客户端后，服务端每次检测都会收不到数据，现象是：前3次是连续执行了超时回调函数socket\_read，这3次没有经过每隔1秒执行。  
                     后面再每秒检测7次，即每秒执行一次回调函数。总计10次没有收到来自客户端的数据，判断客户端已退出。  
   
  测试结果判断: libevent对定时事件支持排队,即有多少次定时,它就执行回调函数多少次.  
  编程建议：服务端的心跳频率要小于等于客户端的心跳频率.（小于，会有丢包现象，但我们的需求只是检测客户端是否存在）

小结：这个小例子用来做进程管理，客户端是不行的，因为这里的客户端也是libevent的超时事件，它在轮询超时事件的时候会一直占用进程的cpu，所以这样是不行的，所以客户端的定时发送心跳包应该改用信号做成一个小模块加入到客户端进程中。