记录 Libevent的常用功能示例

介绍

Libevent是开源社区一款高性能的I/O框架库，是reactor模式的优秀体现。网上相关资料很多，这篇博文主要以尽量简练的代码实现TcpServer服务器功能，代码覆盖大部分的常用函数接口。通过代码能对Libevent的整体框架，函数使用有直观的理解，配置好Libevent库的前提下，下方代码可直接复制运行。

代码

#include <string.h>
#include <errno.h>
#include <stdio.h>
#include <thread>
#include <unordered_map>
#include <utility>
#include <signal.h>
#include <event2/bufferevent.h>
#include <event2/buffer.h>
#include <event2/listener.h>
#include <event2/util.h>
#include <event2/event.h>

//读取客户端
void do_read(evutil_socket_t fd, short event, void *arg) 
{
	printf("do_read! fd=%d threadid=%d ", fd, std::this_thread::get_id());

	//继续等待接收数据  
	char buf[1024] = {0};    
	int len = recv(fd, buf, 1024, 0);
	if (len > 0) 
	{
		printf("data=%s\n", buf);
		send(fd, buf, len, 0); 
	}
	else if (len == 0)
	{
		printf("client close\n");
		struct event *ev = (struct event *)arg;
		event_free(ev);
		closesocket(fd);
	}
}

void read_cb(struct bufferevent *bev, void *arg)
{
	evutil_socket_t fd = bufferevent_getfd(bev);
	evbuffer *evBuf = bufferevent_get_input(bev);
	int size = evbuffer_get_length(evBuf);

	char data[128] = {0};
	int len = evbuffer_remove(evBuf, data, size);
	printf("read_cb fd=%d threadid=%d size=%d data=%s\n", fd, std::this_thread::get_id(), size, data);

	send(fd, data, len, 0);
}

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 = %d, ", 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("Other error\n");

	bufferevent_free(bev);
}

//回调函数，用于监听连接进来的客户端socket
void do_accept(evutil_socket_t fd, short event, void *arg) 
{   
	struct sockaddr_in client_addr; //客户端网络地址结构体   
	int in_size = sizeof(struct sockaddr_in);

	//客户端socket  
	int client_socketfd = accept(fd, (struct sockaddr *) &client_addr, &in_size); //等待接受请求，这边是阻塞式的  
	if (client_socketfd < 0) 
	{
		puts("accpet error");
		exit(1);
	}

	printf("do_accept! fd=%d threadid=%d\n", client_socketfd, std::this_thread::get_id());

	//类型转换
	struct event_base *base_ev = (struct event_base *)arg;

	//socket发送欢迎信息  
	char msg[] = "Welcome to Connect!";
	int size = send(client_socketfd, msg, strlen(msg), 0);

	//创建一个事件，这个事件主要用于监听和读取客户端传递过来的数据 持久类型
#if 0
	struct event *ev = event_new(base_ev, client_socketfd, EV_TIMEOUT | EV_READ | EV_PERSIST, do_read, event_self_cbarg());
	event_add(ev, NULL);
#else
	struct bufferevent *bev = bufferevent_socket_new(base_ev, client_socketfd, BEV_OPT_CLOSE_ON_FREE);
	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);
#endif

}

void signal_cb(evutil_socket_t fd, short event, void *arg)
{
	struct event *signal = (struct event *)arg;

	printf("signal_cb: got signal %d\n", event_get_signal(signal));
	event_del(signal);

	event_base_loopbreak(event_get_base(signal));
}

int main(int argc, char **argv)
{
	//初始化windows socket库
	WORD wVersionRequested = MAKEWORD(2, 2);
	WSADATA wsa_data;
	WSAStartup(0x0201, &wsa_data);
 
	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(8888); //服务器端口号    

	//创建服务端套接字  
	int 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)) < 0) 
	{
		puts("bind error");
		return 0;
	}

	//监听 监听队列长度  
	listen(server_socketfd, 10);

	printf("Server Start! main threadid=%d\n", std::this_thread::get_id());

	auto func = [=]() 
	{
		//创建event_base 事件的集合，多线程的话 每个线程都要初始化一个event_base
		struct event_base *base_ev = event_base_new();

		//创建一个事件，类型为持久性EV_PERSIST，回调函数为do_accept（主要用于监听连接进来的客户端）
		//将base_ev传递到do_accept中的arg参数
		struct event *ev = event_new(base_ev, server_socketfd, EV_TIMEOUT | EV_READ | EV_PERSIST, do_accept, base_ev);
		event_add(ev, NULL);

		//信号事件
		struct event *signal_int = evsignal_new(base_ev, SIGINT, signal_cb, event_self_cbarg());
		event_add(signal_int, NULL);

		//事件循环
		event_base_dispatch(base_ev);
		//销毁event_base
		event_base_free(base_ev);
	};

	std::thread thread(func);
	thread.join();

	printf("Server Stop\n");
	WSACleanup();

	getchar();
	return 0;
}