C++實現基于reactor的百萬級并發(fā)服務器

更新時間：2025年02月12日 08:37:26 作者：humanGetup

本文介紹了基于Reactor模式的百萬級并發(fā)服務器,使用epoll進行高效I/O多路復用,支持多個端口的監(jiān)聽,并通過回調機制處理每個連接的接收和發(fā)送操作,需要的朋友可以參考下

一、基于 Reactor 模式的百萬級并發(fā)服務器是什么？

基于 Reactor 模式的百萬級并發(fā)服務器 是指一個能夠高效地處理百萬級并發(fā)連接的服務器架構，它通常使用 Reactor 設計模式來管理大量的客戶端連接。Reactor 模式是一種事件驅動模式，主要用于 I/O 多路復用，使得服務器可以在單一線程或少量線程中高效地處理大量并發(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做下標
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(&current, NULL);	
		
		int time_used = TIME_SUB_MS(current, begin);		
		memcpy(&begin, &current, 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 多路復用網絡服務器實現。它的核心功能是監(jiān)聽多個端口，接受來自客戶端的連接，并且通過回調機制處理接收到的數據和發(fā)送的數據。它利用了 epoll 的高效事件驅動模型來處理多個并發(fā)連接。

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：用于存儲讀取和寫入數據的緩沖區(qū)大小。
CONNECTION_SIZE：最大連接數。
MAX_PORTS：最大監(jiān)聽的端口數。
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 結構體用于管理每個連接的狀態(tài)。它包含了與連接相關的各種信息，比如讀取緩沖區(qū)、寫入緩沖區(qū)、讀取和寫入的數據長度、回調函數等。
unino r_action是指讀緩沖區(qū)對應的回調函數，上面的recallback對應寫緩沖區(qū)的回調函數

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 系統(tǒng)調用與 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）注冊事件并初始化連接的狀態(tài)(就是注冊clientfd)。它設置接收回調函數、發(fā)送回調函數，以及連接的讀取和寫入緩沖區(qū)。

4.連接接收與發(fā)送回調函數

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(&current, NULL);
        int time_used = TIME_SUB_MS(current, begin);
        memcpy(&begin, &current, sizeof(struct timeval));
        printf("accept finshed: %d, time_used: %d\n", clientfd, time_used);
    }
    return 0;
}

accept_cb：該函數處理新的客戶端連接。

調用 accept 函數接受連接，返回客戶端的套接字 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：該函數處理從客戶端接收到的數據。

讀取數據到 rbuffer，如果讀取失敗或客戶端斷開連接，則關閉連接。
將接收到的數據復制到 wbuffer，準備發(fā)送。
設置 EPOLLOUT 事件，以便在下一個事件循環(huán)中處理數據發(fā)送（關注寫事件）。

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：該函數處理數據發(fā)送。

從 wbuffer 中發(fā)送數據到客戶端。
設置 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 函數用于初始化服務器：

創(chuàng)建一個 TCP 套接字。
將服務器綁定到指定端口。
開始監(jiān)聽連接。

6.main 函數

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 函數執(zhí)行以下操作：

創(chuàng)建一個 epoll 實例。
為多個端口（port 到 port + MAX_PORTS）初始化服務器，并為每個監(jiān)聽套接字注冊 EPOLLIN 事件。
進入一個無限循環(huán)，等待和處理事件（通過 epoll_wait）。

7.總結：

該程序使用 epoll 進行高效的多路復用網絡 I/O，支持多個端口的監(jiān)聽。它使用回調機制來處理每個連接的接收和發(fā)送操作。程序為每個連接分配一個結構體，管理其緩沖區(qū)和回調函數，通過 epoll 處理異步 I/O 操作。

四、常見問題

1.默認的open files數量限制為1024

解決方案：

輸入

ulimit -a

可查看open files

可以看到現在最多建立1024個連接

輸入

ulimit -n 1048576

可修改open files

此時再輸入

ulimit -a

可以看到：

將服務端和客戶端的open files都設置為1048576,這是實現百萬級并發(fā)的基礎

2.不能分配地址

原因是：五元組的數量不夠

五元組(sip,dip, sport, dport, proto)源ip（本地ip），目的ip（遠程ip），源端口（本機端口），目的端口（遠程端口），協(xié)議

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.系統(tǒng)版本導致的問題

這個版本的ubuntu在處理網絡高并發(fā)時存在問題

解決方案：

修改配置文件 /etc/sysctl.conf，在這個文件的結尾加上

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

進入配置文件，并將上面的內容輸入，然后按 ESC -> ctrl + : -> wq 保存并退出

再按照下圖執(zhí)行四條指令

若輸出如圖，則說明問題已經解決。

記得將服務端和客戶端都按照以上方法配置

五、百萬級并發(fā)結果展示

總結

本文基于reactor設計模式，實現了服務器百萬級并發(fā)

以上就是C++實現基于reactor的百萬級并發(fā)服務器的詳細內容，更多關于C++ reactor并發(fā)服務器的資料請關注腳本之家其它相關文章！

您可能感興趣的文章:

C++ 標準庫中的＜algorithm＞頭文件算法操作總結
C++ 標準庫中的 <algorithm> 頭文件提供了大量有用的算法,主要用于操作容器（如 vector, list, array 等）,這些算法通常通過迭代器來操作容器元素,本文給大家介紹C++ 標準庫中的＜algorithm＞頭文件算法總結,感興趣的朋友一起看看吧
2025-04-04
C語言魔塔游戲的實現代碼
這篇文章主要介紹了C語言魔塔游戲的實現代碼，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧
2020-02-02
OpenCV使用稀疏光流實現視頻對象跟蹤的方法詳解
這篇文章主要為大家詳細介紹了OpenCV如何使用稀疏光流實現視頻對象跟蹤功能，文中的示例代碼講解詳細，具有一定的借鑒價值，需要的可以參考一下
2023-02-02
linux下access函數的用法介紹
access檢查用戶對一個文件的權限情況，根據mode的值檢查調用進程對文件pathname是否具有讀、寫、或執(zhí)行的權限
2013-08-08
高效實現整型數字轉字符串int2str的方法
下面小編就為大家?guī)硪黄咝崿F整型數字轉字符串int2str的方法。小編覺得挺不錯的，現在就分享給大家，也給大家做個參考。一起跟隨小編過來看看吧
2017-03-03
OpenMP?Parallel?Construct的實現原理詳解
在本篇文章當中我們將主要分析?OpenMP?當中的?parallel?construct?具體時如何實現的，以及這個?construct?調用了哪些運行時庫函數，并且詳細分析這期間的參數傳遞,需要的可以參考一下
2023-01-01
C++學習之IO流(輸入輸出流)詳解
流是一種抽象概念，它代表了數據的無結構化傳遞。而用來進行輸入輸出操作的流就稱為IO流。這篇文章主要為大家介紹了C++中IO流的使用詳解，需要的朋友可以參考一下
2021-12-12
c++使用regex報錯regex_error兩種解決方案
C++正則表達式是一個非常強大和實用的工具,但是使用它們時需要注意仔細檢查代碼是否符合語法規(guī)則,這篇文章主要給大家介紹了關于c++使用regex報錯regex_error的兩種解決方案,需要的朋友可以參考下
2024-03-03
C++實現順序排序算法簡單示例代碼
這篇文章主要介紹了C++實現順序排序算法簡單示例代碼,對于學過C++的朋友一定不會陌生,現在重溫一下這個算法,需要的朋友可以參考下
2014-08-08
C語言實現三子棋簡單小游戲
這篇文章主要為大家詳細介紹了C語言實現三子棋簡單小游戲，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下
2021-09-09

亚洲乱码中文字幕综合,中国熟女仑乱hd,亚洲精品乱拍国产一区二区三区,一本大道卡一卡二卡三乱码全集资源,又粗又黄又硬又爽的免费视频

C++實現基于reactor的百萬級并發(fā)服務器

目錄

一、基于 Reactor 模式的百萬級并發(fā)服務器是什么？

二、源碼展示

三、代碼分析

1.定義常量與結構體

2.set_event 函數

3.event_register 函數

4.連接接收與發(fā)送回調函數

5.init_server 函數

6.main 函數

7.總結：

四、常見問題

1.默認的open files數量限制為1024

解決方案：

2.不能分配地址

解決方案：建立多個server(提供sport)

3.系統(tǒng)版本導致的問題

五、百萬級并發(fā)結果展示

總結

相關文章

最新評論

大家感興趣的內容

最近更新的內容

常用在線小工具

亚洲乱码中文字幕综合,中国熟女仑乱hd,亚洲精品乱拍国产一区二区三区,一本大道卡一卡二卡三乱码全集资源,又粗又黄又硬又爽的免费视频

C++實現基于reactor的百萬級并發(fā)服務器

目錄

一、基于 Reactor 模式的百萬級并發(fā)服務器是什么？

二、源碼展示

三、代碼分析

1.定義常量與結構體

2.set_event 函數

3.event_register 函數

4.連接接收與發(fā)送回調函數

5.init_server 函數

6.main 函數

7.總結：

四、常見問題

1.默認的open files數量限制為1024

解決方案：

2.不能分配地址

解決方案：建立多個server(提供sport)

3.系統(tǒng)版本導致的問題

五、百萬級并發(fā)結果展示

總結

相關文章

最新評論

大家感興趣的內容

最近更新的內容

常用在線小工具

一、基于 Reactor 模式的百萬級并發(fā)服務器是什么？

三、代碼分析

四、常見問題

五、百萬級并發(fā)結果展示