您现在的位置是：首页 >技术杂谈 >Linux 五种网络IO模式(阻塞IO、非阻塞IO、IO多路复用、信号驱动IO、异步IO)网站首页技术杂谈
Linux 五种网络IO模式(阻塞IO、非阻塞IO、IO多路复用、信号驱动IO、异步IO)
Linux网络编程中，有五种网络IO模式，分别是阻塞IO、非阻塞IO、IO多路复用、信号驱动IO、异步IO；
虽然说不能全都认识得很透彻，但至少得都知道一点！
开始之前，先了解以下同步IO和异步IO；
1. 同步IO
场景1： 小明去打开水，而开水塔此时没有水，小明在现场一直等待开水到来，或者不断的轮询查看是否有开水，直到有开水取到水为止，这是同步IO的一种案例！
同步IO的特点：
同步IO指的是用户进程触发I/O操作并等待或者轮询的去查看I/O操作是否就绪。
同步IO的执行者是IO操作的发起者。
同步IO需要发起者进行内核态到用户态的数据拷贝过程，所以这里必须阻。
2. 异步IO
场景2： 小明去打开水，而开水塔此时没有水，开水塔的阿姨叫小明把水壶放到现场，来水后会帮他打好水，并打电话叫他来取，这是异步IO的一种案例！
异步IO的特点：
异步IO是指用户进程触发I/O操作以后就立即返回，继续开始做自己的事情，而当I/O操作已经完成的时候会得到I/O完成的通知；
异步IO的执行者是内核线程，内核线程将数据从内核态拷贝到用户态，所以这里没有阻塞。
目录
一、阻塞IO
小明同学急用开水，打开水时发现开水龙头没水，他一直等待直到装满水然后离开。这一过程就可以看成是使用了阻塞IO模型，因为如果水龙头没有水，他也要等到有水并装满杯子才能离开去做别的事情。很显然，这种IO模型是同步的。
在linux 中，默认情况下所有的socket都是blocking IO(阻塞IO)， 一个典型的读操作流程：
二、非阻塞IO
小明同学又一次急用开水，打开水龙头后发现没有水，因为还有其它急事他马上离开了，过一会他又拿着杯子来看看……在中间离开的这些时间里，小明同学离开了装水现场(回到用户进程空间)，可以做他自己的事情。这就是非阻塞IO模型。但是它只有是检查无数据的时候是非阻塞的，在数据到达的时候依然要等待复制数据到用户空间(等着水将水杯装满)，因此它还是同步IO。
当用户线程发起一个read操作后，并不需要等待，而是马上就得到了一个结果。如果结果是一个error时，它就知道数据还没有准备好，于是它可以再次发送read操作。一旦内核中的数据准备好了，并且又再次收到了用户线程的请求，那么它马上就将数据拷贝到了用户线程，然后返回。
所以事实上，在非阻塞IO模型中，用户线程需要不断地询问内核数据是否就绪，也就说非阻塞IO不会交出CPU，而会一直占用CPU。
典型的非阻塞IO模型一般如下：
设置非阻塞常用方式：
方式一：创建socket 时指定
// SOCK_NONBLOCK: 非阻塞IO
server_sockfd = socket(AF_INET, SOCK_DGRAM | SOCK_NONBLOCK, 0); 方式二：在读取数据前通过如下方式设定
fcntl(server_sockfd, F_SETFL, fcntl(server_sockfd, F_GETFL, 0) | O_NONBLOCK);需要包含头文件 #include <fcntl.h> // fcntl
在读取数据接口中，如果还没有数据可以读取，则会立刻返回EAGAIN 和 EWOULDBLOCK；
所以只需要判断返回值即可，如：
int recv_len = recvfrom(server_sockfd, buff, sizeof(buff), 0, (struct sockaddr *)&client_addr, &client_addr_len);
if (recv_len < 0) {
if (EAGAIN == errno || EWOULDBLOCK == errno) {
printf("EAGAIN = %d EWOULDBLOCK = %d errno = %d
", EAGAIN, EWOULDBLOCK, errno);
printf("do something...
");
sleep(2);
continue;
}
perror("recvfrom");
exit(errno);
}
do something... 即可以去做其它事情，做完后再回来看看是否有数据可以读取了！
设置端口复用
int opt = 1;
setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));三、IO多路复用
有一天，学校里面优化了热水的供应，增加了很多水龙头，这个时候小明同学再去装水，舍管阿姨告诉他这些水龙头都还没有水，你可以去忙别的了，等有水了告诉他。于是等啊等(select调用中)，过了一会阿姨告诉他有水了。
这里有两种情况：
情况1: 阿姨只告诉来水了，但没有告诉小明是哪个水龙头来水了，要自己一个一个去尝试。(select/poll 场景)；
即1000个socket，其中一个socket有消息来了，阿姨就会通知，让我们一个一个 的取遍历，找到有数据的那个socket，然后读取数据；
情况2: 舍管阿姨会告诉小明同学哪几个水龙头有水了，小明同学不需要一个个打开看(epoll 场景)；
即1000个socket，其中一个socket有消息来了，阿姨就会通知到具体某一个socket来信息了，不需要我们一个一个的去遍历，效率一下子就高起来了；
当用户进程调用了select，那么整个进程就会被block(阻塞)，而同时，kernel会 “监视”所有select负责的socket，当任何一个socket中的数据准备好了，select就会返回。这个时候用户进程再调用read操作，将数据从kernel(内核)拷贝到用户进程。
所以，IO多路复用的特点是通过一种机制，一个进程能同时等待多个文件描述符，而这些文件描述符（套接字描述符）其中的任意一个进入就绪状态，select()函数就可以返回。
这里需要使用两个system call（select 和 recvfrom），而blocking IO(阻塞IO)只调用了一个system call（recvfrom）。但是，用select的优势在于它可以同时处理多个connection。
如果处理的连接数不是很高的话，使用select/epoll的web server不一定比使用mutil-threading + blocking IO（多线程 + 阻塞IO）的web server性能更好，可能延迟还更大。select/epoll 的优势并不是对于单个连接能处理得更好，而是在于能同时处理更多的连接。
1. select
在一段指定的时间内，监听用户感兴趣的文件描述符,可读、可写和异常等事件。
#include <sys/select.h>
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>
int select(int nfds, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout);
描述：监听多个(最多1024)socket；
参数
nfds
最大的文件描述符加1，使用FD_SETSIZE即可；
readfds
用于监听可读read；不关心则使用：(fd_set *)0;
writefds
用于监听可写write；不关心则使用：(fd_set *)0;
exceptfds
用于监听异常的数据；不关心则使用：(fd_set *)0;
timeout
一个指向timeval结构的指针，用于决定select等待I/o的最长时间；如果为空将一直等待;
返回值
大于0：是已就绪的文件句柄的总数；
等于0：超时；
小于0：表示出错，错误: errno；
void FD_CLR(int fd, fd_set *set); // 一个 fd_set类型变量的所有位都设为 0
int FD_ISSET(int fd, fd_set *set); // 清除某个位时可以使用
void FD_SET(int fd, fd_set *set); // 设置变量的某个位置位
void FD_ZERO(fd_set *set); // 测试某个位是否被置位
例：
select_server.c
#include <sys/types.h>
#include <sys/socket.h>
#include <stdio.h>
#include <sys/time.h>
#include <sys/ioctl.h>
#include <unistd.h>
#include <stdlib.h>
#include <netinet/in.h>
#include <string.h>
#define BUFF_MAX 1024
// 英文小写转换成大写
static void str2up(char *str) {
while (*str) {
if (*str >= 'a' && *str <= 'z') {
*str = *str - 'a' + 'A';
}
str++;
}
}
int main(void) {
int server_sockfd, client_sockfd;
int server_len, client_len;
struct sockaddr_in server_address;
struct sockaddr_in client_address;
int result = 0;
fd_set readfds, writefds, rfds, wfds;
char buff[BUFF_MAX] = { ' ' };
// 建立服务器socket
server_sockfd = socket(AF_INET, SOCK_STREAM, 0);
server_address.sin_family = AF_INET;
server_address.sin_addr.s_addr = htonl(INADDR_ANY);
server_address.sin_port = htons(9100);
server_len = sizeof(server_address);
// 绑定
bind(server_sockfd, (struct sockaddr *)&server_address, server_len);
// 监听，最多监听10个
listen(server_sockfd, 10);
// 清零
FD_ZERO(&readfds);
FD_ZERO(&writefds);
// 将服务端socket加入到集合中
FD_SET(server_sockfd, &readfds);
FD_SET(server_sockfd, &writefds);
while (1) {
int fd;
int nread;
// 将需要监视的描述符集拷贝到select查询队列中，select会对其修改，所以一定要分开使用变量
rfds = readfds;
wfds = writefds;
printf("server waiting...
");
// 无期限阻塞，并测试文件描述符变动
result = select(FD_SETSIZE, &rfds, &wfds, (fd_set *)0, (struct timeval *)0); // FD_SETSIZE，默认最大文件描述符，在这里最大是1024
if (result < 1) {
perror("select
");
exit(1);
} else if (0 == result) {
printf("time out!
");
}
// 扫描所有的文件描述符
for (fd = 0; fd < FD_SETSIZE; fd++) {
// 找到相关文件描述符，read
if (FD_ISSET(fd, &rfds)) {
// 判断是否为服务器套接字，是则表示客户端请求连接
if (fd == server_sockfd) {
client_len = sizeof(client_address);
client_sockfd = accept(server_sockfd, (struct sockaddr *)&client_address, &client_len);
// 将客户端socket加入到集合中
FD_SET(client_sockfd, &readfds);
//FD_SET(client_sockfd, &writefds);
printf("adding client on fd %d
", client_sockfd);
} else { // 客户端socket中有数据请求时
// 取得数据量交给nread
ioctl(fd, FIONREAD, &nread);
// 客户数据请求完毕，关闭套接字，从集合中清除相应描述符
if (0 == nread) {
close(fd);
FD_CLR(fd, &readfds); // 去掉g关闭的fd
printf("remove client on fd %d
", fd);
} else {
read(fd, buff, BUFF_MAX);
sleep(5);
printf("receive：%s
", buff);
printf("serving client on fd %d
", fd);
FD_SET(client_sockfd, &writefds);
}
}
} else if (FD_ISSET(fd, &wfds)) {
str2up(buff); // 转化为大写
write(fd, buff, sizeof(buff));
memset(buff, 0, BUFF_MAX);
FD_CLR(fd, &writefds);
} else {
//printf("其他
");
}
}
}
return 0;
}
cleint.c
#include <sys/types.h>
#include <sys/socket.h>
#include <stdio.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <stdlib.h>
#include <sys/time.h>
#define BUFF_MAX 1024
int main(int argc, char **argv) {
int client_sockfd;
int len;
struct sockaddr_in address; // 服务器网络地址结构体
int result;
char buff[BUFF_MAX] = { ' ' };
if (argc < 2) {
fprintf(stderr, "missing parameter
");
exit(1);
}
strcpy(buff, argv[1]);
// 建立客户端socket
client_sockfd = socket(AF_INET, SOCK_STREAM, 0);
address.sin_family = AF_INET;
address.sin_addr.s_addr = inet_addr("127.0.0.1");
address.sin_port = htons(9100);
len = sizeof(address);
// 连接服务器
result = connect(client_sockfd, (struct sockaddr *)&address, len);
if (-1 == result) {
perror("connect");
exit(1);
}
// 发送数据给服务器
write(client_sockfd, buff, strlen(buff));
memset(buff, ' ', BUFF_MAX);
// 接收服务器发回来的数据
read(client_sockfd, buff, BUFF_MAX);
printf("receive：%s
", buff);
sleep(3);
close(client_sockfd);
return 0;
}
2. poll
和select 一样，如果没有事件发生，则进入休眠状态，如果在规定时间内有事件发生，则返回成功，规定时间过后仍然没有事件发生则返回失败。可见，等待期间将进程休眠，利用事件驱动来唤醒进程，将更能提高CPU的效率。
poll 和select 区别： select 有文件句柄上线设置，值为FD_SETSIZE，而poll 理论上没有限制!
#include <poll.h>
int poll(struct pollfd *fds, nfds_t nfds, int timeout);
描述：等待文件描述符上的某个事件；
参数
fds
要监视的文件描述符集是在fds参数中指定的，可以传递多个结构体，也就是说可以监测多个驱动设备所产生的事件，只要有一个产生了请求事件，就能立即返回；
struct pollfd {
int fd; /* 文件描述符 open打开的那个 */
short events; /* 请求的事件类型，监视驱动文件的事件掩码 */ POLLIN | POLLOUT
short revents; /* 驱动文件实际返回的事件 */
}
事件类型events 可以为下列值：
POLLIN 有数据可读
POLLRDNORM 有普通数据可读，等效与POLLIN
POLLPRI 有紧迫数据可读
POLLOUT 写数据不会导致阻塞
POLLER 指定的文件描述符发生错误
POLLHUP 指定的文件描述符挂起事件
POLLNVAL 无效的请求，打不开指定的文件描述符
ndfs
监测驱动文件的个数;
timeout
超时时间，单位是ms;
返回值
有事件发生：返回revents域不为0的文件描述符个数；
超时：返回0；
失败：返回-1，并设置错误标志errno；
例：
poll_server.c
#include <sys/socket.h>
#include <stdio.h>
#include <sys/time.h>
#include <unistd.h>
#include <stdlib.h>
#include <netinet/in.h>
#include <poll.h>
#include <string.h>
#define BUFF_MAX 1024
#define MAX_FD 10240
struct pollfd fds[MAX_FD];
int cur_max_fd = 1;
// 英文小写转换成大写
void str2up(char *str) {
while (*str) {
if (*str >= 'a' && *str <= 'z') {
*str = *str - 'a' + 'A';
}
str++;
}
}
void setMaxFD(int fd) {
if (fd >= cur_max_fd) {
cur_max_fd = fd + 1;
}
}
int main(void) {
int server_sockfd, client_sockfd;
int server_len, client_len;
struct sockaddr_in server_address;
struct sockaddr_in client_address;
int result = 0;
char buff[BUFF_MAX] = { ' ' };
// 建立服务器socket
server_sockfd = socket(AF_INET, SOCK_STREAM, 0);
server_address.sin_family = AF_INET;
server_address.sin_addr.s_addr = htonl(INADDR_ANY);
server_address.sin_port = htons(9100);
server_len = sizeof(server_address);
// 绑定
bind(server_sockfd, (struct sockaddr *)&server_address, server_len);
// 监听，最多监听10个
listen(server_sockfd, 10);
// 将服务器添加到监听数组
fds[server_sockfd].fd = server_sockfd;
fds[server_sockfd].events = POLLIN | POLLOUT;
fds[server_sockfd].revents = 0;
setMaxFD(server_sockfd);
while (1) {
int fd = 0, i = 0;
int nread = 0;
printf("server waiting...
");
// 阻塞等待1秒，监听cur_max_fd个socket
result = poll(fds, cur_max_fd, 1000); // 1s
if (result < 0) {
perror("poll
");
exit(1);
} else if (0 == result) {
printf("time out!
");
} else {
// 扫描所有的文件描述符
for (i = 0; i < cur_max_fd; i++) {
// 找到相关文件描述符
if (fds[i].revents) {
fd = fds[i].fd;
// 判断是否为服务器套接字，是则表示客户端请求连接
if (fd == server_sockfd) {
client_len = sizeof(client_address);
client_sockfd = accept(server_sockfd, (struct sockaddr *)&client_address, &client_len);
// 将新的客户端添加到监听数组中
fds[client_sockfd].fd = client_sockfd;
fds[client_sockfd].events = POLLIN;
fds[client_sockfd].revents = 0;
setMaxFD(client_sockfd);
printf("adding client on fd %d
", client_sockfd);
} else { // 客户端socket中有数据请求时
/* 有数据可读 */
if (fds[i].revents & POLLIN) {
// 取得数据量交给nread
nread = read(fd, buff, BUFF_MAX);
// 客户数据请求完毕，关闭套接字，从集合中清除相应描述符
if (0 == nread) {
close(fd);
// 关闭后，监听的该socket需要清0
memset(&fds[i], 0, sizeof(struct pollfd));
printf("remove client on fd %d
", fd);
} else {
sleep(1);
printf("receive：%s
", buff);
printf("serving client on fd %d
", fd);
// 监听写事件，如果写准备好了，再写
fds[i].events = POLLOUT;
//write(fd, buff, sizeof(buff));
}
/* 有数据可写 */
} else if(fds[i].revents & POLLOUT) {
str2up(buff); // 转化为大写
write(fd, buff, strlen(buff));
// 设置回继续监听读数据状态
fds[i].events = POLLIN;
memset(buff, 0, BUFF_MAX);
}
}
}
}
}
}
return 0;
}
client.c
与select中的client.c代码一致！
3. epool
高并发中最常用的技术就是epoll了，epoll效率很高，比select和poll都要高很多！
epoll的事件队列是由红黑树构成的，因此，他获取数据的速度超快！
epoll一般由三个常用的函数构成，epoll_create 和 epoll_clt 和 epoll_wait。
1). epoll_create
#include <sys/epoll.h>
int epoll_create(int size);
描述：创建epoll句柄，打开一个epoll文件描述符；
参数
size
该参数已经没有意义，填大于0的整数即可！
返回值
成功：返回epoll文件描述符；
失败：返回-1，并设置错误标志errno；
2). epoll_ctl
#include <sys/epoll.h>
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
描述：向epoll对象中台南佳、修改或删除事件；
参数
epfd
epoll文件描述符；
op
取值如下：
EPOLL_CTL_ADD 添加心得事件到epoll中；
EPOLL_CTL_MOD 修改epoll中的事件；
EPOLL_CTL_DEL 删除epoll中的事件；
fd
socket文件描述符，被监视的文件描述符；
event
事件，结构体如下：
typedef union epoll_data {
void *ptr;
int fd;
uint32_t u32;
uint64_t u64;
} epoll_data_t;
struct epoll_event {
uint32_t events; /* Epoll events */
epoll_data_t data; /* User data variable */
};
events取值：
EPOLLIN 表示有数据可以读(或接受客户端连接和关闭连接)；
EPOLLOUT 表示有数据可以写(数据已准备好，可以发送出去)；
EPOLLERR 表示对应的连接发生错误；
EPOLLHUP 表示对应的连接被挂起；
返回值
成功：返回0；
失败：返回-1，并设置错误标志errno；
3). epoll_wait
#include <sys/epoll.h>
int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);
描述：收集在epoll监控的事件中已经发生的事件（在epoll文件描述符上等待I/O事件）；
参数
epfd
epoll文件描述符；
events
已经分配好的epoll_event结构体数组，epoll将会把发生的事件复制到events数组中（events不可以是空指针，内核只负责把数据复制到这个events数据中，不会帮助我们在用户态中分配内存。内核这种做法效率很高！）；
maxevents
本次可以返回的最大事件数目，通常maxevents参数与预分配的events数组的大小是相等的；
timeout
表示在没有检测到事件发生时最多等待的事件(单位为毫秒)，如果timeout为0，立即返回，不会等待；-1则表示无期限阻塞；3000则表示等待3秒；
返回值
大于0，则是返回有事件的文件符个数；
等于0，超时返回；
失败：返回-1，并设置错误标志；
4). 基于epoll的回声服务器案例
客户端循环1000次给服务器发送数据，每次循环睡眠2毫秒，服务器接收后在发送回给客户端，客户端正常接收。
server.c
#include <stdio.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <stdlib.h>
#include <errno.h>
#include <sys/epoll.h>
#include <unistd.h>
#include <fcntl.h>
#include <string.h>
#define SERVER_PORT 9100
#define BUFF_SIZE 1024
#define EPOLL_EVENT_SIZE 1024
int epoll_fd = 0;
typedef struct _ConnectStat {
char buff[BUFF_SIZE];
int fd;
struct epoll_event ev;
} ConnectStat;
// 设置不阻塞
void set_nonblock(int fd) {
int f1 = fcntl(fd, F_GETFL);
fcntl(fd, F_SETFL, f1 | O_NONBLOCK);
}
int startup(void) {
int return_value = -1;
int ret = 0;
int listen_socket = 0; // 服务器套接字
struct sockaddr_in server_addr;
// 创建通信套接字
listen_socket = socket(AF_INET, SOCK_STREAM, 0);
if (-1 == listen_socket) {
fprintf(stderr, "socket failed! reason: %s
", strerror(errno));
return return_value;
}
// 清空标志，写上地址和端口号
bzero(&server_addr, sizeof(server_addr));
server_addr.sin_family = AF_INET; // 选择协议组ipv4
server_addr.sin_addr.s_addr = htonl(INADDR_ANY); // 监听本地所有ip地址
server_addr.sin_port = htons(SERVER_PORT); // 绑定端口号
// 绑定
ret = bind(listen_socket, (struct sockaddr *)&server_addr, sizeof(server_addr));
if (-1 == ret) {
fprintf(stderr, "bind failed! reason: %s
", strerror(errno));
return return_value;
}
// 设置端口复用
int opt = 1;
setsockopt(listen_socket, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
// 监听
ret = listen(listen_socket, 16);
if (-1 == ret) {
fprintf(stderr, "listen failed! reason: %s
", strerror(errno));
return return_value;
}
return listen_socket;
}
ConnectStat *stat_init(int fd) {
ConnectStat *temp = (ConnectStat *)malloc(sizeof(ConnectStat));
if (!temp) {
fprintf(stderr, "ConnectStat malloc failed! reason: %s
", strerror(errno));
return NULL;
}
memset(temp, 0, sizeof(ConnectStat));
temp->fd = fd;
memset(temp->buff, ' ', sizeof(temp->buff));
return temp;
}
int connect_handle(int new_fd) {
ConnectStat *stat = stat_init(new_fd);
if (!stat) {
return -1;
}
set_nonblock(new_fd);
stat->ev.events = EPOLLIN;
stat->ev.data.ptr = stat;
// 添加到epoll监听池中
int ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, new_fd, &stat->ev);
if (-1 == ret) {
fprintf(stderr, "connect_handle epoll_ctl failed! reason: %s
", strerror(errno));
return -1;
}
return 0;
}
int main(int argc, char **argv) {
int ret = -1;
//int epoll_fd = 0;
int listen_sock = 0;
listen_sock = startup();
if (-1 == listen_sock) {
exit(-1);
}
// 创建eopll池
epoll_fd = epoll_create(256);
if (-1 == epoll_fd) {
fprintf(stderr, "epoll_create failed! reason: %s
", strerror(errno));
exit(1);
}
ConnectStat *stat = stat_init(listen_sock);
struct epoll_event ev;
ev.events = EPOLLIN; // 读 事件
ev.data.ptr = stat;
// 托管
ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, listen_sock, &ev);
if (-1 == ret) {
fprintf(stderr, "epoll_ctl failed! reason: %s
", strerror(errno));
exit(2);
}
struct epoll_event revs[EPOLL_EVENT_SIZE];
int timeout = 3000; // 3s
int num = 0;
while (1) {
// 监听事件
num = epoll_wait(epoll_fd, revs, EPOLL_EVENT_SIZE, timeout);
switch (num) {
case 0: {
printf("timeout!
");
}
break;
case -1: {
fprintf(stderr, "epoll_wait failed! reason: %s
", strerror(errno));
}
break;
default: {
struct sockaddr_in perr;
socklen_t len = sizeof(perr);
int i = 0;
for (; i < num; i++) {
// 获取数据结构体
ConnectStat *stat = (ConnectStat *) revs[i].data.ptr;
if (!stat) {
fprintf(stderr, "stat = NULL, i = %d
", i);
continue;
}
int fd = stat->fd; // 获取事件fd
if (fd == listen_sock && (revs[i].events & EPOLLIN)) {
// 链接客户端
int new_fd = accept(listen_sock, (struct sockaddr *)&perr, &len);
if (-1 == new_fd) {
printf("accept failed! reason: %s
", strerror(errno));
// 客户端已断开
} else {
printf("get a new client: %s : %d
", inet_ntoa(perr.sin_addr), ntohs(perr.sin_port));
connect_handle(new_fd);
}
// 有数据 读
} else if (revs[i].events & EPOLLIN) {
// 读取客户端发送过来的数据
int ret = read(fd, stat->buff, sizeof(stat->buff));
if (ret > 0) {
printf("%s
", stat->buff);
// 将当前fd设置为写状态，当数据准备好后，会触发写事件去发送数据
stat->ev.events = EPOLLOUT;
ret = epoll_ctl(epoll_fd, EPOLL_CTL_MOD, fd, &stat->ev);
if (-1 == ret) {
fprintf(stderr, "epoll_ctl failed! reason: %s
", strerror(errno));
continue;
}
// 客户端主动断开
} else if (0 == ret) {
printf("client %d close!
", fd);
epoll_ctl(epoll_fd, EPOLL_CTL_DEL, fd, NULL);
close(fd);
continue;
} else {
printf("read failed! reason: %s
", strerror(errno));
continue;
}
// 有数据 写
} else if (revs[i].events & EPOLLOUT) {
// 将收到的数据完整的发回给客户端
int ret = write(fd, stat->buff, strlen(stat->buff));
if (-1 == ret) {
fprintf(stderr, "write failed! reason: %s
", strerror(errno));
continue;
}
// 将当前fd重新设置回监听读数据状态
stat->ev.events = EPOLLIN;
ret = epoll_ctl(epoll_fd, EPOLL_CTL_MOD, fd, &stat->ev);
if (-1 == ret) {
fprintf(stderr, "(write)epoll_ctl failed! reason: %s
", strerror(errno));
continue;
}
} else {
printf("--else
");
}
}
}
break;
}
}
return 0;
}
client.c
#include <sys/types.h>
#include <sys/socket.h>
#include <stdio.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <stdlib.h>
#include <sys/time.h>
#include <string.h>
#define BUFF_MAX 1024
#define COUNT 1000
int main(int argc, char **argv) {
int client_sockfd;
int len;
struct sockaddr_in address; // 服务器网络地址结构体
int result;
char buff[BUFF_MAX] = { ' ' };
if (argc < 2) {
fprintf(stderr, "missing parameter
");
exit(1);
}
strcpy(buff, argv[1]);
// 建立客户端socket
client_sockfd = socket(AF_INET, SOCK_STREAM, 0);
address.sin_family = AF_INET;
address.sin_addr.s_addr = inet_addr("127.0.0.1");
address.sin_port = htons(9100);
len = sizeof(address);
// 连接服务器
result = connect(client_sockfd, (struct sockaddr *)&address, len);
if (-1 == result) {
perror("connect");
exit(1);
}
char buff1[BUFF_MAX] = { ' ' };
int a = COUNT;
while (a-- >= 0) {
// 发送数据给服务器
write(client_sockfd, buff, strlen(buff));
//memset(buff, ' ', BUFF_MAX);
// 接收服务器发回来的数据
read(client_sockfd, buff1, BUFF_MAX);
printf("receive：%s
", buff1);
usleep(2000);// 睡眠2毫秒
memset(buff1, ' ', BUFF_MAX);
}
close(client_sockfd);
return 0;
}
4. 水平触发和边缘触发
1). 水平触发
Level_triggered(水平触发)：当被监控的文件描述符上有可读写事件发生时，epoll_wait()会通知处理程序去读写。如果这次没有把数据一次性全部读写完(如读写缓冲区太小)，那么下次调用 epoll_wait()时，它还会通知你在上没读写完的文件描述符上继续读写，当然如果你一直不去读写，它会一直通知你！！！如果系统中有大量你不需要读写的就绪文件描述符，而它们每次都会返回，这样会大大降低处理程序检索自己关心的就绪文件描述符的效率！！！
设置方式: 默认即水平触发；
即上面的epoll服务器代码就是默认的水平触发！
2). 边缘触发
Edge_triggered(边缘触发)：当被监控的文件描述符上有可读写事件发生时，epoll_wait()会通知处理程序去读写。如果这次没有把数据全部读写完(如读写缓冲区太小)，那么下次调用epoll_wait()时，它不会通知你，也就是它只会通知你一次，直到该文件描述符上出现第二次可读写事件才会通知你！！！这种模式比水平触发效率高，系统不会充斥大量你不关心的就绪文件描述符！！！
设置方式: stat->_ev.events = EPOLLIN | EPOLLET
任何情况下，推荐使用边缘触发！
边缘案例：（回声服务器）
101234567890 10是int类型十，后面的都是数据；客户端需要组合成这样的数据结构然后发送给服务器，服务器接收到后也要进行解析，得到数据大小后再分配相应的内存进行读取数据；
客户端循环两千次，发送数据给服务器，每次循环睡眠两毫秒。
server.c
#include <stdio.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <stdlib.h>
#include <errno.h>
#include <sys/epoll.h>
#include <unistd.h>
#include <fcntl.h>
#include <string.h>
#define SERVER_PORT 9100
#define BUFF_SIZE 8
#define DATA_LEN_BYTES 4
#define EPOLL_EVENT_SIZE 1024
#define DATA_BUFF_SIZE 4096
int epoll_fd = 0;
//char data_buff[DATA_BUFF_SIZE] = { ' ' };
typedef struct _ConnectStat {
int fd; // 客户端socket
struct epoll_event ev;
char *data_buff; // 存储读取的数据
} ConnectStat;
// 设置不阻塞
void set_nonblock(int fd) {
int f1 = fcntl(fd, F_GETFL);
fcntl(fd, F_SETFL, f1 | O_NONBLOCK);
}
int startup(void) {
int return_value = -1;
int ret = 0;
int listen_socket = 0; // 服务器套接字
struct sockaddr_in server_addr;
// 创建通信套接字
listen_socket = socket(AF_INET, SOCK_STREAM, 0);
if (-1 == listen_socket) {
fprintf(stderr, "socket failed! reason: %s
", strerror(errno));
return return_value;
}
// 清空标志，写上地址和端口号
bzero(&server_addr, sizeof(server_addr));
server_addr.sin_family = AF_INET; // 选择协议组ipv4
server_addr.sin_addr.s_addr = htonl(INADDR_ANY); // 监听本地所有ip地址
server_addr.sin_port = htons(SERVER_PORT); // 绑定端口号
// 绑定
ret = bind(listen_socket, (struct sockaddr *)&server_addr, sizeof(server_addr));
if (-1 == ret) {
fprintf(stderr, "bind failed! reason: %s
", strerror(errno));
return return_value;
}
// 设置端口复用
int opt = 1;
setsockopt(listen_socket, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
// 监听
ret = listen(listen_socket, 16);
if (-1 == ret) {
fprintf(stderr, "listen failed! reason: %s
", strerror(errno));
return return_value;
}
return listen_socket;
}
ConnectStat *stat_init(int fd) {
ConnectStat *temp = (ConnectStat *)malloc(sizeof(ConnectStat));
if (!temp) {
fprintf(stderr, "ConnectStat malloc failed! reason: %s
", strerror(errno));
return NULL;
}
memset(temp, 0, sizeof(ConnectStat));
temp->fd = fd;
//memset(temp->data_buff, ' ', sizeof(temp->data_buff));
temp->data_buff = NULL;
return temp;
}
int connect_handle(int new_fd) {
ConnectStat *stat = stat_init(new_fd);
if (!stat) {
return -1;
}
set_nonblock(new_fd);
stat->ev.events = EPOLLIN | EPOLLET; // 边缘触发：EPOLLET
stat->ev.data.ptr = stat;
// 添加到epoll监听池中
int ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, new_fd, &stat->ev);
if (-1 == ret) {
fprintf(stderr, "connect_handle epoll_ctl failed! reason: %s
", strerror(errno));
return -1;
}
return 0;
}
int main(int argc, char **argv) {
int ret = -1;
//int epoll_fd = 0;
int listen_sock = 0;
listen_sock = startup();
if (-1 == listen_sock) {
exit(-1);
}
// 创建eopll池
epoll_fd = epoll_create(256);
if (-1 == epoll_fd) {
fprintf(stderr, "epoll_create failed! reason: %s
", strerror(errno));
exit(1);
}
ConnectStat *stat = stat_init(listen_sock);
struct epoll_event ev;
ev.events = EPOLLIN | EPOLLET; // 读 事件, EPOLLET: 边缘触发
ev.data.ptr = stat;
// 托管
ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, listen_sock, &ev);
if (-1 == ret) {
fprintf(stderr, "epoll_ctl failed! reason: %s
", strerror(errno));
exit(2);
}
struct epoll_event revs[EPOLL_EVENT_SIZE];
int timeout = 3000; // 3s
int num = 0;
while (1) {
// 监听事件
num = epoll_wait(epoll_fd, revs, EPOLL_EVENT_SIZE, timeout);
switch (num) {
case 0: {
printf("timeout!
");
}
break;
case -1: {
fprintf(stderr, "epoll_wait failed! reason: %s
", strerror(errno));
}
break;
default: {
struct sockaddr_in perr;
socklen_t len = sizeof(perr);
int ret = 0;
char buff[BUFF_SIZE];
int i = 0;
for (; i < num; i++) {
// 获取数据结构体
ConnectStat *stat = (ConnectStat *) revs[i].data.ptr;
if (!stat) {
fprintf(stderr, "stat = NULL, i = %d
", i);
continue;
}
int fd = stat->fd; // 获取事件fd
if (fd == listen_sock && (revs[i].events & EPOLLIN)) {
// 链接客户端
int new_fd = accept(listen_sock, (struct sockaddr *)&perr, &len);
if (-1 == new_fd) {
printf("accept failed! reason: %s
", strerror(errno));
continue;
// 客户端链接
} else {
printf("get a new client: %s : %d
", inet_ntoa(perr.sin_addr), ntohs(perr.sin_port));
connect_handle(new_fd);
}
// 有数据 读
} else if (revs[i].events & EPOLLIN) {
memset(buff, ' ', sizeof(buff));
// 读取客户端发送过来的数据
ret = read(fd, buff, DATA_LEN_BYTES); // 读取头部信息，四个字节
if (ret > 0) {
int buff_size = *(int *)buff; // 获取int类型数据，即获取数据的大小
// 根据接收数据大小分配相应的内存大小
stat->data_buff = (char *)malloc(buff_size + 1);
memset(stat->data_buff, ' ', buff_size + 1);
// 使用循环将数据全都读取出来
while (1) {
ret = read(fd, buff, sizeof(buff));
if (-1 == ret) {
if (errno == EAGAIN) {
//printf("data reive successes!
");
} else {
printf("data receive failed! reason: %s
", strerror(errno));
}
break;
}
// 将数据拷贝到大容量数组中保存
//memcpy(data_buff + strlen(data_buff), buff, ret); // 内存拷贝
strncat(stat->data_buff, buff, ret); // 在字符串尾部追加字符串
memset(buff, 0, sizeof(buff));
}
stat->data_buff[strlen(stat->data_buff)] = ' ';
printf("%s
", stat->data_buff);
// 将当前fd设置为写状态，当数据准备好后，会触发写事件去发送数据
stat->ev.events = EPOLLOUT | EPOLLET;
ret = epoll_ctl(epoll_fd, EPOLL_CTL_MOD, fd, &stat->ev);
if (-1 == ret) {
fprintf(stderr, "epoll_ctl failed! reason: %s
", strerror(errno));
continue;
}
// 客户端主动断开
} else if (0 == ret) {
printf("client %d close!
", fd);
ret = epoll_ctl(epoll_fd, EPOLL_CTL_DEL, fd, NULL);
if (-1 == ret) {
fprintf(stderr, "epoll_ctl failed! reason: %s
", strerror(errno));
continue;
}
close(fd);
free(stat);
stat = NULL;
continue;
} else {
printf("read failed! reason: %s
", strerror(errno));
continue;
}
// 有数据 写
} else if (revs[i].events & EPOLLOUT) {
// 将收到的数据完整的发回给客户端
int ret = write(fd, stat->data_buff, strlen(stat->data_buff));
if (-1 == ret) {
fprintf(stderr, "write failed! reason: %s
", strerror(errno));
continue;
}
// 将当前fd重新设置回监听读数据状态
stat->ev.events = EPOLLIN | EPOLLET;
ret = epoll_ctl(epoll_fd, EPOLL_CTL_MOD, fd, &stat->ev);
if (-1 == ret) {
fprintf(stderr, "(write)epoll_ctl failed! reason: %s
", strerror(errno));
continue;
}
} else {
printf("--else
");
}
}
}
break;
}
}
return 0;
}
client.c
#include <sys/types.h>
#include <sys/socket.h>
#include <stdio.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <stdlib.h>
#include <sys/time.h>
#include <string.h>
#define BUFF_MAX 1024
#define COUNT 2000
#define DATA_LEN_BYTES 4
int main(int argc, char **argv) {
int client_sockfd;
int len;
struct sockaddr_in address; // 服务器网络地址结构体
int result;
char *message = NULL;
char buff1[BUFF_MAX] = { ' ' };
if (argc < 2) {
fprintf(stderr, "missing parameter
");
exit(1);
}
message = argv[1];
int ms_len = strlen(message);
// 组装数据包
char *buff = (char *)malloc(ms_len + DATA_LEN_BYTES);
memset(buff, ' ', ms_len + DATA_LEN_BYTES);
*((int *)buff) = ms_len;
int size = *((int *)buff);
printf("size = %d
", size);
memcpy(buff + DATA_LEN_BYTES, message, ms_len);
// 建立客户端socket
client_sockfd = socket(AF_INET, SOCK_STREAM, 0);
address.sin_family = AF_INET;
address.sin_addr.s_addr = inet_addr("127.0.0.1");
address.sin_port = htons(9100);
len = sizeof(address);
// 连接服务器
result = connect(client_sockfd, (struct sockaddr *)&address, len);
if (-1 == result) {
perror("connect");
exit(1);
}
int a = COUNT;
while (a-- >= 0) {
// 发送数据给服务器
write(client_sockfd, buff, ms_len + DATA_LEN_BYTES);
// 接收服务器发回来的数据
read(client_sockfd, buff1, BUFF_MAX);
printf("receive：%s
", buff1);
usleep(2000); // 休眠2毫秒
memset(buff1, ' ', BUFF_MAX);
}
close(client_sockfd);
return 0;
}
5. 封装epoll框架
高性能、高并发，封装了epoll的框架资源-CSDN文库
https://download.csdn.net/download/cpp_learner/87630114
6. libevent
libevent是一个轻量级的开源的高性能的事件触发的网络库，适用于windows、linux、bsd等多种平台，内部使用select、epoll、kqueue等系统调用管理事件机制。
它被众多的开源项目使用，例如大名鼎鼎的memcached等。
特点：
事件驱动，高性能;
轻量级，专注于网络(相对于ACE);
开放源码，代码相当精炼、易读;
跨平台，支持Windows、Linux、BSD和Mac OS;
支持多种I/O多路复用技术（epoll、poll、dev/poll、select和kqueue等），在不同的操作系统下，做了多路复用模型的抽象，可以选择使用不同的模型，通过事件函数提供服务;
支持I/O，定时器和信号等事件;
采用Reactor模式;
libevent是一个典型的reactor模式的实现。
普通的函数调用机制：程序调用某个函数，函数执行，程序等待，函数将结果返回给调用程序（如果含有函数返回值的话），也就是顺序执行的。
Reactor模式的基本流程：应用程序需要提供相应的接口并且注册到reactor反应器上，如果相应的事件发生的话，那么reactor将自动调用相应的注册的接口函数（类似于回调函数）通知你，所以libevent是事件触发的网络库。
libevent的功能
Libevent提供了事件通知，io缓存事件，定时器，超时，异步解析dns，事件驱动的http server以及一个rpc框架。
事件通知：当文件描述符可读可写时将执行回调函数。
IO缓存：缓存事件提供了输入输出缓存，能自动的读入和写入，用户不必直接操作io。
定时器：libevent提供了定时器的机制，能够在一定的时间间隔之后调用回调函数。
信号：触发信号，执行回调。
异步的dns解析：libevent提供了异步解析dns服务器的dns解析函数集。
事件驱动的http服务器：libevent提供了一个简单的，可集成到应用程序中的HTTP服务器。
RPC客户端服务器框架：libevent为创建RPC服务器和客户端创建了一个RPC框架，能自动的封装和解封数据结构。
简单来讲，就是将需要监听的事件添加到监听队列中，当有某种事件触发时，会调用回调函数去处理！
1). libevent安装
解压命令： tar -zxvf libevent-2.1.12-stable.tar.gz
进入解压后的路径，依次执行以下命令：
./configure --disable-openssl
make
make install
使用命令检测是否安装成功：ls -la /usr/local/include | grep event
如果出现event2 和 event.h，说明已经安装成功了
头文件在路径：cd /usr/local/include
库文件在路径：cd /usr/local/lib
2). libevent主要API介绍
A. event_base_new
struct event_base *event_base_new(void);
event_base_new（）函数分配并且返回一个新的具有默认设置的event_base。函数会检测环境变量，返回一个到event_base的指针。如果发生错误，则返回NULL。选择各种方法时，函数会选择OS支持的最快方法。
大多数程序使用这个函数就够了。
event_base_new（）函数声明在 #include <event2/event.h> 中!
B. event_new
struct event *event_new(struct event_base *base, evutil_socket_t fd, short what, event_callback_fn cb, void *arg);
使用event_new（）接口创建事件;
参数三(what)使用以下宏：
EV_TIMEOUT
EV_READ
EV_WRITE
EV_SIGNAL
EV_PERSIST // 持续监听
EV_ET
参数4(cb)是一个回调函数，需要传一个函数进去，其类型是函数指针，类型如下：
typedef void (*event_callback_fn) (evutil_socket_t fd, short events, void *arg); // arg是传的参数参数5(arg)是给回调函数的参数；
event_new()试图分配和构造一个用于base的新的事件。what参数是上述标志的集合。如果fd非负，则它是将被观察其读写事件的文件。
事件被激活时，libevent将调用cb函数，传递这些参数：文件描述符fd，表示所有被触发事件的位字段，以及构造事件时的arg参数。
发生内部错误，或者传入无效参数时，event_new()将返回NULL。
所有新创建的事件都处于已初始化和非未决状态，调用event_add()可以使其成为未决的。
要释放事件，调用event_free()。对未决或者激活状态的事件调用event_free()是安全的：在释放事件之前，函数将会使事件成为非激活和非未决的。
void event_free(struct event *event);
例：
struct event *ev_listen = event_new(base, socket, EV_READ | EV_PERSIST, accept_connection, base);C. event_add
int event_add(struct event *ev, const struct timeval *timeout);
构造事件之后，在将其添加到event_base之前实际上是不能对其做任何操作的。使用event_add()将事件添加到event_base;
在非未决的事件上调用event_add()将使其在配置的event_base中成为未决的。
成功时函数返回0，失败时返回-1。
如果tv为NULL，添加的事件不会超时。否则，tv以秒和微秒指定超时值。
如果对已经未决的事件调用event_add()，事件将保持未决状态，并在指定的超时时间被重新调度。
注意：不要设置tv为希望超时事件执行的时间。如果在2010年1月1日设置“tv->tv_sec = time(NULL)+10;”，超时事件将会等待40年，而不是10秒。
例：
event_add(ev_listen, NULL);D. event_del
int event_del(struct event *ev);
对已经初始化的事件调用event_del()将使其成为非未决和非激活的。如果事件不是未决的或者激活的，调用将没有效果。成功时函数返回0，失败时返回-1。
注意：如果在事件激活后，其回调被执行前删除事件，回调将不会执行。
函数定义在<event2/event.h>中;
E. event_base_dispatch
int event_base_dispatch(struct event_base *event_base);
event_base_dispatch()等同于没有设置标志的event_base_loop()。所以，event_base_dispatch()将一直运行，直到没有已经注册的事件了，或者调用了event_base_loopbreak()或者event_base_loopexit()为止。
函数定义在<event2/event.h>中;
F. event_base_free
void event_base_free(struct event_base *base);
使用完event_base之后，使用event_base_free（）进行释放;
注意：这个函数不会释放当前与event_base关联的任何事件，或者关闭他们的套接字，或者释放任何指针。
event_base_free()定义在<event2/event.h>中;
G. event_set
void event_set(struct event *event, evutil_socket_t fd, short what, void (*Callback) (evutil_socket_t, short, void *), void *arg);
重新对事件进行设置；
调用此函数后，必须要将struct event_base *base 重新赋值给他，如下：
event_set(ev, fd, EV_READ, do_echo_handler, (void *)stat);
ev->ev_base = ev_base; // 必须重置事件集合然后需要将ev重新添加到base集合中；
event_add(ev, NULL);H. event_assign
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添加到base中；
除了event参数必须指向一个未初始化的事件之外，event_assign()的参数与event_new()的参数相同。成功时函数返回0，如果发生内部错误或者使用错误的参数，函数返回-1。
警告
不要对已经在event_base中未决的事件调用event_assign()，这可能会导致难以诊断的错误。如果已经初始化和成为未决的，调用event_assign()之前需要调用event_del()。
例：
event_assign(ev, base, sockfd, EV_READ , do_echo_request, (void*)stat);I. evconnlistener_new_bind
struct evconnlistener *evconnlistener_new_bind(struct event_base *base, evconnlistener_cb cb, void *ptr, unsigned flags, int backlog, const struct sockaddr *sa, int socklen);
listen、connect、bind、accept，创建监听对象，在指定的地址上监听接下来的TCP连接;
base参数都是监听器用于监听连接的event_base。cb是收到新连接时要调用的回调函数；如果cb为NULL，则监听器是禁用的，直到设置了回调函数为止。ptr指针将传递给回调函数。flags参数控制回调函数的行为。backlog是任何时刻网络栈允许处于还未接受状态的最大连接数。如果backlog是负的，libevent会试图挑选一个较好的值；如果为0，libevent认为已经对提供的套接字调用了listen()。
接收到新连接会调用提供的回调函数。listener参数是接收连接的连接监听器。sock参数是新接收的套接字。addr和len参数是接收连接的地址和地址长度。
包含头文件：#include <event2/listener.h>
参数二：
typedef void (*evconnlistener_cb) (struct evconnlistener *listener, evutil_socket_t sock, struct sockaddr *addr, int len, void *ptr);
参数四：
LEV_OPT_REUSEABLE 端口可重复监听，可重用；
LEV_OPT_CLOSE_ON_FREE 释放连接监听器会关闭底层套接字；
LEV_OPT_CLOSE_ON_EXEC 连接监听器会为底层套接字设置 close-on-exec；
......
例：
struct sockaddr_in sin;
memset(&sin, 0, sizeof(struct sockaddr_in));
sin.sin_family = AF_INET;
sin.sin_port = htons(9999);
//server_addr.sin_addr.s_addr = inet_addr("127.0.0.1");
base = event_base_new();
struct evconnlistener *listener = evconnlistener_new_bind(base, listener_cb, base,
LEV_OPT_REUSEABLE | LEV_OPT_CLOSE_ON_FREE,
10, (struct sockaddr *)&sin,
sizeof(struct sockaddr_in)); J. evconnlistener_free
void evconnlistener_free(struct evconnlistener *lev);
释放连接监听器;
K. bufferevent_read
size_t bufferevent_read(struct bufferevent *bufev, void *data, size_t size);
从bufferevent的输入缓冲区读取数据；
bufferevent_read()至多从输入缓冲区移除size字节的数据，将其存储到内存中data处。函数返回实际移除的字节数。
注意，对于bufferevent_read()，data处的内存块必须有足够的空间容纳size字节数据。
L. bufferevent_write
int bufferevent_write(struct bufferevent *bufev, const void *data, size_t size);
向bufferevent的输出缓冲区添加数据;
bufferevent_write()将内存中从data处开始的size字节数据添加到输出缓冲区的末尾;
成功时返回0，发生错误时则返回-1;
M. bufferevent_socket_new
struct bufferevent *bufferevent_socket_new(struct event_base *base, evutil_socket_t fd, enum bufferevent_options options);
base是event_base，options是表示bufferevent选项（BEV_OPT_CLOSE_ON_FREE等）的位掩码，fd是一个可选的表示套接字的文件描述符。如果想以后设置文件描述符，可以设置fd为-1。
成功时函数返回一个bufferevent，失败则返回NULL。
例：
// 针对已经存在的socket创建bufferevent对象
// BEV_OPT_CLOSE_ON_FREE:如果释放bufferevent对象，则关闭连接
struct bufferevent *bev = bufferevent_socket_new(base, fd, BEV_OPT_CLOSE_ON_FREE);
// BEV_OPT_CLOSE_ON_FREE:释放bufferevent时关闭底层传输端口。这将关闭底层套接字，释放底层bufferevent等。N. bufferevent_setcb
void bufferevent_setcb(struct bufferevent *bufev, bufferevent_data_cb readcb, bufferevent_data_cb writecb, bufferevent_event_cb eventcb, void *cbarg);
typedef void (*bufferevent_data_cb) (struct bufferevent *bev, void *ctx);
typedef void (*bufferevent_event_cb) (struct bufferevent *bev, short events, void *ctx);
bufferevent_setcb()函数修改bufferevent的一个或者多个回调。readcb、writecb和eventcb函数将分别在已经读取足够的数据、已经写入足够的数据，或者发生错误时被调用。每个回调函数的第一个参数都是发生了事件的bufferevent，最后一个参数都是调用bufferevent_setcb()时用户提供的cbarg参数：可以通过它向回调传递数据。
要禁用回调，传递NULL而不是回调函数。注意：bufferevent的所有回调函数共享单个cbarg，所以修改它将影响所有回调函数。
O. bufferevent_enable
int bufferevent_enable(struct bufferevent *bufev, short event);
函数内部调用了event_add将相应读写事件加入事件监听队列;
例：
bufferevent_enable(bev, EV_READ | EV_PERSIST); // evnet_add,使bufferevent 生效P. 其他
// 允许多次绑定同一个地址，要用在socket和bind之间
evutil_make_listen_socket_reuseable(listener);// 关闭套接字
evutil_closesocket(socket);// 跨平台统一接口，将套接字设置为非阻塞状态
evutil_make_socket_nonblocking(listener);3). 回声服务器案例一
#include <stdio.h>
#include <string.h>
#include <errno.h>
#include <unistd.h>
#include <stdlib.h>
#include <event.h>
#include <event2/event.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <fcntl.h>
#include <assert.h>
typedef struct _ConnectStat ConnectStat;
#define BUFLEN 1024
struct _ConnectStat {
int fd;
char send_buf[BUFLEN];
struct event *ev;
};
//echo 服务实现相关代码
ConnectStat * stat_init(int fd);
void do_echo_handler(int fd, short events, void *arg);
void do_welcome_handler(int fd, short events, void *arg);
void do_echo_response(int fd, short events, void *arg);
void accept_connection(int fd, short events, void *arg);
struct event_base *ev_base;
void usage(const char* argv)
{
printf("%s:[ip][port]
", argv);
}
// 设置不阻塞
void set_nonblock(int fd)
{
int fl = fcntl(fd, F_GETFL);
fcntl(fd, F_SETFL, fl | O_NONBLOCK);
}
int startup(char* _ip, int _port) //创建一个套接字，绑定，检测服务器
{
//sock
//1.创建套接字
int sock = socket(AF_INET, SOCK_STREAM, 0);
if (sock < 0)
{
perror("sock");
exit(2);
}
// 端口复用
int opt = 1;
setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
//2.填充本地 sockaddr_in 结构体（设置本地的IP地址和端口）
struct sockaddr_in local;
local.sin_port = htons(_port);
local.sin_family = AF_INET;
local.sin_addr.s_addr = inet_addr(_ip);
//3.bind（）绑定
if (bind(sock, (struct sockaddr*)&local, sizeof(local)) < 0)
{
perror("bind");
exit(3);
}
//4.listen（）监听 检测服务器
if (listen(sock, 5) < 0)
{
perror("listen");
exit(4);
}
return sock; //这样的套接字返回
}
ConnectStat * stat_init(int fd) {
ConnectStat * temp = NULL;
temp = (ConnectStat *)malloc(sizeof(ConnectStat));
if (!temp) {
fprintf(stderr, "malloc failed. reason: %m
");
return NULL;
}
memset(temp, ' ', sizeof(ConnectStat));
temp->fd = fd;
//temp->status = 0;
}
//void do_welcome_handler(int fd, void *data) {
void do_welcome_handler(int fd, short events, void *arg) {
const char * WELCOME= "Welcome.
";
int wlen = strlen(WELCOME);
int n ;
ConnectStat * stat = (ConnectStat *)(arg);
if( (n = write(fd, WELCOME, wlen)) != wlen ){
if(n<=0){
fprintf(stderr, "write failed[len:%d], reason: %s
",n,strerror(errno));
} else fprintf(stderr, "send %d bytes only ,need to send %d bytes.
",n,wlen);
} else {
event_set(stat->ev, fd, EV_READ, do_echo_handler, (void *)stat);
stat->ev->ev_base = ev_base; // 必须重置事件集合
event_add(stat->ev, NULL);
}
}
void do_echo_handler(int fd, short events, void *arg) {
ConnectStat * stat = (ConnectStat *)(arg);
char * p = NULL;
assert(stat!=NULL);
p = stat->send_buf;
*p++ = '-';
*p++ = '>';
ssize_t _s = read(fd, p, BUFLEN - (p - stat->send_buf) - 1); //2字节"->" +字符结束符.
if (_s > 0)
{
*(p+_s) = ' '; // p[_s] = ' ';
printf("receive from client: %s", p);
if(!strncasecmp(p, "quit", 4)){//退出.
event_free(stat->ev); // 自动解除监听事件，再释放event
close(fd);
free(stat);
return ;
}
event_set(stat->ev, fd, EV_WRITE, do_echo_response, (void *)stat);
stat->ev->ev_base = ev_base; // 必须重置事件集合
event_add(stat->ev, NULL);
}else if (_s == 0) //client:close
{
fprintf(stderr,"Remote connection[fd: %d] has been closed
", fd);
event_free(stat->ev); // 自动解除监听事件，再释放event
close(fd);
free(stat);
}
else //err occurred.
{
fprintf(stderr,"read faield[fd: %d], reason:%s [%ld]
",fd , strerror(errno), _s);
}
}
void do_echo_response(int fd, short events, void *arg) {
ConnectStat * stat = (ConnectStat *)(arg);
int len = strlen(stat->send_buf);
int _s = write(fd, stat->send_buf, len);
memset(stat->send_buf, ' ', len);
if(_s>0){
event_set(stat->ev, fd, EV_READ, do_echo_handler, (void *)stat);
stat->ev->ev_base = ev_base; // 必须重置事件集合
event_add(stat->ev, NULL);
}else if(_s==0){
fprintf(stderr,"Remote connection[fd: %d] has been closed
", fd);
event_free(stat->ev); // 自动解除监听事件，再释放event
close(fd);
free(stat);
}else {
fprintf(stderr,"read faield[fd: %d], reason:%s [%d]
",fd ,strerror(errno), _s);
}
}
//read()
//注册写事件
//写事件就绪
//write()
void accept_connection(int fd, short events, void *arg) {
struct sockaddr_in peer;
socklen_t len = sizeof(peer);
int new_fd = accept(fd, (struct sockaddr*)&peer, &len);
if (new_fd > 0)
{
ConnectStat *stat = stat_init(new_fd);
set_nonblock(new_fd);
printf("new client: %s:%d
", inet_ntoa(peer.sin_addr), ntohs(peer.sin_port));
struct event *ev = event_new(ev_base, new_fd, EV_WRITE, do_welcome_handler, (void *)stat);
stat->ev = ev;
event_add(ev, NULL);
} else {
fprintf(stderr, "error!
");
}
}
int main(int argc,char **argv){
if (argc != 3) //检测参数个数是否正确
{
usage(argv[0]);
exit(1);
}
int listen_sock = startup(argv[1], atoi(argv[2])); //创建一个绑定了本地 ip 和端口号的套接字描述符
//初始化异步事件处理框架epoll
ev_base = event_base_new();
//ConnectStat * stat = stat_init(listen_sock);
struct event *ev_listen = event_new(ev_base, listen_sock, EV_READ|EV_PERSIST, accept_connection, (void*)ev_base);
event_add(ev_listen, NULL);
event_base_dispatch(ev_base);
event_base_free(ev_base);
return 0;
}
4). 回声服务器案例二
client.c
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <errno.h>
#include <unistd.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <event.h>
#include <event2/util.h>
int connect_server(const char *server_ip, int port);
void cmd_read_data(int fd, short events, void *arg);
void socket_read_data(int fd, short events, void *arg);
int main(int argc, char **argv) {
if (argc < 3) {
printf("please input 2 parameters!
");
return -1;
}
// 两个参数依次是服务器的IP地址和端口号
int sockfd = connect_server(argv[1], atoi(argv[2]));
if (-1 == sockfd) {
perror("tcp_connect error!");
return -1;
}
printf("connect to server successfully
");
struct event_base *base = event_base_new();
// 监听服务端发送的消息
struct event *ev_sockfd = event_new(base, sockfd, EV_READ | EV_PERSIST, socket_read_data, NULL);
event_add(ev_sockfd, NULL);
// 监听终端输入事件
struct event *ev_cmd = event_new(base, STDIN_FILENO, EV_READ | EV_PERSIST, cmd_read_data, (void *)&sockfd);
event_add(ev_cmd, NULL);
// 事件循环
event_base_dispatch(base);
event_base_free(base);
printf("finished
");
return 0;
}
void cmd_read_data(int fd, short events, void *arg) {
char msg[1024] = { ' ' };
int ret = read(fd, msg, sizeof(msg) - 1);
if (0 == ret) {
printf("connection close. exit!
");
exit(1);
}
if (ret < 0) {
perror("read failed!");
exit (1);
}
int sockfd = *((int *)arg);
if (msg[ret - 1] == '
') {
msg[ret - 1] = ' ';
} else {
msg[ret] = ' ';
}
/*
int count = 1000;
while (count-- > 0) {
write(sockfd, msg, ret);
int len = read(sockfd, msg, sizeof(msg) - 1);
msg[len] = ' ';
printf("recv from server <<< %s
", msg);
usleep(10000);
}
*/
// 把终端的消息发送给服务器端，客户端忽略性能考虑，直接利用阻塞方式发送
//printf("write to server >>> %s
", msg);
ret = write(sockfd, msg, ret);
if (ret == -1) {
perror("write to server failed!");
exit(1);
}
if (strncmp(msg, "exit", 4) == 0) {
memset(msg, 0, sizeof(msg));
write(sockfd, msg, sizeof(msg));
usleep(100000); // 100ms
close(sockfd);
exit(1);
}
}
void socket_read_data(int fd, short events, void *arg) {
char msg[1024] = { ' ' };
// 不考虑一次读不完数据的情况
int len = read(fd, msg, sizeof(msg) - 1);
if (0 == len) {
printf("connection close. exit!
");
exit(1);
} else if (len < 0) {
perror("read failed!");
return ;
}
msg[len] = ' ';
printf("recv from server <<< %s
", msg);
}
typedef struct sockaddr SA;
int connect_server(const char *server_ip, int port) {
int sockfd, status, save_errno;
struct sockaddr_in server_addr;
memset(&server_addr, 0, sizeof(server_addr));
server_addr.sin_family = AF_INET;
server_addr.sin_port = htons(port);
status = inet_aton(server_ip, &server_addr.sin_addr);
if (0 == status) {
errno = EINVAL;
return -1;
}
sockfd = socket(PF_INET, SOCK_STREAM, 0);
status = connect(sockfd, (SA *)&server_addr, sizeof(server_addr));
if (-1 == status) {
save_errno = errno;
close(sockfd);
errno = save_errno; // the close may be error
return -1;
}
return sockfd;
}server.c
#include <stdio.h>
#include <string.h>
#include <errno.h>
#include <stdlib.h>
#include <unistd.h>
#include <event.h>
#include <event2/event.h>
#include <assert.h>
#define BUFLEN 1024
typedef struct _ConnectStat {
struct event *ev;
char buf[BUFLEN];
}ConnectStat;
// echo 服务实现相关代码
ConnectStat *stat_init(struct event *ev);
void accept_connection(int fd, short events, void *arg);
void do_echo_request(int fd, short events, void *arg);
void do_echo_response(int fd, short events, void *arg);
int tcp_server_init(int port, int listen_num);
struct event_base *base;
int main(int argc, char **argv) {
int listener = tcp_server_init(9999, 1024);
if (-1 == listener) {
perror("tcp_server_init error");
return -1;
}
base = event_base_new();
// 添加监听客户端请求链接事件
struct event *ev_listen = event_new(base, listener, EV_READ | EV_PERSIST, accept_connection, base);
event_add(ev_listen, NULL);
event_base_dispatch(base);
event_base_free(base);
return 0;
}
ConnectStat *stat_init(struct event *ev) {
ConnectStat *temp = (ConnectStat *)malloc(sizeof(ConnectStat));
if (!temp) {
fprintf(stderr, "malloc failed. reason: %s
", strerror(errno));
return NULL;
}
memset(temp, ' ', sizeof(ConnectStat));
temp->ev = ev;
return temp;
}
void accept_connection(int fd, short events, void *arg) {
evutil_socket_t sockfd;
struct sockaddr_in client;
socklen_t len = sizeof(client);
sockfd = accept(fd, (struct sockaddr *)&client, &len);
if (-1 == sockfd) {
fprintf(stderr, "accept failed! reason: %s
", strerror(errno));
sleep(1);
return ;
}
evutil_make_socket_nonblocking(sockfd); // 设置为非阻塞
char server_ip[64];
inet_ntop(AF_INET, &client.sin_addr.s_addr, server_ip, sizeof(server_ip));
printf("new client: %s:%u
", server_ip, ntohs(client.sin_port));
struct event_base *base = (struct event_base *)arg;
// 仅仅是为了动态创建一个event结构体
struct event *ev = event_new (NULL, -1, 0, NULL, NULL);
ConnectStat *stat = stat_init(ev);
//将动态创建的结构体作为event的回调参数
event_assign(ev, base, sockfd, EV_READ , do_echo_request, (void*)stat);
event_add(ev, NULL);
}
void do_echo_request(int fd, short events, void *arg) {
ConnectStat *stat = (ConnectStat *)arg;
struct event *ev = stat->ev;
char *msg = stat->buf;
//printf("do echo request ...
");
int len = read(fd, msg, BUFLEN - 1);
if (0 == len) {
fprintf(stdin, "connection break !
");
event_free(ev);
close(fd);
free(stat);
return ;
} else if (len < 0) {
fprintf(stderr, "read failed! reason: %s
", strerror(errno));
return ;
}
msg[len] = ' ';
printf("recv from client <<< %s
", msg);
event_set(ev, fd, EV_WRITE, do_echo_response, (void *)stat);
ev->ev_base = base;
event_add(ev, NULL);
}
void do_echo_response(int fd, short events, void *arg) {
ConnectStat *stat = (ConnectStat *)(arg);
struct event *ev = NULL;
assert(stat != NULL);
ev = stat->ev;
int len = strlen(stat->buf);
printf("write to client >>> %s
", stat->buf);
int _s = write(fd, stat->buf, len);
if (_s > 0) {
//printf("write successfully.
");
} else if (0 == _s) { // client close
fprintf(stderr, "remote connection[fd: %d] has been closed
", fd);
event_free(ev);
close(fd);
free(stat);
return ;
} else {
fprintf(stderr, "write failed[fd: %d], reason: %s[%d]
", fd, strerror(errno), _s);
//event_free(ev);
//close(fd);
//free(stat);
return ;
}
event_set(ev, fd, EV_READ, do_echo_request, (void *)stat);
ev->ev_base = base;
event_add(ev, NULL);
//event_assign(ev, base, fd, EV_READ , do_echo_request, (void*)stat);
}
typedef struct sockaddr SA;
int tcp_server_init(int port, int listen_num) {
int errno_save;
evutil_socket_t listener; // int listener;
listener = socket(AF_INET, SOCK_STREAM, 0);
if (-1 == listener) {
return -1;
}
// 允许多次绑定同一个地址，要用在socket和bind之间
evutil_make_listen_socket_reuseable(listener);
struct sockaddr_in sin;
sin.sin_family = AF_INET;
sin.sin_addr.s_addr = 0;
sin.sin_port = htons(port);
if (bind(listener, (SA *)&sin, sizeof(sin)) < 0) {
errno_save = errno;
evutil_closesocket(listener);
errno = errno_save;
return -1;
}
if(listen(listener, listen_num) < 0) {
errno_save = errno;
evutil_closesocket(listener);
errno = errno_save;
return -1;
}
// 跨平台统一接口，将套接字设置为非阻塞状态
evutil_make_socket_nonblocking(listener);
return listener;
}
5). 带有缓存的回声服务器案例
#include <stdio.h>
#include <string.h>
#include <errno.h>
#include <stdlib.h>
#include <unistd.h>
#include <event.h>
#include <event2/event.h>
#include <event2/listener.h>
#include <assert.h>
#include <arpa/inet.h>
#define BUFLEN 1024
typedef struct _ConnectStat {
struct bufferevent *bev;
char buf[BUFLEN];
}ConnectStat;
ConnectStat *stat_init(struct bufferevent *bev);
void do_echo_request(struct bufferevent *bev, void *arg); // 读数据
void do_echo_response(struct bufferevent *bev, void *arg); // 写数据
void event_cb(struct bufferevent *bev, short event, void *arg); // 出错处理函数
int tcp_server_init(int port, int listen_num);
void listener_cb(struct evconnlistener *listener, evutil_socket_t fd, struct sockaddr *sock, int socklen, void *arg); // 监听函数
struct event_base *base;
int main(int argc, char **argv) {
struct sockaddr_in sin;
memset(&sin, 0, sizeof(struct sockaddr_in));
sin.sin_family = AF_INET;
sin.sin_port = htons(9999);
//server_addr.sin_addr.s_addr = inet_addr("127.0.0.1");
base = event_base_new();
// 创建socket，绑定、监听、接受链接
// 创建监听对象，在指定的地址上监听接下来的TCP连接
// listen、connect、bind、accept； LEV_OPT_REUSEABLE：可重用，LEV_OPT_CLOSE_ON_FREE：自动关闭
struct evconnlistener *listener = evconnlistener_new_bind(base, listener_cb, base,
LEV_OPT_REUSEABLE | LEV_OPT_CLOSE_ON_FREE,
1024, (struct sockaddr *)&sin,
sizeof(struct sockaddr_in));
// 监听集合中的事件
event_base_dispatch(base);
// 释放
evconnlistener_free(listener);
event_base_free(base);
return 0;
}
ConnectStat *stat_init(struct bufferevent *bev) {
ConnectStat *temp = NULL;
temp = (ConnectStat *)malloc(sizeof(ConnectStat));
if (!temp) {
fprintf(stderr, "malloc failed. reason: %s
", strerror(errno));
return NULL;
}
memset(temp, ' ', sizeof(ConnectStat));
temp->bev = bev;
return temp;
}
void do_echo_request(struct bufferevent *bev, void *arg) {
ConnectStat *stat = (ConnectStat *)arg;
char *msg = stat->buf;
// 从缓冲区中获取数据
size_t len = bufferevent_read(bev, msg, BUFLEN);
msg[len] = ' ';
printf("recv from client <<< %s
", msg);
// 将数据添加到缓冲区
bufferevent_write(bev, msg, strlen(msg));
}
void do_echo_response(struct bufferevent *bev, void *arg) {
return ;
}
void event_cb(struct bufferevent *bev, short event, void *arg) {
ConnectStat *stat = (ConnectStat *)arg;
if (event & BEV_EVENT_EOF) {
printf("connect cloase
");
} else if (event & BEV_EVENT_ERROR) {
printf("some other error
");
}
// 自动close套接字和free读写缓冲区
bufferevent_free(bev);// 释放bufferevent对象
free(stat);
}
typedef struct sockaddr SA;
int tcp_server_init(int port, int listen_num) {
int errno_save;
evutil_socket_t listener; // int listener
listener = socket(AF_INET, SOCK_STREAM, 0);
if (-1 == listener) {
return -1;
}
// 允许多次绑定同一个地址，要用在socket和bind之间
evutil_make_listen_socket_reuseable(listener);
struct sockaddr_in sin;
sin.sin_family = AF_INET;
sin.sin_addr.s_addr = 0;
sin.sin_port = htons(port);
if (bind(listener, (SA *)&sin, sizeof(sin)) < 0) {
errno_save = errno;
evutil_closesocket(listener);
errno = errno_save;
return -1;
}
if (listen(listener, listen_num) < 0) {
errno_save = errno;
evutil_closesocket(listener);
errno = errno_save;
return -1;
}
// 跨平台统一接口，将套接字设置为非阻塞状态
evutil_make_socket_nonblocking(listener);
return listener;
}
// 一个客户端连接上服务器此函数就会被调用；当此函数被调用时，libevent已经帮我们accept了这个客户端
void listener_cb(struct evconnlistener *listener, evutil_socket_t fd, struct sockaddr *sock, int socklen, void *arg) {
printf("accept a client %d
", fd);
struct event_base *base = (struct event_base *)arg;
// 针对已经存在的socket创建bufferevent对象
// BEV_OPT_CLOSE_ON_FREE:如果释放bufferevent对象，则关闭连接
struct bufferevent *bev = bufferevent_socket_new(base, fd, BEV_OPT_CLOSE_ON_FREE);
// BEV_OPT_CLOSE_ON_FREE:释放bufferevent时关闭底层传输端口。这将关闭底层套接字，释放底层bufferevent等。
ConnectStat *stat = stat_init(bev);
// 给bufferevent设置回调函数
// bufferevent对象、读事件回调函数、写事件回调函数、其他事件回调函数、参数
bufferevent_setcb(bev, do_echo_request, do_echo_response, event_cb, stat); // evnet_set
bufferevent_enable(bev, EV_READ | EV_PERSIST); // evnet_add,使bufferevent 生效
}客户端可以使用上面那个去测试
6). libevent监听信号案例
此处监听Ctrl + C 信号
#include <stdio.h>
#include <signal.h>
#include <event.h>
int signal_count = 0;
void signal_handler(evutil_socket_t fd, short events, void *arg) {
printf("收到信号 %d
", fd);
struct event *ev = (struct event *)arg;
signal_count++;
if (signal_count >= 2) {
// 把事件从集合中删除
event_del(ev);
}
}
int main(int argc, char **argv) {
// 创建事件集合
struct event_base *base = event_base_new();
// 创建事件
struct event ev;
// 把事件和信号绑定
event_assign(&ev, base, SIGINT, EV_SIGNAL | EV_PERSIST, signal_handler, &ev); // 监听ctrl+c信号
// 事件添加到集合中
event_add(&ev, NULL);
// 事件集合循环，当集合中没有事件了，此函数结束返回
event_base_dispatch(base);
// 释放集合
event_base_free(base);
return 0;
}
四、信号驱动IO（不介绍）
使用信号驱动I/O时，当网络套接字可读后，内核通过发送SIGIO信号通知应用进程，于是应用可以开 始读取数据。该方式并不是异步I/O，因为实际读取数据到应用进程缓存的工作仍然是由应用自己负责的。
......
五、异步IO（不介绍）
当用户进程发起一个read操作后，内核收到该read操作后，首先它会立刻返回，所以不会对用户进程 阻塞，然后它会等待数据的准备完成，再把数据拷贝到用户内存，完成之后，它会给用户进程发送一个信号，告诉用户进程read操作已完成.
......
六、总结
到此为止，比较常用的IO模式已经介绍完毕，知识点和代码都已经发出来了；
老实说，具体在项目中我也不知道怎么用这些玩意，毕竟没有接触过服务器开发的企业项目；
先把知识点积累下来，日后有机会入职Linux服务器开发岗位的话，也有相应知识支撑！
站长推荐
- SpringSecurity实现前后端分离认证授权
SpringSecurity实现前后端分离认证授权 - stm32使用HAL库配置串口中断收发数据（保姆级教程）
stm32使用HAL库配置串口中断收发数据（保姆级教程） - U8W/U8W-Mini使用与常见问题解决
U8W/U8W-Mini使用与常见问题解决 - 【社区图书馆】伴我前行的一本书《The C Programming Language》
【社区图书馆】伴我前行的一本书《The C Programming Language》 - Unity创建文本时自动修改默认字体
Unity创建文本时自动修改默认字体