Linux c++ 2.1.3 reactor和百万并发

m0_65365783

316人浏览 · 2026-04-22 11:13:50

m0_65365783 · 2026-04-22 11:13:50 发布

代码部分


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

//定义全局变量，用来建立epoll
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作为下表  创建一个conn的数组，用来存连接和对应的读写混冲去，和回调指针
struct conn conn_list[CONNECTION_SIZE] = {0};


//设置事件
int set_event(int fd, int event, int flag){
    if(flag){ //非0，为建立连接
        struct epoll_event ev;
        ev.events = event;
        ev.data.fd = fd;
        把连接添加进fd，监听读时间
        epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &ev);
    }else {//为0，说明是读或者写，都属于修改 mod
        
        struct epoll_event ev;
        ev.events = event;
        ev.data.fd = fd;
        把epoll里的fd做修改。具体是读还是写，由传入的参数决定
        epoll_ctl(epfd, EPOLL_CTL_MOD, fd, &ev);
     }
   
}

//事件登记
int event_register(int fd, int event){
    
    把fd传入conn数组。
    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);
}



//sockfd-->epollin-->accpet_cb
    建立连接
int accept_cb(int fd){
    struct sockaddr_in  clientaddr;
	socklen_t len = sizeof(clientaddr);

	int clientfd = accept(fd, (struct sockaddr*)&clientaddr, &len);
    //printf("accpet finshed: %d\n", clientfd);

    if(clientfd < 0){
        printf("accept errno:%d --> %s\n", errno, strerror(errno));
        return -1;
    }
    连接建立好，clientfd分配好后，将clientfd放入connlist。监听读事件。
    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 finished:%d, time_used:%d\n",clientfd, time_used);
    }
    return 0;
}

接收数据
int recv_cb(int fd){
    接收数据。count为接收数据的大小
    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;
    }
    conn_list[fd].rlength = count;

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

#if 1 //echo
    接收完数据之后，把数据传给写缓冲区
    conn_list[fd].wlength = conn_list[fd].rlength;
    memcpy(conn_list[fd].wbuffer, conn_list[fd].rbuffer,conn_list[fd].wlength);

#endif
    接收完数据之后，也就是读完之后，该写了。所以设置为0（修改），epollout 监听写
    set_event(fd, EPOLLOUT, 0);

    return count;
}

会发数据
int send_cb(int fd){
    
    int count = send(fd, conn_list[fd].wbuffer, conn_list[fd].wlength, 0);
    发完之后，该读了（in）。连接已经建立。所以是0（mod）
    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(){
    端口2000
    unsigned short port = 2000;
    
    创建epoll
    epfd = epoll_create(1);
    
    一共有20个端口
    for (size_t i = 0; i < MAXPORTS; i++)
    {
        创建sockfd。并加入到connlist
        设置回调函数
        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);
    }
    
    20个端口全部创建好。也及时20个线程。
    gettimeofday(&begin, NULL);
  

    while (1)
    {    监听 epoll里的读事件，放到events里面
        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);
            }
        }
    }
    
}

为什么要有reactor？

简单说：传统的“一个连接一个线程”模式，在成千上万的并发连接面前会崩溃。因为线程的创建、切换和内存开销都太高了。

于是人们开始用 I/O 多路复用（select/poll/epoll）来解决“同时监控大量连接”的问题。但很快发现，光有 epoll 还不够——它只解决了“通知你有事发生”，但没解决“怎么组织代码来处理这些事”。

如果你直接基于 epoll 编程，通常会遇到以下三个非常痛苦的现实问题：

代码像意大利面条
epoll_wait 返回一堆就绪的文件描述符，你得自己判断是读事件、写事件、新连接还是错误。最终代码里到处是 if/else 或 switch/case，读数据和业务逻辑混在一起，改一个地方可能崩一片。
无法自然扩展多线程
epoll 本身没有线程模型。如果想利用多核 CPU，你需要自己去设计任务队列、线程池、同步机制……很容易引入锁竞争和死锁。而 Reactor 模式提供了现成的扩展思路（比如主从 Reactor、多线程 Handler）。
容易出 bug 且难以调试
非阻塞 I/O + epoll 的编程需要你手动维护每个连接的状态（读半包、写半包、缓冲区管理）。稍不注意就漏掉某个事件，或者重复读取，导致死循环或数据错乱。Reactor 把这些底层状态管理封装进框架，开发者只需要填回调逻辑。

Reactor 模式的出现，就是为了在“epoll 这类高效事件通知机制”之上，构建一个：