C++实现基于reactor的百万级并发服务器
目录
- 一、基于 Reactor 模式的百万级并发服务器是什么?
- 二、源码展示
- 三、代码分析
- 1.定义常量与结构体
- 2.set_event 函数
- 3.event_register 函数
- 4.连接接收与发送回调函数
- 5.init_server 函数
- 6.main 函数
- 7.总结:
- 四、常见问题
- 1.默认的open files数量限制为1024
- 解决方案:
- 2.不能分配地址
- 解决方案:建立多个server(提供sport)
- 3.系统版本导致的问题
- 五、百万级并发结果展示
- 总结
一、基于 Reactor 模式的百万级并发服务器是什么?
基于 Reactor 模式的百万级并发服务器 是指一个能够高效地处理百万级并发连接的服务器架构,它通常使用 Reactor 设计模式来管理大量的客户端连接。Reactor 模式是一种事件驱动模式,主要用于 I/O 多路复用,使得服务器可以在单一线程或少量线程中高效地处理大量并发连接,避免了传统的多线程模型中线程开销和上下文切换的性能瓶颈。
二、源码展示
#include <errno.h> #include <stdio.h> #include <sys/socket.h> #include <netinet/in.h> #include <string.h> #include <pthread.h> #include <unistd.h> #include <poll.h> #include <sys/epoll.h> #include <errno.h> #include <sys/time.h> #define BUFFER_LENGTH 1024 #define CONNECTION_SIZE 1048576 #define MAX_PORTS 20 #define TIME_SUB_MS(tv1, tv2) ((tv1.tv_sec - tv2.tv_sec) * 1000 + (tv1.tv_usec - tv2.tv_usec) / 1000) typedef int (*RCALLBACK)(int fd); int accept_cb(int fd); int recv_cb(int fd); int send_cb(int fd); int epfd = 0; struct timeval begin; struct conn { int fd; char rbuffer[BUFFER_LENGTH]; int rlength; char wbuffer[BUFFER_LENGTH]; int wlength; RCALLBACK send_callback; union { RCALLBACK recv_callback; RCALLBACK accept_callback; } r_action; }; //fd做下标 struct conn conn_list[CONNECTION_SIZE] = {0}; int set_event(int fd, int event, int flag) { if (flag) { // non-zero add struct epoll_event ev; ev.events = event; ev.data.fd = fd; epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &ev); } else { // zero mod struct epoll_event ev; ev.events = event; ev.data.fd = fd; epoll_ctl(编程客栈epfd, EPOLL_CTL_MOD, fd, &ev); } } int event_register(int fd, int event) { if (fd < 0) return -1; conn_list[fd].fd = fd; conn_list[fd].r_action.recv_callback = recv_cb; conn_list[fd].send_callback = send_cb; memset(conn_list[fd].rbuffer, 0, BUFFER_LENGTH); conn_list[fd].rlength = 0; memset(conn_list[fd].wbuffer, 0, BUFFER_LENGTH); conn_list[fd].wlength = 0; set_event(fd, event, 1); } // listenfd(sockfd) --> EPOLLIN --> accept_cb int accept_cb(int fd) { struct sockaddr_in clientaddr; socklen_t len = sizeof(clientaddr); int clientfd = accept(fd, (struct sockaddr*)&clientaddr, &len); //printf("accept finshed: %d\n", clientfd); if (clientfd < 0) { printf("accept errno: %d --> %s\n", errno, strerror(errno)); return -1; } event_register(clientfd, EPOLLIN); // | EPOLLET if ((clientfd % 1000) == 0) { struct timeval current; gettimeofday(¤t, NULL); int time_used = TIME_SUB_MS(current, begin); memcpy(&begin, ¤t, sizeof(struct timeval)); printf("accept finshed: %d, time_used: %d\n", clientfd, time_used); } return 0; } int recv_cb(int fd) { memset(conn_list[fd].rbuffer, 0, BUFFER_LENGTH ); int count = recv(fd, conn_list[fd].rbuffer, BUFFER_LENGTH, 0); if (count == 0) { // disconnect printf("client disconnect: %d\n", fd); close(fd); epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL); // unfinished return 0; } else if (count < 0) { // printf("count: %d, errno: %d, %s\n", count, errno, strerror(errno)); close(fd); epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL); return 0; } conn_list[fd].rlength = count; //printf("RECV: %s\n", conn_list[fd].rbuffer); // echo conn_list[fd].wlength = conn_list[fd].rlength; memcpy(conn_list[fd].wbuffer, conn_list[fd].rbuffer, conn_list[fd].wlength); set_event(fd, EPOLLOUT, 0); return count; } int send_cb(int fd) { int count = 0; if (conn_list[fd].wlength != 0) { count = send(fd, conn_list[fd].wbuffer, conn_list[fd].wlength, 0); } set_event(fd, EPOLLIN, 0); return count; } int init_server(unsigned short port) { int sockfd = socket(AF_INET, SOCK_STREAM, 0); struct sockaddr_in servaddr; servaddr.sin_family = AF_INET; servaddr.sin_addr.s_addr = htonl(INADDR_ANY); // 0.0.0.0 servaddr.sin_port = htons(port); // 0-1023, if (-1 == bind(sockfd, (struct sockaddr*)&servaddr, sizeof(struct sockaddr)) ) { printf("bind failed: %s\n", strerror(errno)); } listen(sockfd, 10); //printf("listen finshed: %d\n", sockfd); // 3 return sockfd; } int main() { unsigned short port = 2000; epfd = epoll_create(1); int i = 0; for (i = 0;i < MAX_PORTS;i ++) { int sockfd = init_server(port + i); conn_list[sockfd].fd = sockfd; conn_list[sockfd].r_action.recv_callback = accept_cb; set_event(sockfd, EPOLLIN, 1); } gettimeofday(&begin, NULL); while (1) { // mainloop struct epoll_event events[1024] = {0}; int nready = epoll_wait(epfd, events, 1024, -1); int i = 0; for (i = 0;i < nready;i ++) { int connfd = events[i].data.fd; if (events[i].events & EPOLLIN) { conn_list[connfd].r_action.recv_callback(connfd); } if (events[i].events & EPOLLOUT) { conn_list[connfd].send_callback(connfd); } } } }
三、代码分析
这段代码是一个简单的基于 epoll
的 I/O 多路复用网络服务器实现。它的核心功能是监听多个端口,接受来自客户端的连接,并且通过回调机制处理接收到的数据和发送的数据。它利用了 epoll
的高效事件驱动模型来处理多个并发连接。
1.定义常量与结构体
#define BUFFER_LENGTH 1024 #define CONNECTION_SIZE 1048576 #define MAX_PORTS 20 #define TIME_SUB_MS(tv1, tv2) ((tv1.tv_sec - tv2.tv_sec) * 1000 + (tv1.tv_usec - tv2.tv_usec) / 1000) typedef int (*RCALLBACK)(int fd);
BUFFER_LENGTH
:用于存储读取和写入数据的缓冲区大小。CONNECTION_SIZE
:最大连接数。MAX_PORTS
:最大监听的端口数。TIME_SUB_MS
宏用于计算两个struct timeval
类型的时间差(单位为毫秒)。RCALLBACK
定义了一个函数指针类型,表示回调函数。
struct conn { int fd; char rbuffer[BUFFER_LENGTH]; int rlength; char wbuffer[BUFFER_LENGTH]; int wlength; RCALLBACK send_callback; union { RCALLBACK recv_callback; RCALLBACK accept_callback; } r_action; };
conn
结构体用于管理每个连接的状态。它包含了与连接相关的各种信息,比如读取缓冲区、写入缓冲区、读取和写入的数据长度、回调函数等。- unino r_action是指读缓冲区对应的回调函数,上面的recallback对应写缓冲区的回调函数
2.set_event 函数
int set_event(int fd, int event, int flag) { if (flag) { // non-zero add struct epoll_event ev; ev.events = event; ev.data.fd = fd; epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &ev); } else { // zero mod struct epoll_event ev; ev.events = event; ev.data.fd = fd; epoll_ctl(epfd, EPOLL_CTL_MOD, fd, &ev); } }
set_event
函数用于向epoll
添加或修改事件。根据flag
的值,决定是添加事件(EPOLL_CTL_ADD
)还是修改事件(EPOLL_CTL_MOD
)。通过epoll_ctl
系统调用与epoll
文件描述符epfd
交互来管理事件。
3.event_register 函数
int event_register(int fd, int event) { if (fd < 0) return -1; conn_list[fd].fd = fd; conn_list[fd].r_action.recv_callback = recv_cb; conn_list[fd].send_callback = send_cb; memset(conn_list[fd].rbuffer, 0, BUFFER_LENGTH); conn_list[fd].rlength = 0; memset(conn_list[fd].wbuffer, 0, BUFFER_LENGTH); conn_list[fd].wlength = 0; set_event(fd, event, 1); }
event_register
函数用于为一个连接(fd
)注册事件并初始化连接的状态(就是注册clientfd)。它设置接收回调函数、发送回调函数,以及连接的读取和写入缓冲区。
4.连接接收与发送回调函数
int accept_cb(int fd) { struct sockaddr_in clientaddr; socklen_t len = sizeof(clientaddr); int clientfd = accept(fd, (struct sockaddr*)&clientaddr, &len); if (clientfd < 0) { printf("accept errno: %d --> www.devze.com%s\n", errno, strerror(errno)); return -1; } event_register(clientfd, EPOLLIN); if ((clientfd % 1000) == 0) { struct timeval current; gettimeofday(¤t, NULL); int time_used = TIME_SUB_MS(current, javascriptbegin); memcpy(&begin, ¤t, sizeof(struct timeval)); printf("accept finshed: %d, time_used: %d\n", clientfd, time_used); } return 0; }
accept_cb
:该函数处理新的客户端连接。
- 调用
accept
函数接受连接,返回客户端的套接字clientfd
。 - 注册
clientfd
的事件(监听EPOLLIN
)。 - 打印每次接受连接所花费的时间。
int recv_cb(int fd) { memset(conn_list[fd].rbuffer, 0, BUFFER_LENGTH); int count = recv(fd, conn_list[fd].rbuffer, BUFFER_LENGTH, 0); if (count == 0) { printf("client disconnect: %d\n", fd); close(fd); epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL); return 0; } else if (count < 0) { printf("count: %d, errno: %d, %s\n", count, errno, strerror(errno)); close(fd); epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL); return 0; } conn_list[fd].rlength = count; conn_list[fd].wlength = conn_list[fd].rlength; memcpy(conn_list[fd].wbuffer, conn_list[fd].rbuffer, conn_list[fd].wlength); set_event(fd, EPOLLOUT, 0); return count; }
recv_cb
:该函数处理从客户端接收到的数据。
- 读取数据到
rbuffer
,如果读取失败或客户端断开连接,则关闭连接。 - 将接收到的数据复制到
wbuffer
,准备发送。 - 设置
EPOLLOUT
事件,以便在下一个事件循环中处理数据发送(关注写事件)。
int send_HhmVxwfDcb(int fd) { int count = 0; if (conn_list[fd].wlength != 0) { count = send(fd, conn_list[fd].wbuffer, conn_list[fd].wlength, 0); } set_event(fd, EPOLLIN, 0); return count; }
send_cb
:该函数处理数据发送。
- 从
wbuffer
中发送数据到客户端。 - 设置
EPOLLIN
事件,以便处理接收数据(关注读事件)。
5.init_server 函数
int init_server(unsigned short port) { int sockfd = socket(AF_INET, SOCK_STREAM, 0); struct sockaddr_in servaddr; servaddr.sin_family = AF_INET; servaddr.sin_addr.s_addr = htonl(INADDR_ANY); servaddr.sin_port = htons(port); if (-1 == bind(sockfd, (struct sockaddr*)&servaddr, sizeof(struct sockaddr))) { printf("bind failed: %s\n", strerror(errno)); } listen(sockfd, 10); return sockfd; }
init_server
函数用于初始化服务器:
- 创建一个 TCP 套接字。
- 将服务器绑定到指定端口。
- 开始监听连接。
6.main 函数
int main() { unsigned short port = 2000; epfd = epoll_create(1); int i = python0; for (i = 0; i < MAX_PORTS; i++) { int sockfd = init_server(port + i); conn_list[sockfd].fd = sockfd; conn_list[sockfd].r_action.recv_callback = accept_cb; set_event(sockfd, EPOLLIN, 1); } gettimeofday(&begin, NULL); while (1) { struct epoll_event events[1024] = {0}; int nready = epoll_wait(epfd, events, 1024, -1); int i = 0; for (i = 0; i < nready; i++) { int connfd = events[i].data.fd; if (events[i].events & EPOLLIN) { conn_list[connfd].r_action.recv_callback(connfd); } if (events[i].events & EPOLLOUT) { conn_list[connfd].send_callback(connfd); } } } }
main
函数执行以下操作:
- 创建一个
epoll
实例。 - 为多个端口(
port
到port + MAX_PORTS
)初始化服务器,并为每个监听套接字注册EPOLLIN
事件。 - 进入一个无限循环,等待和处理事件(通过
epoll_wait
)。
7.总结:
该程序使用 epoll
进行高效的多路复用网络 I/O,支持多个端口的监听。它使用回调机制来处理每个连接的接收和发送操作。程序为每个连接分配一个结构体,管理其缓冲区和回调函数,通过 epoll
处理异步 I/O 操作。
四、常见问题
1.默认的open files数量限制为1024
解决方案:
输入
ulimit -a
可查看open files
可以看到现在最多建立1024个连接
输入
ulimit -n 1048576
可修改open files
此时再输入
ulimit -a
可以看到:
将服务端和客户端的open files都设置为1048576,这是实现百万级并发的基础
2.不能分配地址
原因是:五元组的数量不够
五元组(sip,dip, sport, dport, proto)源ip(本地ip),目的ip(远程ip),源端口(本机端口),目的端口(远程端口),协议
eg:
192.168.127.128sip
192.168.127.129dip
建立连接27999个,占用端口1024-29,023
解决方案:建立多个server(提供sport)
对应main函数这段代码:
#define MAX_PORTS 20 int i = 0; for (i = 0;i < MAX_PORTS;i ++) { int sockfd = init_server(port + i); conn_list[sockfd].fd = sockfd; conn_list[sockfd].r_action.recv_callback = accept_cb; set_event(sockfd, EPOLLIN, 1); }
这个问题解决以前,服务端代码是只调用了一个端口的
3.系统版本导致的问题
这个版本的Ubuntu在处理网络高并发时存在问题
解决方案:
修改配置文件 /etc/sysctl.conf,在这个文件的结尾加上
net.ipv4.tcp_syn_retries = 5
net.ipv4.tcp_syncookies = 1net.ipv4.tcp_mem = 262144 786432 786432net.ipv4.tcp_wmem = 1024 1024 2048net.ipv4.tcp_rmem = 1024 1024 2048fs.file-max = 1048576net.nf_conntrack_max = 1048576net.netfilter.nf_conntrack_tcp_timeout_established = 1200
linux终端中输入
sudo vim /etc/sysctl.conf
进入配置文件,并将上面的内容输入,然后按 ESC -> ctrl + : -> wq 保存并退出
再按照下图执行四条指令
若输出如图,则说明问题已经解决。
记得将服务端和客户端都按照以上方法配置
五、百万级并发结果展示
总结
本文基于reactor设计模式,实现了服务器百万级并发
以上就是C++实现基于reactor的百万级并发服务器的详细内容,更多关于C++ reactor并发服务器的资料请关注编程客栈(www.devze.com)其它相关文章!
精彩评论