1.2 事件驱动reactor的原理与实现

2.2 Reactor 的原理和实现

2.2.1 什么是 Reactor？

Reactor 是一种计算模型，常用于处理异步事件驱动的编程架构。它能够有效地管理和调度多个事件源的输入输出操作，尤其是在高并发的环境中。

🔴 核心思想: 将 IO 的管理转变成了对事件的管理。

在 Reactor 模式中，主要有以下几个组件：

事件源 (Event Source): 这些是产生事件的实体，比如网络连接、文件 IO 等
事件分发器 (Event Demultiplexer): 负责监控所有事件源，并将就绪的事件分发给相应的处理程序
事件处理程序 (Event Handler): 定义如何处理特定类型的事件

Reactor 模式的优点在于，它能够有效地使用系统资源，减少上下文切换的开销，从而提高应用程序的性能。

Reactor 的核心: 不同的 IO 事件对应不同的回调函数：

register - 注册
callback - 回调

io         ->       event      ->       call back
listenfd   ->       EPOLLIN    ->       accept_cb
clientfd   ->       EPOLLIN    ->       recv_cb
clientfd   ->       EPOLLOUT   ->       send_cb

两类事件，两类 fd (IO)。我们要采用事件的处理方式，所以要更关注事件和回调的匹配。每一个 IO 与之对应的有哪些匹配的参数。

EPOLLIN 这个事件对应两种情况，分别是客户端连入和客户端发来消息；EPOLLOUT 这个事件对应客户端可写。

2.2.2 Reactor 的实现

#include "../base_tool.h"
#include <sys/epoll.h>
#include <signal.h>
#include <string.h>

int epfd = 0;
#define BUFFER_LENGTH 1024
#define CONN_SIZE 1048576

typedef int (*RCALLBACK)(int fd);

struct conn {  // 什么事件，执行什么回调函数
    int fd;
    char rbuffer[BUFFER_LENGTH];
    char rlength;
    char wbuffer[BUFFER_LENGTH];
    char wlength;

    RCALLBACK send_callback;
    union {
        RCALLBACK recv_callback;
        RCALLBACK accept_callback;  // 只有 EPOLLIN 和 EPOLLOUT 两个事件，所以accept和recv是或的关系
    } r_action;
};

struct conn conn_list[CONN_SIZE] = {0};

long GetUnixTime() {
    auto now = std::chrono::system_clock::now();
    // 转换为时间戳
    auto timestamp = std::chrono::system_clock::to_time_t(now);
    return timestamp;
}

void set_event(int fd, int event, int flag) {  // 设置事件
    if (flag) {  // add
        struct epoll_event ev;
        ev.events = event;
        ev.data.fd = fd;
        epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &ev);
    } else {  // modify
        struct epoll_event ev;
        ev.events = event;
        ev.data.fd = fd;
        epoll_ctl(epfd, EPOLL_CTL_MOD, fd, &ev);
    }
}

int init_server(unsigned short port) {  // 初始化一个server
    int sockfd = socket(AF_INET, SOCK_STREAM, 0);  // 创建一个socket
    struct sockaddr_in servaddr;

    servaddr.sin_family = AF_INET;                 // 指定IP地址版本为IPV4
    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);  // 绑定本地0.0.0.0
    servaddr.sin_port = htons(port);               // 0~1023系统默认，需要给一个大于1024的地址，小于65535

    if (-1 == bind(sockfd, (struct sockaddr *)&servaddr, sizeof(struct sockaddr))) {
        std::cout << "bind failed" << std::endl;
        return -1;
    }
    
    listen(sockfd, 10);
    printf("listen finished\n");
    return sockfd;
}

int recv_callback(int fd) {
    int count = recv(fd, conn_list[fd].rbuffer, BUFFER_LENGTH, 0);
    conn_list[fd].rlength = count;
    
    if (count == 0) {
        printf("client disconnect%d\n", fd);
        epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL);  // TODO: unfinished
        close(fd);
        return -1;
    }
    
    printf("RECV :%s\n", conn_list[fd].rbuffer);

    // echo: 默认将客户端发送的内容，回给客户端
    memcpy(conn_list[fd].wbuffer, conn_list[fd].rbuffer, conn_list[fd].rlength);
    conn_list[fd].wlength = conn_list[fd].rlength;
    set_event(fd, EPOLLOUT, 0);
    
    return count;
}

int send_callback(int fd) {
    int count = send(fd, conn_list[fd].wbuffer, BUFFER_LENGTH, 0);
    printf("SEND :%s\n", conn_list[fd].wbuffer);
    set_event(fd, EPOLLIN, 0);
    return count;
}

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_callback;
    conn_list[fd].send_callback = send_callback;
    
    // 为新的链接设置回调
    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);
    return 0;
}

int accept_callback(int fd) {
    struct sockaddr client_addr;
    socklen_t client_addr_len = sizeof(client_addr);
    int clientfd = accept(fd, &client_addr, &client_addr_len);

    printf("accept clienfd=%d\n", clientfd);
    
    if (clientfd < 0) {
        printf("accept failed %s\n", strerror(errno));
        return -1;
    }
    
    event_register(clientfd, EPOLLIN);  // 接受到新的客户端fd，注册EPOLLIN事件
    return 0;
}

int main(int argc, char **argv) {
    if (argc <= 1) {
        printf("Usage: %s port\n", argv[0]);
        exit(0);
    }
    
    unsigned short port = atoi(argv[1]);
    int sockfd = init_server(port);

    epfd = epoll_create(1);
    conn_list[sockfd].fd = sockfd;
    conn_list[sockfd].r_action.recv_callback = accept_callback;
    set_event(sockfd, EPOLLIN, 1);
    
    while (1) {  // mainloop
        struct epoll_event events[1024] = {0};
        int nready = epoll_wait(epfd, events, 1024, -1);
        
        for (int i = 0; i < nready; i++) {
            // 当有epoll连入时，根据具体时间，调用对应的函数
            int connfd = events[i].data.fd;
            
            // 一次循环中，IO的读写可能并存，一起处理
            if (events[i].events & EPOLLIN) {
                conn_list[connfd].r_action.recv_callback(connfd);  
                // 可能是accept_callback，他们共用一个函数空间，所以都能调到
            }

            if (events[i].events & EPOLLOUT) {
                conn_list[connfd].send_callback(connfd);
            }
        }
    }
}

核心要点:

不同的 IO 事件，做不同的 action
IO 做到独立，每个事件互相独立

2.2.3 测试并发的问题解决

2.2.3.1 文件打开数量过多

解决方案:

1	ulimit -n 1048576

或者在该文件后添加相应配置。

2.2.3.2 客户端提示：不能分配请求地址

原因: 五元组不够（可以开辟的端口数不足导致的）

TCP 分为五元组：(sip, dip, sport, dport, proto) - 源 IP、目标 IP、源端口、目标端口、协议

解决方案: 服务器可以选择启 20 个端口的服务器

// main    
epfd = epoll_create(1);

for (int i = 0; i < MAX_PORT; i++) {  // 初始化20个server，每个都注册
    int sockfd = init_server(port + i);
    conn_list[sockfd].fd = sockfd;
    conn_list[sockfd].r_action.recv_callback = accept_callback;
    set_event(sockfd, EPOLLIN, 1);
}