云计算百科
云计算领域专业知识百科平台
当前位置：网硕互联帮助中心 > 服务器百科 > 正文

网络编程(六)：C++实现基于reactor的百万级并发服务器

2025-04-18 分类：服务器百科 阅读(6) 评论(0)

目录

前言

一、基于 Reactor 模式的百万级并发服务器是什么？

二、源码展示

三、代码分析

1.定义常量与结构体

2.set_event 函数

3.event_register 函数

4.连接接收与发送回调函数

5.init_server 函数

6.main 函数

7.总结：

四、常见问题

1.默认的open files数量限制为1024​编辑

解决方案：

2.不能分配地址

解决方案：建立多个server(提供sport)

3.系统版本导致的问题

解决方案：

五、百万级并发结果展示

总结

前言

这篇文章的代码基于本系列上一篇《网络编程(五)》的reactor设计模式，做了一些改进而成。

0voice · GitHub

一、基于 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_LENGTH1024
#define CONNECTION_SIZE 1048576

#define MAX_PORTS20

#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_PORTS20

    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设计模式，实现了服务器百万级并发

    赞(0)
    未经允许不得转载：网硕互联帮助中心 » 网络编程(六)：C++实现基于reactor的百万级并发服务器
    分享到： 更多 (0)
    标签：

    相关推荐

    评论 抢沙发

    评论前必须登录！

     

    热门标签

    服务器(2585)运维(1533)linux(758)网络(316)美国站群服务器(308)香港云服务器(305)租用香港服务器(301)香港服务器(298)香港站群服务器租用(298)租用香港站群服务器(294)香港服务器租用(293)香港站群服务器(289)美国云服务器(286)香港VPS(280)美国服务器租用(279)美国服务器(276)人工智能(186)python(165)java(165)ubuntu(150)windows(138)c++(123)数据库(123)ssh(120)前端(120)docker(118)tcp/ip(114)nginx(90)网络协议(89)开发语言(87)

    置顶推荐

    热门文章

    最新文章