欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

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

 更新時間：2025年02月12日 08:37:26   作者：humanGetup  
本文介紹了基于Reactor模式的百萬級并發(fā)服務(wù)器,使用epoll進行高效I/O多路復用,支持多個端口的監(jiān)聽,并通過回調(diào)機制處理每個連接的接收和發(fā)送操作,需要的朋友可以參考下

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

基于 Reactor 模式的百萬級并發(fā)服務(wù)器 是指一個能夠高效地處理百萬級并發(fā)連接的服務(wù)器架構(gòu)，它通常使用 Reactor 設(shè)計模式來管理大量的客戶端連接。Reactor 模式是一種事件驅(qū)動模式，主要用于 I/O 多路復用，使得服務(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做下標
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 多路復用網(wǎng)絡(luò)服務(wù)器實現(xiàn)。它的核心功能是監(jiān)聽多個端口，接受來自客戶端的連接，并且通過回調(diào)機制處理接收到的數(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ū)對應的回調(diào)函數(shù)，上面的recallback對應寫緩沖區(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(&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：該函數(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ù)復制到 wbuffer，準備發(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 事件。
  • 進入一個無限循環(huán)，等待和處理事件（通過 epoll_wait）。

7.總結(jié)：

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

四、常見問題

1.默認的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（遠程ip），源端口（本機端口），目的端口（遠程端口），協(xié)議

eg:

192.168.127.128sip

192.168.127.129dip

建立連接27999個，占用端口1024-29,023

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

對應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)版本導致的問題

這個版本的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

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

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

若輸出如圖，則說明問題已經(jīng)解決。

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

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

總結(jié)

本文基于reactor設(shè)計模式，實現(xiàn)了服務(wù)器百萬級并發(fā)

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

相關(guān)文章

  • C++ 標準庫中的 ＜algorithm＞ 頭文件算法操作總結(jié)

    C++ 標準庫中的 ＜algorithm＞ 頭文件算法操作總結(jié)

    C++ 標準庫中的 <algorithm> 頭文件提供了大量有用的算法,主要用于操作容器（如 vector, list, array 等）,這些算法通常通過迭代器來操作容器元素,本文給大家介紹C++ 標準庫中的 ＜algorithm＞ 頭文件算法總結(jié),感興趣的朋友一起看看吧
    2025-04-04
  • C語言魔塔游戲的實現(xiàn)代碼

    C語言魔塔游戲的實現(xiàn)代碼

    這篇文章主要介紹了C語言魔塔游戲的實現(xiàn)代碼，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧
    2020-02-02
  • OpenCV使用稀疏光流實現(xiàn)視頻對象跟蹤的方法詳解

    OpenCV使用稀疏光流實現(xiàn)視頻對象跟蹤的方法詳解

    這篇文章主要為大家詳細介紹了OpenCV如何使用稀疏光流實現(xiàn)視頻對象跟蹤功能，文中的示例代碼講解詳細，具有一定的借鑒價值，需要的可以參考一下
    2023-02-02
  • linux下access函數(shù)的用法介紹

    linux下access函數(shù)的用法介紹

    access檢查用戶對一個文件的權(quán)限情況，根據(jù)mode的值檢查調(diào)用進程對文件pathname是否具有讀、寫、或執(zhí)行的權(quán)限
    2013-08-08
  • 高效實現(xiàn)整型數(shù)字轉(zhuǎn)字符串int2str的方法

    高效實現(xiàn)整型數(shù)字轉(zhuǎn)字符串int2str的方法

    下面小編就為大家?guī)硪黄咝崿F(xiàn)整型數(shù)字轉(zhuǎn)字符串int2str的方法。小編覺得挺不錯的，現(xiàn)在就分享給大家，也給大家做個參考。一起跟隨小編過來看看吧
    2017-03-03
  • OpenMP?Parallel?Construct的實現(xiàn)原理詳解

    OpenMP?Parallel?Construct的實現(xiàn)原理詳解

    在本篇文章當中我們將主要分析?OpenMP?當中的?parallel?construct?具體時如何實現(xiàn)的，以及這個?construct?調(diào)用了哪些運行時庫函數(shù)，并且詳細分析這期間的參數(shù)傳遞,需要的可以參考一下
    2023-01-01
  • C++學習之IO流(輸入輸出流)詳解

    C++學習之IO流(輸入輸出流)詳解

    流是一種抽象概念，它代表了數(shù)據(jù)的無結(jié)構(gòu)化傳遞。而用來進行輸入輸出操作的流就稱為IO流。這篇文章主要為大家介紹了C++中IO流的使用詳解，需要的朋友可以參考一下
    2021-12-12
  • c++使用regex報錯regex_error兩種解決方案

    c++使用regex報錯regex_error兩種解決方案

    C++正則表達式是一個非常強大和實用的工具,但是使用它們時需要注意仔細檢查代碼是否符合語法規(guī)則,這篇文章主要給大家介紹了關(guān)于c++使用regex報錯regex_error的兩種解決方案,需要的朋友可以參考下
    2024-03-03
  • C++實現(xiàn)順序排序算法簡單示例代碼

    C++實現(xiàn)順序排序算法簡單示例代碼

    這篇文章主要介紹了C++實現(xiàn)順序排序算法簡單示例代碼,對于學過C++的朋友一定不會陌生,現(xiàn)在重溫一下這個算法,需要的朋友可以參考下
    2014-08-08
  • C語言實現(xiàn)三子棋簡單小游戲

    C語言實現(xiàn)三子棋簡單小游戲

    這篇文章主要為大家詳細介紹了C語言實現(xiàn)三子棋簡單小游戲，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下
    2021-09-09

最新評論