C++實(shí)現(xiàn)基于reactor的百萬(wàn)級(jí)并發(fā)服務(wù)器
更新時(shí)間：2025年02月12日 08:37:26 作者：humanGetup
本文介紹了基于Reactor模式的百萬(wàn)級(jí)并發(fā)服務(wù)器,使用epoll進(jìn)行高效I/O多路復(fù)用,支持多個(gè)端口的監(jiān)聽(tīng),并通過(guò)回調(diào)機(jī)制處理每個(gè)連接的接收和發(fā)送操作,需要的朋友可以參考下
一、基于 Reactor 模式的百萬(wàn)級(jí)并發(fā)服務(wù)器是什么？
基于 Reactor 模式的百萬(wàn)級(jí)并發(fā)服務(wù)器 是指一個(gè)能夠高效地處理百萬(wàn)級(jí)并發(fā)連接的服務(wù)器架構(gòu)，它通常使用 Reactor 設(shè)計(jì)模式來(lái)管理大量的客戶端連接。Reactor 模式是一種事件驅(qū)動(dòng)模式，主要用于 I/O 多路復(fù)用，使得服務(wù)器可以在單一線程或少量線程中高效地處理大量并發(fā)連接，避免了傳統(tǒng)的多線程模型中線程開(kāi)銷和上下文切換的性能瓶頸。
二、源碼展示
#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);
}
}
}
}
三、代碼分析
這段代碼是一個(gè)簡(jiǎn)單的基于 epoll 的 I/O 多路復(fù)用網(wǎng)絡(luò)服務(wù)器實(shí)現(xiàn)。它的核心功能是監(jiān)聽(tīng)多個(gè)端口，接受來(lái)自客戶端的連接，并且通過(guò)回調(diào)機(jī)制處理接收到的數(shù)據(jù)和發(fā)送的數(shù)據(jù)。它利用了 epoll 的高效事件驅(qū)動(dòng)模型來(lái)處理多個(gè)并發(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：用于存儲(chǔ)讀取和寫(xiě)入數(shù)據(jù)的緩沖區(qū)大小。CONNECTION_SIZE：最大連接數(shù)。MAX_PORTS：最大監(jiān)聽(tīng)的端口數(shù)。TIME_SUB_MS宏用于計(jì)算兩個(gè)struct timeval類型的時(shí)間差（單位為毫秒）。RCALLBACK定義了一個(gè)函數(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)體用于管理每個(gè)連接的狀態(tài)。它包含了與連接相關(guān)的各種信息，比如讀取緩沖區(qū)、寫(xiě)入緩沖區(qū)、讀取和寫(xiě)入的數(shù)據(jù)長(zhǎng)度、回調(diào)函數(shù)等。- unino r_action是指讀緩沖區(qū)對(duì)應(yīng)的回調(diào)函數(shù)，上面的recallback對(duì)應(yīng)寫(xiě)緩沖區(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）。通過(guò)epoll_ctl系統(tǒng)調(diào)用與epoll文件描述符epfd交互來(lái)管理事件。
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ù)用于為一個(gè)連接（fd）注冊(cè)事件并初始化連接的狀態(tài)(就是注冊(cè)clientfd)。它設(shè)置接收回調(diào)函數(shù)、發(fā)送回調(diào)函數(shù)，以及連接的讀取和寫(xiě)入緩沖區(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。 - 注冊(cè)
clientfd的事件（監(jiān)聽(tīng)EPOLLIN）。 - 打印每次接受連接所花費(fèi)的時(shí)間。
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，如果讀取失敗或客戶端斷開(kāi)連接，則關(guān)閉連接。 - 將接收到的數(shù)據(jù)復(fù)制到
wbuffer，準(zhǔn)備發(fā)送。 - 設(shè)置
EPOLLOUT事件，以便在下一個(gè)事件循環(huán)中處理數(shù)據(jù)發(fā)送（關(guān)注寫(xiě)事件）。
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)建一個(gè) TCP 套接字。
- 將服務(wù)器綁定到指定端口。
- 開(kāi)始監(jiān)聽(tīng)連接。
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)建一個(gè)
epoll實(shí)例。 - 為多個(gè)端口（
port到port + MAX_PORTS）初始化服務(wù)器，并為每個(gè)監(jiān)聽(tīng)套接字注冊(cè)EPOLLIN事件。 - 進(jìn)入一個(gè)無(wú)限循環(huán)，等待和處理事件（通過(guò)
epoll_wait）。
7.總結(jié)：
該程序使用 epoll 進(jìn)行高效的多路復(fù)用網(wǎng)絡(luò) I/O，支持多個(gè)端口的監(jiān)聽(tīng)。它使用回調(diào)機(jī)制來(lái)處理每個(gè)連接的接收和發(fā)送操作。程序?yàn)槊總€(gè)連接分配一個(gè)結(jié)構(gòu)體，管理其緩沖區(qū)和回調(diào)函數(shù)，通過(guò) epoll 處理異步 I/O 操作。
四、常見(jiàn)問(wèn)題
1.默認(rèn)的open files數(shù)量限制為1024
解決方案：
輸入
ulimit -a
可查看open files
可以看到現(xiàn)在最多建立1024個(gè)連接
輸入
ulimit -n 1048576
可修改open files
此時(shí)再輸入
ulimit -a
可以看到：
將服務(wù)端和客戶端的open files都設(shè)置為1048576,這是實(shí)現(xiàn)百萬(wàn)級(jí)并發(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個(gè)，占用端口1024-29,023
解決方案：建立多個(gè)server(提供sport)
對(duì)應(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);
}這個(gè)問(wèn)題解決以前，服務(wù)端代碼是只調(diào)用了一個(gè)端口的
3.系統(tǒng)版本導(dǎo)致的問(wèn)題
這個(gè)版本的ubuntu在處理網(wǎng)絡(luò)高并發(fā)時(shí)存在問(wèn)題
解決方案：
修改配置文件 /etc/sysctl.conf，在這個(gè)文件的結(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í)行四條指令
若輸出如圖，則說(shuō)明問(wèn)題已經(jīng)解決。
記得將服務(wù)端和客戶端都按照以上方法配置
五、百萬(wàn)級(jí)并發(fā)結(jié)果展示
總結(jié)
本文基于reactor設(shè)計(jì)模式，實(shí)現(xiàn)了服務(wù)器百萬(wàn)級(jí)并發(fā)
以上就是C++實(shí)現(xiàn)基于reactor的百萬(wàn)級(jí)并發(fā)服務(wù)器的詳細(xì)內(nèi)容，更多關(guān)于C++ reactor并發(fā)服務(wù)器的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
C++ 標(biāo)準(zhǔn)庫(kù)中的 ＜algorithm＞ 頭文件算法操作總結(jié)
C++ 標(biāo)準(zhǔn)庫(kù)中的 <algorithm> 頭文件提供了大量有用的算法,主要用于操作容器（如 vector, list, array 等）,這些算法通常通過(guò)迭代器來(lái)操作容器元素,本文給大家介紹C++ 標(biāo)準(zhǔn)庫(kù)中的 ＜algorithm＞ 頭文件算法總結(jié),感興趣的朋友一起看看吧2025-04-04
這篇文章主要介紹了C語(yǔ)言魔塔游戲的實(shí)現(xiàn)代碼，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2020-02-02
OpenCV使用稀疏光流實(shí)現(xiàn)視頻對(duì)象跟蹤的方法詳解
這篇文章主要為大家詳細(xì)介紹了OpenCV如何使用稀疏光流實(shí)現(xiàn)視頻對(duì)象跟蹤功能，文中的示例代碼講解詳細(xì)，具有一定的借鑒價(jià)值，需要的可以參考一下2023-02-02
access檢查用戶對(duì)一個(gè)文件的權(quán)限情況，根據(jù)mode的值檢查調(diào)用進(jìn)程對(duì)文件pathname是否具有讀、寫(xiě)、或執(zhí)行的權(quán)限2013-08-08
高效實(shí)現(xiàn)整型數(shù)字轉(zhuǎn)字符串int2str的方法
下面小編就為大家?guī)?lái)一篇高效實(shí)現(xiàn)整型數(shù)字轉(zhuǎn)字符串int2str的方法。小編覺(jué)得挺不錯(cuò)的，現(xiàn)在就分享給大家，也給大家做個(gè)參考。一起跟隨小編過(guò)來(lái)看看吧2017-03-03
OpenMP?Parallel?Construct的實(shí)現(xiàn)原理詳解
在本篇文章當(dāng)中我們將主要分析?OpenMP?當(dāng)中的?parallel?construct?具體時(shí)如何實(shí)現(xiàn)的，以及這個(gè)?construct?調(diào)用了哪些運(yùn)行時(shí)庫(kù)函數(shù)，并且詳細(xì)分析這期間的參數(shù)傳遞,需要的可以參考一下2023-01-01
流是一種抽象概念，它代表了數(shù)據(jù)的無(wú)結(jié)構(gòu)化傳遞。而用來(lái)進(jìn)行輸入輸出操作的流就稱為IO流。這篇文章主要為大家介紹了C++中IO流的使用詳解，需要的朋友可以參考一下2021-12-12
c++使用regex報(bào)錯(cuò)regex_error兩種解決方案
C++正則表達(dá)式是一個(gè)非常強(qiáng)大和實(shí)用的工具,但是使用它們時(shí)需要注意仔細(xì)檢查代碼是否符合語(yǔ)法規(guī)則,這篇文章主要給大家介紹了關(guān)于c++使用regex報(bào)錯(cuò)regex_error的兩種解決方案,需要的朋友可以參考下2024-03-03
C++實(shí)現(xiàn)順序排序算法簡(jiǎn)單示例代碼
這篇文章主要介紹了C++實(shí)現(xiàn)順序排序算法簡(jiǎn)單示例代碼,對(duì)于學(xué)過(guò)C++的朋友一定不會(huì)陌生,現(xiàn)在重溫一下這個(gè)算法,需要的朋友可以參考下2014-08-08
C語(yǔ)言實(shí)現(xiàn)三子棋簡(jiǎn)單小游戲
這篇文章主要為大家詳細(xì)介紹了C語(yǔ)言實(shí)現(xiàn)三子棋簡(jiǎn)單小游戲，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2021-09-09