开启极速之旅:了解Linux网络设计中的Reactor模型与百万级并发实践

简介: 本文将带您进入Linux网络设计的奇妙世界,着重介绍了Reactor模型和百万级并发的实践经验。在快节奏的现代互联网环境下,高性能的网络应用对于提供卓越的用户体验至关重要。通过深入探索Reactor模型,我们将揭示其在构建高并发应用中的关键作用。从基本概念到实际应用,您将了解到如何通过Reactor模型优化网络通信,实现快速响应和高吞吐量。此外,我们将分享一些实践中的经验和技巧,包括事件驱动编程、多线程与多进程处理、负载均衡等方面,以帮助您更好地应对百万级并发挑战。

一、Reactor网络模型简介

什么是并发:网络并发,通俗的讲就是服务器可以承载的客户端数量,即服务器可以稳定保证客户端同时接入的数量。

Reactor模型开发效率比直接使用IO多路复用要高,它一般是单线程的,设计目标是希望一个线程使用CPU的全部资源;带来的优点是,在每个事件处理中很多时候不需要考虑共享资源的互斥访问。

Reactor模式是处理并发IO比较常见的模式,用于同步IO,核心思想是将所有要处理的IO事件注册到一个中心IO多路复用器上,同时主线程或进程阻塞在IO多路复用器上;一旦有事件到来或准备就绪,多路复用器返回并将事先注册的相应 I/O 事件分发到对应的处理器中。

二、Reactor的优点

1、响应快;不必为单个同步事件阻塞,虽然Reactor本身依然是同步的。
2、编程相对简单;可以最大程度的避免复杂的多线程及同步问题,尽可能的避免多线程、多进程的切换开销。
3、可扩展性;可通过增加Reactor实例个数,充分利用CPU资源。
4、高复用性;Reactor模型本身与事件处理逻辑无关,具有很高的复用性。

三、实现过程

step 1:定义Reactor模型相关结构体

reactor数据结构设计图如下:
image.png

结构说明:以fd作为索引,存放在block中;当一个fd到来时,通过fd/MAX先找到fd对应的block号,再通过fd%MAX找到对应的偏移地址。例如来了个fd=10000,每个块存放的最大item数量MAX=1024,那么fd对应的block序号等于10000/1024=9;偏移量等于10000%1024=784。这样就可以找到fd对应的数据存放地址item。

数据结构的代码实现如下:

struct ntyevnt{
   
   
    int fd;//事件fd
    char buffer[BUFFER_LENGTH];//缓冲区
    int length;//缓存长度
    int status;//状态

    int events;//事件
    void *arg;//callback的参数
    int(*callback)(int fd, int events, void* arg);//回调函数
};
struct eventblock{
   
   
    struct *sock_items;//事件集合
    struct eventblock *next;//指向下一个内存块
};
struct reactor{
   
   
    int epfd;//epoll的文件描述符
    int blkcnt;//事件块的数量
    struct eventblock *evtblk;//事件块的起始地址
};

step 2:实现Reactor容器初始化功能

我们这里使用epoll作为IO多路复用器。
思路:初始化reactor内存块,避免脏数据;创建events和block并初始化,将events添加到block中,将block添加到reactor的链表中管理。

int ntyreactor_init(struct ntyreactor *reactor)
{
   
   
    if (reactor == NULL)
        return -1;
    memset(reactor, 0, sizeof(struct ntyreactor));
    //创建epoll,作为IO多路复用器
    reactor->epfd = epoll_create(1);
    if (reactor->epfd <= 0)
    {
   
   
        printf("create epfd in %s error %s\n", __func__, strerror(errno));
        return -2;
    }

    // 创建事件集
    struct ntyevnt *events = (struct ntyevnt *)malloc(MAX_EPOLL_EVENTS * sizeof(struct ntyevnt));
    if (events == NULL)
    {
   
   
        printf("create ntyevnt in %s error %s\n", __func__, strerror(errno));
        close(reactor->epfd);
        return -3;
    }
    memset(events, 0, (MAX_EPOLL_EVENTS) * sizeof(struct ntyevent));

    //创建事件内存块
    struct eventblock *block = (struct eventblock*)malloc(sizeof(struct eventblock));
    if (block == NULL)
    {
   
   
        printf("create eventblock in %s error %s\n", __func__, strerror(errno));
        free(events);
        close(reactor->epfd);
        return -4;
    }
    block->events = events;
    block->next = NULL;

    // reactor初始化赋值
    reactor->evblks=block;
    reactor->blkcnt = 1;

    return 0;
}

step 3:实现socket初始化功能

定义成一个函数,方便初始化多个监听端口。

int init_sock(short port)
{
   
   
    int ret = 0;
    int fd = socket(AF_INET, SOCK_STREAM, 0);//创建套字接
    if (fd == -1)
    {
   
   
        printf("create socket in %s error %s\n", __func__, strerror(errno));
        return -1;
    }
    ret=fcntl(fd, F_SETFL, O_NONBLOCK);//设置非阻塞
    if (ret == -1)
    {
   
   
        printf("fcntl O_NONBLOCK in %s error %s\n", __func__, strerror(errno));
        return -1;
    }

    // 设置属性
    struct sockaddr_in server_addr;
    memset(&server_addr, 0, sizeof(server_addr));
    server_addr.sin_family = AF_INET;// IPV4
    server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    server_addr.sin_port = htons(port);

    // 绑定
    ret = bind(fd, (struct sockaddr*)&server_addr, sizeof(server_addr));
    if (ret == -1)
    {
   
   
        printf("bind() in %s error %s\n", __func__, strerror(errno));
        return -1;
    }

    //监听
    ret = listen(fd, 20);
    if (ret < 0)
    {
   
   
        printf("listen failed : %s\n", strerror(errno));
        return -1;
    }

    printf("listen server port : %d\n", port);

    return fd;
}

step 4:实现Reactor动态扩容功能

为了实现高并发,服务器需要监听多个端口。当高并发时需要reactor容器进行扩容管理。
核心思路:找到链表的末端,分别为events和block分配内存并初始化,将events添加到block中,将block添加到reactor的链表中管理。

int ntyreactor_alloc(struct ntyreactor *reactor)
{
   
   
    if (reactor == NULL)
        return -1;
    if (reactor->evblks == NULL)
        return -1;

    //找到链表末端
    struct eventblock *blk = reactor->evblks;
    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 ntyevent 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 -3;
    }
    block->events = evs;
    block->next = NULL;

    //实现扩容
    blk->next = block;
    reactor->blkcnt++;

    return 0;
}

step 5:实现Reactor索引功能

思路:通过fd/MAX先找到fd对应的block号,再通过fd%MAX找到对应的偏移地址。
例如来了个fd=10000,每个块存放的最大item数量MAX=1024,那么fd对应的block序号等于10000/1024=9;偏移量等于10000%1024=784。这样就可以找到fd对应的数据存放地址item。

struct ntyevent *ntyreactor_idx(struct ntyreactor *reactor, int sockfd)
{
   
   
    if (reactor == NULL)
        return NULL;
    if (reactor->evblks == NULL)
        return NULL;
    // fd所在block序号
    int blkidx = sockfd / MAX_EPOLL_EVENTS;
    while (blkidx >= reactor->blkcnt)
    {
   
   
        // 扩容
        ntyreactor_alloc(reactor);
    }

    //找到fd对应block的位置
    int i = 0;
    struct eventblock *blk = reactor->evblks;
    while (i++ != blkidx && blk != NULL)
    {
   
   
        blk = blk->next;
    }

    // 返回item 地址
    return &blk->events[sockfd%MAX_EPOLL_EVENTS];
}

step 6:实现设置事件信息功能

将事件的相关信息保存到数据结构中。主要实现填充关键信息到event结构体中。

void nty_event_set(struct ntyevent *ev,int fd,NCALLBACK callback,void *arg)
{
   
   
    ev->fd = fd;
    ev->events = 0;
    ev->callback = callback;
    ev->arg = arg;
}

step 7:实现IO事件监听功能

这里使用epoll作为IO多路复用器,将事件添加到epoll中监听。
思路:主要是epoll_ctl操作,将事件添加到reactor的event结构体中。

int nty_event_add(int epfd, int events, struct ntyevent *ev)
{
   
   
    // 设置epoll事件信息
    struct epoll_event ep_ev = {
   
    0,{
   
   0} };
    ep_ev.data.ptr = ev;
    ep_ev.events = ev->events = events;

    // 判断,设置epfd的操作模式
    int op;
    if (ev->status == 1)
        op = EPOLL_CTL_MOD;
    else
    {
   
   
        op = EPOLL_CTL_ADD;
        ev->status = 1;
    }


    // 设置epoll
    int ret = epoll_ctl(epfd, op, ev->fd, &ep_ev);
    if (ret < 0)
    {
   
   
        printf("event add failed [fd=%d], events[%d],ret:%d\n", ev->fd, events,ret);
        printf("event add failed in %s error %s\n", __func__, strerror(errno));
        return -1;
    }

    return 0;
}

step 8:实现IO事件移除功能

由于设置了非阻塞模式,当事件到来时,需要暂时移除监听,避免干扰。

int nty_event_del(int epfd, struct ntyevent *event)
{
   
   
    if (event->status != 1)
        return -1;

    struct epoll_event ep_ev = {
   
    0,{
   
   0} };
    ep_ev.data.ptr = event;
    event->status = 0;
    // 移除fd的监听
    epoll_ctl(epfd, EPOLL_CTR_DEL, &ep_ev);
    return 0;
}

step 9:实现Reactor事件监听功能

思路:设置fd的事件信息,添加事件到epoll监听。

int ntyreactor_addlistener(struct ntyreactor *reactor,int sockfd,NCALLBACK *acceptor)
{
   
   
    if (reactor == NULL)
        return -1;
    if (reactor->evblks == NULL)
        return -1;

    // 找到fd对应的event地址
    struct ntyevent *event = ntyreactor_idx(reactor, sockfd);
    if (event == NULL)
        return -1;
    // 设置fd的事件信息
    nty_event_set(event, sockfd, acceptor, reactor);
    // 添加事件到epoll监听
    nty_event_add(reactor->epfd, EPOLLIN, event);

    return 0;
}

step 10:实现recv回调函数

思路:找到fd对应的信息内存块;使用recv接收数据;暂时移除该事件的监听;如果接收成功,设置监听事件为是否可写,添加到IO多路复用器(epoll)中;返回收到的数据长度。

int recv_cb(int fd, int events, void *arg)
{
   
   
    struct ntyreactor *reactor = (struct ntyreactor *)arg;
    if (reactor == NULL)
        return -1;
    // 找到fd对应的event地址
    struct ntyevent *event = ntyreactor_idx(reactor, fd);
    if (event == NULL)
        return -1;
    // 接收数据
    int len = recv(fd, event->buffer, BUFFER_LENGTH, 0);
    // 暂时移除监听
    nty_event_del(reactor->epfd, event);
    if (len > 0)
    {
   
   
        event->length = len;
        event->buffer[len] = '\0';
        printf("recv [%d]:%s\n", fd, event->buffer);
        //设置fd的事件信息
        nty_event_set(event, fd, send_cb, reactor);
        // 添加事件到epoll监听
        nty_event_add(reactor->epfd, EPOLLOUT, event);
    }
    else if (len == 0)
    {
   
   
        nty_event_del(reactor->epfd, event);
        printf("recv_cb --> disconnect\n");
        close(event->fd);
    }
    else {
   
   

        if (errno == EAGAIN && errno == EWOULDBLOCK) {
   
    //

        }
        else if (errno == ECONNRESET) {
   
   
            nty_event_del(reactor->epfd, event);
            close(event->fd);
        }
        printf("recv[fd=%d] error[%d]:%s\n", fd, errno, strerror(errno));

    }
    return len;
}

step 11:实现send回调函数

思路:找到fd对应的信息内存块;使用send发送数据;暂时移除该事件的监听;如果发送成功,设置监听事件为是否可读,添加到IO多路复用器(epoll)中;返回发送的数据长度。

int send_cb(int fd, int events, void *arg)
{
   
   
    struct ntyreactor *reactor = (struct ntyreactor*)arg;
    if (reactor == NULL)
    {
   
   
        return -1;
    }

    // 查找fd对应的信息存放内存块
    struct ntyevent *ev = ntyreactor_idx(reactor, fd);
    if (ev == NULL)
        return -1;

    int len = send(fd, ev->buffer, BUFFER_LENGTH, 0);
    // 暂时移除监听
    nty_event_del(reactor->epfd, ev);

    if (len > 0)
    {
   
   

        printf("send[fd=%d], [%d]%s\n", fd, len, ev->buffer);
        nty_event_set(ev, fd, recv_cb, reactor);
        nty_event_add(reactor->epfd, EPOLLIN, ev);
    }
    else
    {
   
   
        close(ev->fd);
        printf("send[fd=%d] error %s\n", fd, strerror(errno));
    }
    return len;
}

step 12:实现accept回调函数

思路:使用accept获得连接的客户端fd;设置客户端fd为非阻塞模式;找到fd对应的信息内存块;设置fd的事件信息;设置监听事件为是否可读,添加到IO多路复用器(epoll)中。

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 client_fd = accept(fd,(struct sockaddr*)&client_addr,&len);
    if (client_fd == -1)
    {
   
   
        printf("accept: %s\n", strerror(errno));
        return -1;
    }

    int flag = fcntl(client_fd, F_SETFL, O_NONBLOCK);
    if ((flag = fcntl(client_fd, F_SETFL, O_NONBLOCK)) < 0) {
   
   
        printf("%s: fcntl nonblocking failed, %d\n", __func__, MAX_EPOLL_EVENTS);
        return -1;
    }

    // 找到fd对应的event地址
    struct ntyevent *event = ntyreactor_idx(reactor, client_fd);
    if (event == NULL)
        return -1;
    // 设置fd的事件信息
    nty_event_set(event, client_fd, recv_cb, reactor);
    // 添加事件到epoll监听
    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), client_fd);

    return 0;
}

step 13:实现reactor运行函数

主要是epoll的等待功能,将监听到的事件进行回调处理。

int ntyreactor_run(struct ntyreactor *reactor)
{
   
   
    if (reactor == NULL)
        return -1;
    if (reactor->epfd < 0)
        return -1;
    if (reactor->evblks == NULL)
        return -1;

    struct epoll_event events[MAX_EPOLL_EVENTS + 1];
    int i;
    while (1)
    {
   
   
        // epoll监听客户端接入
        int nready = epoll_wait(reactor->epfd, events, MAX_EPOLL_EVENTS, 1000);
        if (nready < 0)
        {
   
   
            printf("epoll wait error\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);
            }
        }
    }

}

step 14:实现reactor销毁功能

int ntyreactor_destory(struct ntyreactor *reactor)
{
   
   
    // 关闭epoll
    close(reactor->epfd);

    struct eventblock *blk= reactor->evblks;
    struct eventblock *next;

    while (blk != NULL)
    {
   
   
        next = blk->next;
        // 释放内存块
        free(blk->events);
        free(blk);
        blk = next;
    }
    return 0;
}

简单使用示例

int main(int argc,char* argv[])
{
   
   
    // 创建reactor对象
    struct ntyreactor *reactor = (struct ntyreactor*)malloc(sizeof(struct ntyreactor));
    // 初始化reactor容器
    ntyreactor_init(reactor);

    // 定义监控的socket开始端口
    unsigned short port = SERVER_PORT;
    if (argc == 2)
    {
   
   
        port = atoi(argv[1]);
    }

    // 初始化套字接和监听端口列表
    int i = 0;
    int sockfds[PORT_COUNT] = {
   
    0 };
    for(i=0;i<PORT_COUNT;i++)
    {
   
   
        // 初始化socket端口
        sockfds[i] = init_sock(port + i);
        // 添加事件监听
        ntyreactor_addlistener(reactor, sockfds[i], accept_cb);
    }

    // reactor运行,主要是epoll的循环监听
    ntyreactor_run(reactor);
    // 销毁 reactor
    ntyreactor_destory(reactor);

    // 关闭socket集
    for (i = 0; i < PORT_COUNT; i++)
    {
   
   
        close(sockfds[i]);
    }

    // 释放reactor
    free(reactor);

    return 0;
}

四、总结

1、创建一个reactor对象,分配内存。
2、Reactor容器初始化:
(1)创建IO多路复用器epoll,将文件描述符epfd保存到reactor容器中;
(2)创建一个事件集合MAX_EPOLL_EVENTS * sizeof(struct events);
(3)创建一个管理模块,用于管理(2)中创建的事件集合。这是一个链表,一个指针指向(2)中创建的事件集合,一个指针指向下一个块(block);
(4)将事件集合添加到管理模块(block)中进管理;
(5)将管理模块(block)保存到reactor容器,同时reactor的管理模块(block)置为1;
(6)注意:如果申请内存使用malloc函数,最好使用memset将内存块初始化,避免脏数据。
3、soket初始化:
(1)创建一个监听socket fd,socket(AF_INET,SOCK_STREAM,0);
(2)设置socket fd为非阻塞模式,fcntl(fd,SETFL,O_NONBLOCK);
(3)配置socket属性,主要是struct sockaddr_in结构体的sin_family、sin_addr.s_addr、sin_port;
(4)bind();
(5)listen();
4、事件监听
(1)以fd为key找到对应的管理块序号和偏移地址;
(2)设置事件信息;
(3)添加监听事件:是否可读。
5、reactor主循环(mainloop),主要是epoll的循环监听,处理事件到相关回调函数。
(1)事件监听,epoll_wait;
(2)回调函数recv,处理可读事件(EPOLLIN);
(3)回调函数send,处理可写事件(EPOLLOUT)。
6、销毁reactor,注意是事件集内存块的释放和链表的内存释放。
7、close监听端口socket fd。
8、释放reactor内存。
image.png

相关实践学习
CentOS 7迁移Anolis OS 7
龙蜥操作系统Anolis OS的体验。Anolis OS 7生态上和依赖管理上保持跟CentOS 7.x兼容,一键式迁移脚本centos2anolis.py。本文为您介绍如何通过AOMS迁移工具实现CentOS 7.x到Anolis OS 7的迁移。
目录
相关文章
|
4天前
|
监控 安全 BI
什么是零信任模型?如何实施以保证网络安全?
随着数字化转型,网络边界不断变化,组织需采用新的安全方法。零信任基于“永不信任,永远验证”原则,强调无论内外部,任何用户、设备或网络都不可信任。该模型包括微分段、多因素身份验证、单点登录、最小特权原则、持续监控和审核用户活动、监控设备等核心准则,以实现强大的网络安全态势。
|
17天前
|
监控 算法 Linux
Linux内核锁机制深度剖析与实践优化####
本文作为一篇技术性文章,深入探讨了Linux操作系统内核中锁机制的工作原理、类型及其在并发控制中的应用,旨在为开发者提供关于如何有效利用这些工具来提升系统性能和稳定性的见解。不同于常规摘要的概述性质,本文将直接通过具体案例分析,展示在不同场景下选择合适的锁策略对于解决竞争条件、死锁问题的重要性,以及如何根据实际需求调整锁的粒度以达到最佳效果,为读者呈现一份实用性强的实践指南。 ####
|
17天前
|
缓存 监控 网络协议
Linux操作系统的内核优化与实践####
本文旨在探讨Linux操作系统内核的优化策略与实际应用案例,深入分析内核参数调优、编译选项配置及实时性能监控的方法。通过具体实例讲解如何根据不同应用场景调整内核设置,以提升系统性能和稳定性,为系统管理员和技术爱好者提供实用的优化指南。 ####
|
1月前
|
缓存 Linux 开发者
Linux内核中的并发控制机制:深入理解与应用####
【10月更文挑战第21天】 本文旨在为读者提供一个全面的指南,探讨Linux操作系统中用于实现多线程和进程间同步的关键技术——并发控制机制。通过剖析互斥锁、自旋锁、读写锁等核心概念及其在实际场景中的应用,本文将帮助开发者更好地理解和运用这些工具来构建高效且稳定的应用程序。 ####
41 5
|
1月前
|
存储 网络协议 安全
30 道初级网络工程师面试题,涵盖 OSI 模型、TCP/IP 协议栈、IP 地址、子网掩码、VLAN、STP、DHCP、DNS、防火墙、NAT、VPN 等基础知识和技术,帮助小白们充分准备面试,顺利踏入职场
本文精选了 30 道初级网络工程师面试题,涵盖 OSI 模型、TCP/IP 协议栈、IP 地址、子网掩码、VLAN、STP、DHCP、DNS、防火墙、NAT、VPN 等基础知识和技术,帮助小白们充分准备面试,顺利踏入职场。
83 2
|
1月前
|
Linux 数据库
Linux内核中的锁机制:保障并发操作的数据一致性####
【10月更文挑战第29天】 在多线程编程中,确保数据一致性和防止竞争条件是至关重要的。本文将深入探讨Linux操作系统中实现的几种关键锁机制,包括自旋锁、互斥锁和读写锁等。通过分析这些锁的设计原理和使用场景,帮助读者理解如何在实际应用中选择合适的锁机制以优化系统性能和稳定性。 ####
60 6
|
1月前
|
运维 网络协议 算法
7 层 OSI 参考模型:详解网络通信的层次结构
7 层 OSI 参考模型:详解网络通信的层次结构
122 1
|
1月前
|
关系型数据库 MySQL Linux
Linux环境下MySQL数据库自动定时备份实践
数据库备份是确保数据安全的重要措施。在Linux环境下,实现MySQL数据库的自动定时备份可以通过多种方式完成。本文将介绍如何使用`cron`定时任务和`mysqldump`工具来实现MySQL数据库的每日自动备份。
97 3
|
1月前
|
网络协议 算法 网络性能优化
计算机网络常见面试题(一):TCP/IP五层模型、TCP三次握手、四次挥手,TCP传输可靠性保障、ARQ协议
计算机网络常见面试题(一):TCP/IP五层模型、应用层常见的协议、TCP与UDP的区别,TCP三次握手、四次挥手,TCP传输可靠性保障、ARQ协议、ARP协议
|
6天前
|
SQL 安全 网络安全
网络安全与信息安全:知识分享####
【10月更文挑战第21天】 随着数字化时代的快速发展,网络安全和信息安全已成为个人和企业不可忽视的关键问题。本文将探讨网络安全漏洞、加密技术以及安全意识的重要性,并提供一些实用的建议,帮助读者提高自身的网络安全防护能力。 ####
43 17