开发者

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

开发者 https://www.devze.com 2025-04-10 14:52 出处：网络 作者： humanGetup
目录一、基于 Reactor 模式的百万级并发服务器是什么？二、源码展示三、代码分析1.定义常量与结构体2.set_event 函数3.event_register 函数4.连接接收与发送回调函数5.init_server 函数6.main 函数7.总结：
相关专题：C++ reactor并发服务器C++ reactor百万级并发服务器C++服务器
目录
  • 一、基于 Reactor 模式的百万级并发服务器是什么？
  • 二、源码展示
  • 三、代码分析
    • 1.定义常量与结构体
    • 2.set_event 函数
    • 3.event_register 函数
    • 4.连接接收与发送回调函数
    • 5.init_server 函数
    • 6.main 函数
    • 7.总结：
  • 四、常见问题
    • 1.默认的open files数量限制为1024
      • 解决方案：
    • 2.不能分配地址
      • 解决方案：建立多个server(提供sport)
    • 3.系统版本导致的问题
    • 五、百万级并发结果展示
      • 总结

        一、基于 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 --> www.devze.com%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, javascriptbegin);
        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_HhmVxwfDcb(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 = python0;
    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

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

        解决方案：

        输入

        ulimit -a

        可查看open files

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

        可以看到现在最多建立1024个连接

        输入

        ulimit -n 1048576

        可修改open files

        此时再输入

        ulimit -a

        可以看到：

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

        将服务端和客户端的open files都设置为1048576,这是实现百万级并发的基础

        2.不能分配地址

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

        原因是：五元组的数量不够

        五元组(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.系统版本导致的问题

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

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

        这个版本的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 保存并退出

        再按照下图执行四条指令

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

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

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

        五、百万级并发结果展示

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

        总结

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

        以上就是C++实现基于reactor的百万级并发服务器的详细内容，更多关于C++ reactor并发服务器的资料请关注编程客栈(www.devze.com)其它相关文章！

        0

        上一篇:

        :下一篇

        更多 开发 相关资讯：

        精彩评论

        暂无评论...
        登录 注册
        Qedev.com
        验证码 换一张
        取 消

        关注公众号

        热门标签

        图文推荐

        Java中如何正确的停掉线程

        Java中如何正确的停掉线程
        Spring Cache 整合 Redis 实现高效缓存的方法

        Spring Cache 整合 Redis 实现高效缓存的方法
        QT Creator配置Kit的实现示例

        QT Creator配置Kit的实现示例
        从基础到高级详解Python文本分词的完全指南

        从基础到高级详解Python文本分词的完全指南
        从基础到高级详解Python数值格式化输出的完全指南

        从基础到高级详解Python数值格式化输出的完全指南
        Go语言sync锁与对象池的实现

        Go语言sync锁与对象池的实现
        Go语言ants协程池的具体使用

        Go语言ants协程池的具体使用
        Spring Boot中处理Servlet路径映射问题解决

        Spring Boot中处理Servlet路径映射问题解决

        开发者开发者网给大家分享系统运维,大数据运维,云计算,编程开发技巧,路由交换,运维和开发相关的资讯及技术文章，同时StackOverflow中文社区，知识经验交流分享。

        法律声明：本站内容均为网友上传，网站举办方负责审核和监督，如存在版权或非法内容，欢迎举报，我们将尽快予以删除。邮箱：devze@qq.com

        Copyright © 2018-2020 开发者. All rights reserved. Powered by 开发者ICP备案号: 京ICP备10032868号-9