脚本之家 服务器常用软件
快捷导航
您的位置：首页 → 软件编程 → C 语言 → C++ reactor并发服务器

C++实现基于reactor的百万级并发服务器

 更新时间：2025年02月12日 08:37:26   作者：humanGetup  
本文介绍了基于Reactor模式的百万级并发服务器,使用epoll进行高效I/O多路复用,支持多个端口的监听,并通过回调机制处理每个连接的接收和发送操作,需要的朋友可以参考下

一、基于 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(&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 多路复用网络服务器实现。它的核心功能是监听多个端口，接受来自客户端的连接，并且通过回调机制处理接收到的数据和发送的数据。它利用了 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 --> %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 的事件（监听 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_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：该函数处理数据发送。

  • 从 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 = 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 函数执行以下操作：

  • 创建一个 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 = 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 保存并退出

再按照下图执行四条指令

若输出如图，则说明问题已经解决。

记得将服务端和客户端都按照以上方法配置

五、百万级并发结果展示

总结

本文基于reactor设计模式，实现了服务器百万级并发

以上就是C++实现基于reactor的百万级并发服务器的详细内容，更多关于C++ reactor并发服务器的资料请关注脚本之家其它相关文章！

您可能感兴趣的文章:

相关文章

  • C++ 标准库中的 ＜algorithm＞ 头文件算法操作总结

    C++ 标准库中的 ＜algorithm＞ 头文件算法操作总结

    C++ 标准库中的 <algorithm> 头文件提供了大量有用的算法,主要用于操作容器（如 vector, list, array 等）,这些算法通常通过迭代器来操作容器元素,本文给大家介绍C++ 标准库中的 ＜algorithm＞ 头文件算法总结,感兴趣的朋友一起看看吧
    2025-04-04
  • C语言魔塔游戏的实现代码

    C语言魔塔游戏的实现代码

    这篇文章主要介绍了C语言魔塔游戏的实现代码，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
    2020-02-02
  • OpenCV使用稀疏光流实现视频对象跟踪的方法详解

    OpenCV使用稀疏光流实现视频对象跟踪的方法详解

    这篇文章主要为大家详细介绍了OpenCV如何使用稀疏光流实现视频对象跟踪功能，文中的示例代码讲解详细，具有一定的借鉴价值，需要的可以参考一下
    2023-02-02
  • linux下access函数的用法介绍

    linux下access函数的用法介绍

    access检查用户对一个文件的权限情况，根据mode的值检查调用进程对文件pathname是否具有读、写、或执行的权限
    2013-08-08
  • 高效实现整型数字转字符串int2str的方法

    高效实现整型数字转字符串int2str的方法

    下面小编就为大家带来一篇高效实现整型数字转字符串int2str的方法。小编觉得挺不错的，现在就分享给大家，也给大家做个参考。一起跟随小编过来看看吧
    2017-03-03
  • OpenMP Parallel Construct的实现原理详解

    OpenMP Parallel Construct的实现原理详解

    在本篇文章当中我们将主要分析 OpenMP 当中的 parallel construct 具体时如何实现的，以及这个 construct 调用了哪些运行时库函数，并且详细分析这期间的参数传递,需要的可以参考一下
    2023-01-01
  • C++学习之IO流(输入输出流)详解

    C++学习之IO流(输入输出流)详解

    流是一种抽象概念，它代表了数据的无结构化传递。而用来进行输入输出操作的流就称为IO流。这篇文章主要为大家介绍了C++中IO流的使用详解，需要的朋友可以参考一下
    2021-12-12
  • c++使用regex报错regex_error两种解决方案

    c++使用regex报错regex_error两种解决方案

    C++正则表达式是一个非常强大和实用的工具,但是使用它们时需要注意仔细检查代码是否符合语法规则,这篇文章主要给大家介绍了关于c++使用regex报错regex_error的两种解决方案,需要的朋友可以参考下
    2024-03-03
  • C++实现顺序排序算法简单示例代码

    C++实现顺序排序算法简单示例代码

    这篇文章主要介绍了C++实现顺序排序算法简单示例代码,对于学过C++的朋友一定不会陌生,现在重温一下这个算法,需要的朋友可以参考下
    2014-08-08
  • C语言实现三子棋简单小游戏

    C语言实现三子棋简单小游戏

    这篇文章主要为大家详细介绍了C语言实现三子棋简单小游戏，文中示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
    2021-09-09

最新评论