AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

epoll封装reactor原理剖析与代码实现(2)

2024-01-15 91
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： epoll封装reactor原理剖析与代码实现(2)

reactor优点

reactor模式是编写高性能网络服务器的必备技术之一，它具有如下优点：


  1. 响应快，不必为单个同步时间所阻塞，虽然 reactor本身依然是同步的
  2. 编程相对简单，可以最大程度的避免复杂的多线程及同步问题，并且避免了多线程/进程的切换开销
  3. 可扩展性，可以方便的通过增加 reactor实例个数来充分利用 CPU 资源
  4. 可复用性，reactor 框架本身与具体事件处理逻辑无关，具有很高的复用性

 reactor模型开发效率上比起直接使用 IO 复用要高，它通常是单线程的，设计目标是希望单线程使用一颗 CPU 的全部资源，但也有附带优点，即每个事件处理中很多时候可以不考虑共享资源的互斥访问。可是缺点也是明显的，现在的硬件发展，已经不再遵循摩尔定律，CPU 的频率受制于材料的限制不再有大的提升，而改为是从核数的增加上提升能力，当程序需要使用多核资源时，reactor模型就会悲剧。

 如果程序业务很简单，例如只是简单的访问一些提供了并发访问的服务，就可以直接开启多个反应堆，每个反应堆对应一颗 CPU 核心，这些反应堆上跑的请求互不相关，这是完全可以利用多核的。例如 Nginx 这样的 http 静态服务器。

reactor多种模型

单reactor + 单线程模型

  单reactor单线程模型，指的是所有的 IO 操作（读，写，建立连接）都在同一个线程上面完成

缺点：

  1. 由于只有一个线程，因此事件是顺序处理的，一个线程同时只能做一件事情，事件的优先级得不到保证
  2. 不能充分利用多核CPU

单reactor + 线程池(Thread Pool)模型

 相比于单reactor单线程模型，此模型中收到请求后，不在reactor线程计算，而是使用线程池来计算，这会充分的利用多核CPU。采用此模式时有可能存在多个线程同时计算同一个连接上的多个请求，算出的结果的次序是不确定的， 所以需要网络框架在设计协议时带一个id标示，以便以便让客户端区分response对应的是哪个request。

多reactor + 多线程模型

 此模式的特点是每个线程一个循环， 有一个main reactor负责accept连接, 然后把该连接挂在某个sub reactor中，这样该连接的所有操作都在那个sub reactor所处的线程中完成。多个连接可能被分配到多个线程中，充分利用CPU。在应用场景中，reactor的个数可以采用 固定的个数，比如跟CPU数目一致。此模型与单reactor多线程模型相比，减少了进出thread pool两次上下文切换，小规模的计算可以在当前IO线程完成并且返回结果，降低响应的延迟。并可以有效防止当IO压力过大时一个reactor处理能力饱和问题。

多reactor + 线程池(Thread Pool)模型

  此模型是上面两个的混合体，它既使用多个 reactors 来处理 IO，又使用线程池来处理计算。此模式适适合既有突发IO(利用Multiple Reactor分担)，又有突发计算的应用（利用线程池把一个连接上的计算任务分配给多个线程）。

注意点

注意：前面介绍的四种reactor 模式在具体实现时为了简应该遵循的原则是：每个文件描述符只由一个线程操作。这样可以轻轻松松解决消息收发的顺序性问题，也避免了关闭文件描述符的各种race condition。一个线程可以操作多个文件描述符，但是一个线程不能操作别的线程拥有的文件描述符。这一点不难做到。epoll也遵循了相同的原则。Linux文档中并没有说明，当一个线程证阻塞在epoll_wait时，另一个线程往epoll fd添加一个新的监控fd会发生什么。新fd上的事件会不会在此次epoll_wait调用中返回？为了稳妥起见，我们应该吧对同一个 epoll fd的操作(添加、删除、修改等等)都放到同一个线程中执行。

reactor完善版代码

由于fd的数量未知，这里设计ntyreactor 里面包含 eventblock ，eventblock 包含1024个fd。每个fd通过 fd/1024定位到在第几个eventblock，通过fd%1024定位到在eventblock第几个位置。

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/epoll.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <unistd.h>
#include <errno.h>
#define BUFFER_LENGTH           4096
#define MAX_EPOLL_EVENTS        1024
#define SERVER_PORT             8081
#define PORT_COUNT              100
typedef int (*NCALLBACK)(int, int, void *);
struct ntyevent {
    int fd;
    int events;
    void *arg;
    NCALLBACK callback;
    int status;
    char buffer[BUFFER_LENGTH];
    int length;
};
struct eventblock {
    struct eventblock *next;
    struct ntyevent *events;
};
struct ntyreactor {
    int epfd;
    int blkcnt;
    struct eventblock *evblk;
};
int recv_cb(int fd, int events, void *arg);
int send_cb(int fd, int events, void *arg);
struct ntyevent *ntyreactor_find_event_idx(struct ntyreactor *reactor, int sockfd);
void nty_event_set(struct ntyevent *ev, int fd, NCALLBACK *callback, void *arg) {
    ev->fd = fd;
    ev->callback = callback;
    ev->events = 0;
    ev->arg = arg;
}
int nty_event_add(int epfd, int events, struct ntyevent *ev) {
    struct epoll_event ep_ev = {0, {0}};
    ep_ev.data.ptr = ev;
    ep_ev.events = ev->events = events;
    int op;
    if (ev->status == 1) {
        op = EPOLL_CTL_MOD;
    }
    else {
        op = EPOLL_CTL_ADD;
        ev->status = 1;
    }
    if (epoll_ctl(epfd, op, ev->fd, &ep_ev) < 0) {
        printf("event add failed [fd=%d], events[%d]\n", ev->fd, events);
        return -1;
    }
    return 0;
}
int nty_event_del(int epfd, struct ntyevent *ev) {
    struct epoll_event ep_ev = {0, {0}};
    if (ev->status != 1) {
        return -1;
    }
    ep_ev.data.ptr = ev;
    ev->status = 0;
    epoll_ctl(epfd, EPOLL_CTL_DEL, ev->fd, &ep_ev);
    return 0;
}
int recv_cb(int fd, int events, void *arg) {
    struct ntyreactor *reactor = (struct ntyreactor *) arg;
    struct ntyevent *ev = ntyreactor_find_event_idx(reactor, fd);
    int len = recv(fd, ev->buffer, BUFFER_LENGTH, 0); //
    nty_event_del(reactor->epfd, ev);
    if (len > 0) {
        ev->length = len;
        ev->buffer[len] = '\0';
//        printf("recv[%d]:%s\n", fd, ev->buffer);
        printf("recv fd=[%d\n", fd);
        nty_event_set(ev, fd, send_cb, reactor);
        nty_event_add(reactor->epfd, EPOLLOUT, ev);
    }
    else if (len == 0) {
        close(ev->fd);
        //printf("[fd=%d] pos[%ld], closed\n", fd, ev-reactor->events);
    }
    else {
        close(ev->fd);
//        printf("recv[fd=%d] error[%d]:%s\n", fd, errno, strerror(errno));
    }
    return len;
}
int send_cb(int fd, int events, void *arg) {
    struct ntyreactor *reactor = (struct ntyreactor *) arg;
    struct ntyevent *ev = ntyreactor_find_event_idx(reactor, fd);
    int len = send(fd, ev->buffer, ev->length, 0);
    if (len > 0) {
//        printf("send[fd=%d], [%d]%s\n", fd, len, ev->buffer);
        printf("send fd=[%d\n]", fd);
        nty_event_del(reactor->epfd, ev);
        nty_event_set(ev, fd, recv_cb, reactor);
        nty_event_add(reactor->epfd, EPOLLIN, ev);
    }
    else {
        nty_event_del(reactor->epfd, ev);
        close(ev->fd);
        printf("send[fd=%d] error %s\n", fd, strerror(errno));
    }
    return len;
}
int accept_cb(int fd, int events, void *arg) {//非阻塞
    struct ntyreactor *reactor = (struct ntyreactor *) arg;
    if (reactor == NULL) return -1;
    struct sockaddr_in client_addr;
    socklen_t len = sizeof(client_addr);
    int clientfd;
    if ((clientfd = accept(fd, (struct sockaddr *) &client_addr, &len)) == -1) {
        printf("accept: %s\n", strerror(errno));
        return -1;
    }
    if ((fcntl(clientfd, F_SETFL, O_NONBLOCK)) < 0) {
        printf("%s: fcntl nonblocking failed, %d\n", __func__, MAX_EPOLL_EVENTS);
        return -1;
    }
    struct ntyevent *event = ntyreactor_find_event_idx(reactor, clientfd);
    nty_event_set(event, clientfd, recv_cb, reactor);
    nty_event_add(reactor->epfd, EPOLLIN, event);
    printf("new connect [%s:%d], pos[%d]\n",
           inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port), clientfd);
    return 0;
}
int init_sock(short port) {
    int fd = socket(AF_INET, SOCK_STREAM, 0);
    fcntl(fd, F_SETFL, O_NONBLOCK);
    struct sockaddr_in server_addr;
    memset(&server_addr, 0, sizeof(server_addr));
    server_addr.sin_family = AF_INET;
    server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    server_addr.sin_port = htons(port);
    bind(fd, (struct sockaddr *) &server_addr, sizeof(server_addr));
    if (listen(fd, 20) < 0) {
        printf("listen failed : %s\n", strerror(errno));
    }
    return fd;
}
int ntyreactor_alloc(struct ntyreactor *reactor) {
    if (reactor == NULL) return -1;
    if (reactor->evblk == NULL) return -1;
    struct eventblock *blk = reactor->evblk;
    while (blk->next != NULL) {
        blk = blk->next;
    }
    struct ntyevent *evs = (struct ntyevent *) malloc((MAX_EPOLL_EVENTS) * sizeof(struct ntyevent));
    if (evs == NULL) {
        printf("ntyreactor_alloc ntyevents failed\n");
        return -2;
    }
    memset(evs, 0, (MAX_EPOLL_EVENTS) * sizeof(struct ntyevent));
    struct eventblock *block = (struct eventblock *) malloc(sizeof(struct eventblock));
    if (block == NULL) {
        printf("ntyreactor_alloc eventblock failed\n");
        return -2;
    }
    memset(block, 0, sizeof(struct eventblock));
    block->events = evs;
    block->next = NULL;
    blk->next = block;
    reactor->blkcnt++; //
    return 0;
}
struct ntyevent *ntyreactor_find_event_idx(struct ntyreactor *reactor, int sockfd) {
    int blkidx = sockfd / MAX_EPOLL_EVENTS;
    while (blkidx >= reactor->blkcnt) {
        ntyreactor_alloc(reactor);
    }
    int i = 0;
    struct eventblock *blk = reactor->evblk;
    while (i++ < blkidx && blk != NULL) {
        blk = blk->next;
    }
    return &blk->events[sockfd % MAX_EPOLL_EVENTS];
}
int ntyreactor_init(struct ntyreactor *reactor) {
    if (reactor == NULL) return -1;
    memset(reactor, 0, sizeof(struct ntyreactor));
    reactor->epfd = epoll_create(1);
    if (reactor->epfd <= 0) {
        printf("create epfd in %s err %s\n", __func__, strerror(errno));
        return -2;
    }
    struct ntyevent *evs = (struct ntyevent *) malloc((MAX_EPOLL_EVENTS) * sizeof(struct ntyevent));
    if (evs == NULL) {
        printf("ntyreactor_alloc ntyevents failed\n");
        return -2;
    }
    memset(evs, 0, (MAX_EPOLL_EVENTS) * sizeof(struct ntyevent));
    struct eventblock *block = (struct eventblock *) malloc(sizeof(struct eventblock));
    if (block == NULL) {
        printf("ntyreactor_alloc eventblock failed\n");
        return -2;
    }
    memset(block, 0, sizeof(struct eventblock));
    block->events = evs;
    block->next = NULL;
    reactor->evblk = block;
    reactor->blkcnt = 1;
    return 0;
}
int ntyreactor_destory(struct ntyreactor *reactor) {
    close(reactor->epfd);
    //free(reactor->events);
    struct eventblock *blk = reactor->evblk;
    struct eventblock *blk_next = NULL;
    while (blk != NULL) {
        blk_next = blk->next;
        free(blk->events);
        free(blk);
        blk = blk_next;
    }
    return 0;
}
int ntyreactor_addlistener(struct ntyreactor *reactor, int sockfd, NCALLBACK *acceptor) {
    if (reactor == NULL) return -1;
    if (reactor->evblk == NULL) return -1;
    struct ntyevent *event = ntyreactor_find_event_idx(reactor, sockfd);
    nty_event_set(event, sockfd, acceptor, reactor);
    nty_event_add(reactor->epfd, EPOLLIN, event);
    return 0;
}
_Noreturn int ntyreactor_run(struct ntyreactor *reactor) {
    if (reactor == NULL) return -1;
    if (reactor->epfd < 0) return -1;
    if (reactor->evblk == NULL) return -1;
    struct epoll_event events[MAX_EPOLL_EVENTS + 1];
    int i;
    while (1) {
        int nready = epoll_wait(reactor->epfd, events, MAX_EPOLL_EVENTS, 1000);
        if (nready < 0) {
            printf("epoll_wait error, exit\n");
            continue;
        }
        for (i = 0; i < nready; i++) {
            struct ntyevent *ev = (struct ntyevent *) events[i].data.ptr;
            if ((events[i].events & EPOLLIN) && (ev->events & EPOLLIN)) {
                ev->callback(ev->fd, events[i].events, ev->arg);
            }
            if ((events[i].events & EPOLLOUT) && (ev->events & EPOLLOUT)) {
                ev->callback(ev->fd, events[i].events, ev->arg);
            }
        }
    }
}
// <remoteip, remoteport, localip, localport,protocol>
int main(int argc, char *argv[]) {
    unsigned short port = SERVER_PORT; // listen 8081
    if (argc == 2) {
        port = atoi(argv[1]);
    }
    struct ntyreactor *reactor = (struct ntyreactor *) malloc(sizeof(struct ntyreactor));
    ntyreactor_init(reactor);
    int i = 0;
    int sockfds[PORT_COUNT] = {0};
    for (i = 0; i < PORT_COUNT; i++) {
        sockfds[i] = init_sock(port + i);
        ntyreactor_addlistener(reactor, sockfds[i], accept_cb);
    }
    ntyreactor_run(reactor);
    ntyreactor_destory(reactor);
    for (i = 0; i < PORT_COUNT; i++) {
        close(sockfds[i]);
    }
    free(reactor);
    return 0;
}


文章标签：
Java
应用服务中间件
监控
Linux
nginx
cheems~
目录
相关文章
新法
|
6月前
|
存储 Linux 调度
io复用之epoll核心源码剖析
epoll底层实现中有两个关键的数据结构，一个是eventpoll另一个是epitem，其中eventpoll中有两个成员变量分别是rbr和rdlist,前者指向一颗红黑树的根，后者指向双向链表的头。而epitem则是红黑树节点和双向链表节点的综合体，也就是说epitem即可作为树的节点，又可以作为链表的节点，并且epitem中包含着用户注册的事件。当用户调用epoll_create()时，会创建eventpoll对象（包含一个红黑树和一个双链表）；
新法
122 0
io复用之epoll核心源码剖析
mikechen的互联网架构
|
11小时前
|
网络协议 Dubbo Java
一文搞懂NIO、AIO、BIO的核心区别(建议收藏)
本文详细解析了NIO、AIO、BIO的核心区别，NIO的三个核心概念，以及NIO在Java框架中的应用等。关注【mikechen的互联网架构】，10年+BAT架构经验倾囊相授。
mikechen的互联网架构
14 4
一文搞懂NIO、AIO、BIO的核心区别(建议收藏)
nanshaws
|
1月前
|
NoSQL Java Redis
Reactor实战，创建一个简单的单线程Reactor（理解了就相当于理解了多线程的Reactor）
本文通过一个简单的单线程Reactor模式的Java代码示例，展示了如何使用NIO创建一个服务端，处理客户端的连接和数据读写，帮助理解Reactor模式的核心原理。
nanshaws
25 0
Reactor实战，创建一个简单的单线程Reactor（理解了就相当于理解了多线程的Reactor）
小狮子安度因1
|
3月前
|
存储 网络协议 安全
Epoll的实现原理
Epoll的实现原理
小狮子安度因1
37 4
A码农先森
|
5月前
|
应用服务中间件 Go nginx
Swoole 源码分析之 epoll 多路复用模块
IO多路复用技术通过使用少量的线程或进程同时监视多个IO事件，能够更高效地处理大量的IO操作，从而提高系统的性能和资源利用率。
A码农先森
56 0
Swoole 源码分析之 epoll 多路复用模块
wjq++
|
5月前
|
存储 安全 网络协议
epoll的底层实现原理
epoll的底层实现原理
wjq++
51 0
dearQiHao
|
6月前
|
监控 安全 Linux
reactor的原理与实现
前情回顾 网络IO，会涉及到两个系统对象：   一个是用户空间调用的进程或线程   一个是内核空间的内核系统 如果发生IO操作read时，会奖励两个阶段：
dearQiHao
67 1
cheems~
|
6月前
|
Linux API C++
epoll封装reactor原理剖析与代码实现(1)
epoll封装reactor原理剖析与代码实现(1)
cheems~
196 0
Aries_Ro
|
6月前
|
监控 Linux API
epoll-reactor模型原理及代码解析
epoll-reactor模型原理及代码解析
Aries_Ro
144 0
cheems~
|
6月前
|
存储 安全 网络协议
epoll的实现原理
epoll的实现原理
cheems~
73 0