服务器IO多路复用之epoll反应堆

思想

epoll反应堆的核心思想：将文件描述符、事件、回调函数使用自定义结构体封装在一起，当某个文件描述符的事件被触发，会自动调用回调函数既可以实现对应的IO操作。最终目的实现不同的文件描述对应不同的事件处理函数。

实例

#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/types.h>      
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/ip.h>
#include <sys/epoll.h>

/* 定义epoll反应堆结构体 */
struct my_event
{
    int epfd;/* epoll实例的文件描述符 */
    int (*callback)(struct my_event *ev);/* 回调函数的参数 */
    int fd;/* 文件描述符 */
    uint32_t events;/* 事件 */
};

/* 定义通信套接字文件描述符connfd在有客户端发送数据请求的读事件的回调函数 */
int read_data(struct my_event *ev)
{
    int ret;
    char buf[128];
    struct epoll_event event;
    int epfd = ev->epfd;
    int connfd = ev->fd;

    /* 接收客户端的数据请求 */
    memset(buf, 0, sizeof(buf));
    ret = read(connfd, buf, sizeof(buf));
    if (ret == -1)
    {
        perror("read");
        return -1;
    }
    else if (ret == 0)
    {
        printf("fd = %d quit\n", connfd);
        /*说明对端(客户端)退出：完成下树，将退出的文件描述符fd及其事件从epoll实例中删除 */
        event.events = ev->events;
        event.data.ptr = NULL;
        ret = epoll_ctl(epfd, EPOLL_CTL_DEL, connfd, &event);
        if (ret == -1)
        {
            perror("epoll_ctl->EPOLL_CTL_DEL");
            return -1;
        }
        close(connfd);
        /* 释放空间 */
        if (ev != NULL)
        {
            free(ev);
            ev = NULL;
        }
        return 0;
    }
    else
    {
        /* 调用write数据的回写 */
        ret = write(connfd, buf, sizeof(buf));
        if (ret == -1)
        {
            perror("write");
            return -1;
        }
    }
}


/* 定义监听套接字文件描述符listenfd在检测到有客户端连接的读事件的回调函数 */
int accept_client(struct my_event *ev)
{
    int ret;
    struct epoll_event event;
    struct sockaddr_in cltaddr;
    socklen_t addrlen = sizeof(cltaddr);
    int listenfd = ev->fd;
    int connfd;
    int epfd = ev->epfd;
    struct my_event *my_ev;

    connfd = accept(listenfd, (struct sockaddr *)&cltaddr, &addrlen);
    if (connfd == -1)
    {
        perror("accept");
        return -1;
    }

    printf("connfd = %d client connect success\n", connfd);

    /* 反应堆上树：有新的客户端连接成功,将新的通信套接字文件描述符、事件以及回调函数添加到epoll实例中 */
    event.events = EPOLLIN;/* 读事件 */
    my_ev = calloc(1, sizeof(struct my_event));
    my_ev->epfd = epfd;/* epoll实例的文件描述符 */
    my_ev->fd = connfd;/* 文件描述符 */
    my_ev->callback = read_data;/* 回调函数 */
    event.data.ptr = my_ev;
    ret = epoll_ctl(epfd, EPOLL_CTL_ADD, connfd, &event);
    if (ret == -1)
    {
        perror("epoll_ctl->EPOLL_CTL_ADD");
        return -1;
    }
}


int main()
{
    int listenfd;
    int ret;
    int count;
    int optval;
    struct sockaddr_in srvaddr;
    int epfd;
    int i;
    struct epoll_event event;
    struct epoll_event events[8];
    struct my_event *my_ev;

    /* 1. 创建TCP流式套接字文件 */
    listenfd = socket(AF_INET, SOCK_STREAM, 0);
    if (listenfd == -1)
    {
        perror("socket");
        return -1;
    }

    /* 设置套接字允许IP地址和端口被重新设置 */
    optval = 1;/* 允许IP地址和端口被重新设置 */
    ret = setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval));
    if (ret == -1)
    {
        perror("setsockopt");
        return -1;
    }

    /* 2. 设置服务器主机的IP地址和端口号 */
    srvaddr.sin_family = AF_INET;
    srvaddr.sin_port = htons(8888);
    srvaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    ret = bind(listenfd, (const struct sockaddr *)&srvaddr, sizeof(srvaddr));
    if (ret == -1)
    {
        perror("bind");
        return -1;
    }

    /* 3. 启动服务器监听客户端的连接请求 */
    ret = listen(listenfd, 1024);
    if (ret == -1)
    {
        perror("listen");
        return -1;
    }
    printf("listenfd = %d server init success\n", listenfd);

    /* 创建一个epoll实例 */
    epfd = epoll_create(512);
    if (epfd == -1)
    {
        perror("epoll_create");
        return -1;
    }

    /* 反应堆上树：将文件描述符listenfd的读事件以及回调函数添加到epoll实例中 */
    event.events = EPOLLIN;/* 事件 */
    my_ev = calloc(1, sizeof(struct my_event));
    my_ev->epfd = epfd;
    my_ev->fd = listenfd;/* 文件描述符 */
    my_ev->callback = accept_client;/* 回调函数 */
    event.data.ptr = my_ev;
    ret = epoll_ctl(epfd, EPOLL_CTL_ADD, listenfd, &event);
    if (ret == -1)
    {
        perror("epoll_ctl->EPOLL_CTL_ADD");
        return -1;
    }

    while(1)
    {
        /* 检测是否有资源准备就绪，准备就绪返回就绪的IO资源数 */
        count = epoll_wait(epfd, events, 512, 5000);
        if (count == -1)
        {
            perror("epoll_wait");
            return -1;
        }
        else if (count == 0)
        {
            fprintf(stderr, "epoll_wait timeout ...\n");
            continue;
        }
        for (i = 0; i < count; i++)
        {
            ((struct my_event *)(events[i].data.ptr))->callback((struct my_event *)(events[i].data.ptr));
        }
    }

    close(listenfd);
    close(epfd);
}