reactor实现服务器百万并发

提示：文章写完后，目录可以自动生成，如何生成可参考右边的帮助文档

文章目录

• 前言
• 1. epoll 和 reactor的区别
• 2. io、事件和回调函数的关系
• 3. 编译过程中遇到的问题
• 4. 代码思路
• 5. 实现代码
• 6. 实现效果

前言

相对于select，poll，epoll，reactor 更加关注具体的事件，select，poll，epoll 通知内核哪些 io 有事件，而 reactor 针对不同的事件执行不同的回调函数，做到把读写事件分开，读事件就执行相应的recv 回调函数，写事件就执行相应的 send 回调函数，可实现更好关注io事件的处理。本文探讨基于 epoll 实现的 reactor。

1. epoll 和 reactor的区别

a. epoll 主要关注网络 io 的管理；而reactor更关注io事件的处理。 b. epoll执行时，数据一次性接收不完时，后面的数据没有地方放，不适合大包和多包处理；而reactor 把读写数据分别放在 rbuffer 和 wbuffer，通过 rlength 和 wlength 依次存放之后的数据，并且可以使得io 同时读和写。

2. io、事件和回调函数的关系

io主要分为 litsenfd 和 clientfd，而 listenfd 对应事件是 EPOLLIN，其对应的回调函数是accept_cb，clientfd对应事件主要分为 EPOLLIN 和 EPOLLOUT，分别对应回调函数 recv_cb 和 send_cb。

3. 编译过程中遇到的问题

(1) 编译时内存不够

解决办法：gcc编译参数加上 -mcmodel=medium。

(2) 编译时遇到 too many open files

解决办法1：使用ulimit -a 看open files数量，通过ulimit -n 数字改其大小，不是永久修改，重启虚拟机后消失。

解决方法2：通过sudo vim /etc/security/limits.conf在文件末尾添加下图内容。保存后重启虚拟机即可。修改配置文件，涉及写磁盘，每一次重启虚拟机会重新加载，永久修改。

(3) 端口数量不够

解决方法：sudo vim /etc/sysctl.conf命令加上以下内容（增加了端口数量以及限制），保存完之后通过sudo sysctl -p命令运行。

(4) 内核模块没有加载（net.nf_conntrack）没有加载。

解决方法：sudo modprobe nf_conntrack命令加载模块（modeprobe智能地在内核中加载模块或者一处模块）

(5) 不能分配请求地址，由于五元组不够（源端口号、目的端口号、协议号、源IP、目的IP）

解决方法：增加五元，建立多个ip（sever），建立多个端口（通过以下循环）

int i = 0;

for (i ; i < MAX_PORTS ; i++) {//20个sever

int sockfd = init_server(port + i);

conn_list[sockfd].fd = sockfd; //regist 一个fd
conn_list[sockfd].r_action.recv_callback = accept_cb;//regist一个回调函数

set_event(sockfd, EPOLLIN, 1);//关注EPOLLIN，关注到一个事件
}

(6) Out of memory（进程被kill掉）

解决办法：为虚拟机设置更大的内存。

4. 代码思路

1 使用epoll_create()构建一个机制，返回一个描述符 2 注册一个fd（包括创建套接字、绑定服务器ip和端口、listen监听连接） 3 调用回调函数accept，建立客服端连接，注册对应的event（读写分开） 4 建立相应的event，通过epoll_ctl监听事件类型（注册、修改、删除） 5 通过mainloop执行回调函数，通过epoll_wait返回就绪fd数量，判断每个事件的状态，读状态和写状态分别执行recv和send函数。

5. 实现代码

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

#include "server.h"

#define CONNECTION_SIZE1048576 // 1024 * 1024

#define MAX_PORTS20

#define TIME_SUB_MS(tv1, tv2) ((tv1.tv_sec – tv2.tv_sec) * 1000 + (tv1.tv_usec – tv2.tv_usec) / 1000)

int accept_cb(int fd);
int recv_cb(int fd);
int send_cb(int fd);

int epfd = 0;
struct timeval begin;

struct conn conn_list[CONNECTION_SIZE] = {0};
// fd

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);

conn_list[fd].wlength = conn_list[fd].rlength;
memcpy(conn_list[fd].wbuffer, conn_list[fd].rbuffer, conn_list[fd].wlength);

printf("[%d]RECV: %s\\n", conn_list[fd].rlength, conn_list[fd].rbuffer);

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);

//set_event(fd, EPOLLOUT, 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);
}
}

}

}

6. 实现效果

参考学习：https://github.com/0voice