C++實現(xiàn)基于reactor的百萬級并發(fā)服務(wù)器
一、基于 Reactor 模式的百萬級并發(fā)服務(wù)器是什么?
基于 Reactor 模式的百萬級并發(fā)服務(wù)器 是指一個能夠高效地處理百萬級并發(fā)連接的服務(wù)器架構(gòu),它通常使用 Reactor 設(shè)計模式來管理大量的客戶端連接。Reactor 模式是一種事件驅(qū)動模式,主要用于 I/O 多路復(fù)用,使得服務(wù)器可以在單一線程或少量線程中高效地處理大量并發(fā)連接,避免了傳統(tǒng)的多線程模型中線程開銷和上下文切換的性能瓶頸。
二、源碼展示
#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做下標(biāo) 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 多路復(fù)用網(wǎng)絡(luò)服務(wù)器實現(xiàn)。它的核心功能是監(jiān)聽多個端口,接受來自客戶端的連接,并且通過回調(diào)機(jī)制處理接收到的數(shù)據(jù)和發(fā)送的數(shù)據(jù)。它利用了 epoll
的高效事件驅(qū)動模型來處理多個并發(fā)連接。
1.定義常量與結(jié)構(gòu)體
#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
:用于存儲讀取和寫入數(shù)據(jù)的緩沖區(qū)大小。CONNECTION_SIZE
:最大連接數(shù)。MAX_PORTS
:最大監(jiān)聽的端口數(shù)。TIME_SUB_MS
宏用于計算兩個struct timeval
類型的時間差(單位為毫秒)。RCALLBACK
定義了一個函數(shù)指針類型,表示回調(diào)函數(shù)。
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
結(jié)構(gòu)體用于管理每個連接的狀態(tài)。它包含了與連接相關(guān)的各種信息,比如讀取緩沖區(qū)、寫入緩沖區(qū)、讀取和寫入的數(shù)據(jù)長度、回調(diào)函數(shù)等。- unino r_action是指讀緩沖區(qū)對應(yīng)的回調(diào)函數(shù),上面的recallback對應(yīng)寫緩沖區(qū)的回調(diào)函數(shù)
2.set_event 函數(shù)
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
函數(shù)用于向epoll
添加或修改事件。根據(jù)flag
的值,決定是添加事件(EPOLL_CTL_ADD
)還是修改事件(EPOLL_CTL_MOD
)。通過epoll_ctl
系統(tǒng)調(diào)用與epoll
文件描述符epfd
交互來管理事件。
3.event_register 函數(shù)
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
函數(shù)用于為一個連接(fd
)注冊事件并初始化連接的狀態(tài)(就是注冊clientfd)。它設(shè)置接收回調(diào)函數(shù)、發(fā)送回調(diào)函數(shù),以及連接的讀取和寫入緩沖區(qū)。
4.連接接收與發(fā)送回調(diào)函數(shù)
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 --> %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, begin); memcpy(&begin, ¤t, sizeof(struct timeval)); printf("accept finshed: %d, time_used: %d\n", clientfd, time_used); } return 0; }
accept_cb
:該函數(shù)處理新的客戶端連接。
- 調(diào)用
accept
函數(shù)接受連接,返回客戶端的套接字clientfd
。 - 注冊
clientfd
的事件(監(jiān)聽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
:該函數(shù)處理從客戶端接收到的數(shù)據(jù)。
- 讀取數(shù)據(jù)到
rbuffer
,如果讀取失敗或客戶端斷開連接,則關(guān)閉連接。 - 將接收到的數(shù)據(jù)復(fù)制到
wbuffer
,準(zhǔn)備發(fā)送。 - 設(shè)置
EPOLLOUT
事件,以便在下一個事件循環(huán)中處理數(shù)據(jù)發(fā)送(關(guān)注寫事件)。
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; }
send_cb
:該函數(shù)處理數(shù)據(jù)發(fā)送。
- 從
wbuffer
中發(fā)送數(shù)據(jù)到客戶端。 - 設(shè)置
EPOLLIN
事件,以便處理接收數(shù)據(jù)(關(guān)注讀事件)。
5.init_server 函數(shù)
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
函數(shù)用于初始化服務(wù)器:
- 創(chuàng)建一個 TCP 套接字。
- 將服務(wù)器綁定到指定端口。
- 開始監(jiān)聽連接。
6.main 函數(shù)
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) { 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
函數(shù)執(zhí)行以下操作:
- 創(chuàng)建一個
epoll
實例。 - 為多個端口(
port
到port + MAX_PORTS
)初始化服務(wù)器,并為每個監(jiān)聽套接字注冊EPOLLIN
事件。 - 進(jìn)入一個無限循環(huán),等待和處理事件(通過
epoll_wait
)。
7.總結(jié):
該程序使用 epoll
進(jìn)行高效的多路復(fù)用網(wǎng)絡(luò) I/O,支持多個端口的監(jiān)聽。它使用回調(diào)機(jī)制來處理每個連接的接收和發(fā)送操作。程序為每個連接分配一個結(jié)構(gòu)體,管理其緩沖區(qū)和回調(diào)函數(shù),通過 epoll
處理異步 I/O 操作。
四、常見問題
1.默認(rèn)的open files數(shù)量限制為1024
解決方案:
輸入
ulimit -a
可查看open files
可以看到現(xiàn)在最多建立1024個連接
輸入
ulimit -n 1048576
可修改open files
此時再輸入
ulimit -a
可以看到:
將服務(wù)端和客戶端的open files都設(shè)置為1048576,這是實現(xiàn)百萬級并發(fā)的基礎(chǔ)
2.不能分配地址
原因是:五元組的數(shù)量不夠
五元組(sip,dip, sport, dport, proto)源ip(本地ip),目的ip(遠(yuǎn)程ip),源端口(本機(jī)端口),目的端口(遠(yuǎn)程端口),協(xié)議
eg:
192.168.127.128sip
192.168.127.129dip
建立連接27999個,占用端口1024-29,023
解決方案:建立多個server(提供sport)
對應(yīng)main函數(shù)這段代碼:
#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); }
這個問題解決以前,服務(wù)端代碼是只調(diào)用了一個端口的
3.系統(tǒng)版本導(dǎo)致的問題
這個版本的ubuntu在處理網(wǎng)絡(luò)高并發(fā)時存在問題
解決方案:
修改配置文件 /etc/sysctl.conf,在這個文件的結(jié)尾加上
net.ipv4.tcp_syn_retries = 5
net.ipv4.tcp_syncookies = 1
net.ipv4.tcp_mem = 262144 786432 786432
net.ipv4.tcp_wmem = 1024 1024 2048
net.ipv4.tcp_rmem = 1024 1024 2048
fs.file-max = 1048576
net.nf_conntrack_max = 1048576
net.netfilter.nf_conntrack_tcp_timeout_established = 1200
Linux終端中輸入
sudo vim /etc/sysctl.conf
進(jìn)入配置文件,并將上面的內(nèi)容輸入,然后按 ESC -> ctrl + : -> wq 保存并退出
再按照下圖執(zhí)行四條指令
若輸出如圖,則說明問題已經(jīng)解決。
記得將服務(wù)端和客戶端都按照以上方法配置
五、百萬級并發(fā)結(jié)果展示
總結(jié)
本文基于reactor設(shè)計模式,實現(xiàn)了服務(wù)器百萬級并發(fā)
以上就是C++實現(xiàn)基于reactor的百萬級并發(fā)服務(wù)器的詳細(xì)內(nèi)容,更多關(guān)于C++ reactor并發(fā)服務(wù)器的資料請關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
C++ 標(biāo)準(zhǔn)庫中的 <algorithm> 頭文件算法操作總結(jié)
C++ 標(biāo)準(zhǔn)庫中的 <algorithm> 頭文件提供了大量有用的算法,主要用于操作容器(如 vector, list, array 等),這些算法通常通過迭代器來操作容器元素,本文給大家介紹C++ 標(biāo)準(zhǔn)庫中的 <algorithm> 頭文件算法總結(jié),感興趣的朋友一起看看吧2025-04-04OpenCV使用稀疏光流實現(xiàn)視頻對象跟蹤的方法詳解
這篇文章主要為大家詳細(xì)介紹了OpenCV如何使用稀疏光流實現(xiàn)視頻對象跟蹤功能,文中的示例代碼講解詳細(xì),具有一定的借鑒價值,需要的可以參考一下2023-02-02高效實現(xiàn)整型數(shù)字轉(zhuǎn)字符串int2str的方法
下面小編就為大家?guī)硪黄咝崿F(xiàn)整型數(shù)字轉(zhuǎn)字符串int2str的方法。小編覺得挺不錯的,現(xiàn)在就分享給大家,也給大家做個參考。一起跟隨小編過來看看吧2017-03-03OpenMP?Parallel?Construct的實現(xiàn)原理詳解
在本篇文章當(dāng)中我們將主要分析?OpenMP?當(dāng)中的?parallel?construct?具體時如何實現(xiàn)的,以及這個?construct?調(diào)用了哪些運行時庫函數(shù),并且詳細(xì)分析這期間的參數(shù)傳遞,需要的可以參考一下2023-01-01