epoll 多线程实例

epoll 是 Linux 系统提供的一种高性能的事件驱动 I/O 模型，通过它可以实现多并发连接的管理。多线程的 epoll 实例就是在使用 epoll 的同时，使用多个线程来处理连接。

在多线程的 epoll 实例中，首先需要创建一个 epoll 实例，并将监听的文件描述符添加到其中。接着，创建多个工作线程，每个线程都负责处理一部分连接。在每个线程中，通过 epoll_wait 函数等待事件的发生，并获取到发生事件的文件描述符。然后，根据不同的事件类型进行相应的处理，如读取、写入等。

为了保证多线程之间的线程安全，可以使用互斥锁来保护共享资源，如 epoll 实例以及与连接相关的数据结构。当一个线程对共享资源进行操作时，需要先获取互斥锁，操作完成后再释放锁，以避免多线程竞争导致的数据不一致问题。

多线程的 epoll 实例可以提高处理并发连接的效率。通过将连接分配给不同的线程处理，可以充分利用多核 CPU 的优势，提高系统的吞吐量。此外，使用 epoll 的事件驱动模型，可以避免线程阻塞，进一步提高系统的并发处理能力。

但是需要注意，多线程的 epoll 实例也存在一些问题。例如，线程之间的同步和协调可能会带来一些开销，对于处理短连接的场景，线程创建和销毁的开销可能会比较大。此外，在使用多线程的情况下，还需要考虑线程安全问题，避免竞争导致的数据不一致。

总之，多线程的 epoll 实例是一种高效的并发处理模型，通过充分利用多核 CPU 和事件驱动的特性，可以提高系统的并发性能，适用于处理大量并发连接的场景。

相关问题

linux epoll send recv 实例

下面是一个使用 epoll 实现的简单的 TCP 服务器和客户端的例子。服务器和客户端都是非阻塞的，使用 epoll 监听和处理文件描述符事件。其中，服务器使用了多线程技术，每个线程负责处理一个客户端连接。

服务器代码：

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <errno.h>
#include <signal.h>
#include <netinet/in.h>
#include <sys/epoll.h>
#include <pthread.h>

#define MAX_EVENTS 1024
#define BUF_SIZE 1024
#define THREAD_NUM 4

int listen_fd;
int epoll_fd;
pthread_t thread_pool[THREAD_NUM];

void setnonblocking(int fd) {
    int flags = fcntl(fd, F_GETFL, 0);
    fcntl(fd, F_SETFL, flags | O_NONBLOCK);
}

void add_event(int fd, int events) {
    struct epoll_event ev;
    ev.events = events;
    ev.data.fd = fd;
    epoll_ctl(epoll_fd, EPOLL_CTL_ADD, fd, &ev);
}

void del_event(int fd, int events) {
    struct epoll_event ev;
    ev.events = events;
    ev.data.fd = fd;
    epoll_ctl(epoll_fd, EPOLL_CTL_DEL, fd, &ev);
}

void mod_event(int fd, int events) {
    struct epoll_event ev;
    ev.events = events;
    ev.data.fd = fd;
    epoll_ctl(epoll_fd, EPOLL_CTL_MOD, fd, &ev);
}

void *worker(void *arg) {
    int client_fd = *(int*)arg;
    char buf[BUF_SIZE];
    int nread;

    while (1) {
        nread = recv(client_fd, buf, BUF_SIZE, 0);
        if (nread == -1) {
            if (errno == EAGAIN) {
                printf("recv EAGAIN\n");
                break;
            } else {
                perror("recv");
                break;
            }
        } else if (nread == 0) {
            printf("client close\n");
            break;
        } else {
            printf("recv: %s\n", buf);
            if (send(client_fd, buf, nread, 0) == -1) {
                perror("send");
                break;
            }
        }
    }

    close(client_fd);
    pthread_exit(NULL);
}

void *accepter(void *arg) {
    struct sockaddr_in client_addr;
    socklen_t client_len = sizeof(client_addr);
    int client_fd;
    int i;

    while (1) {
        client_fd = accept(listen_fd, (struct sockaddr*)&client_addr, &client_len);
        if (client_fd == -1) {
            if (errno == EAGAIN) {
                printf("accept EAGAIN\n");
                continue;
            } else {
                perror("accept");
                break;
            }
        } else {
            setnonblocking(client_fd);
            add_event(client_fd, EPOLLIN|EPOLLET);
            printf("new client: %d\n", client_fd);
        }
    }

    close(listen_fd);
    pthread_exit(NULL);
}

int main(int argc, char *argv[]) {
    struct sockaddr_in server_addr;
    int port = 8888;
    struct epoll_event events[MAX_EVENTS];
    int nready;
    int i, j;

    if (argc > 1) {
        port = atoi(argv[1]);
    }

    listen_fd = socket(AF_INET, SOCK_STREAM, 0);
    if (listen_fd == -1) {
        perror("socket");
        exit(1);
    }

    setnonblocking(listen_fd);

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

    if (bind(listen_fd, (struct sockaddr*)&server_addr, sizeof(server_addr)) == -1) {
        perror("bind");
        exit(1);
    }

    if (listen(listen_fd, 1024) == -1) {
        perror("listen");
        exit(1);
    }

    epoll_fd = epoll_create(MAX_EVENTS);
    if (epoll_fd == -1) {
        perror("epoll_create");
        exit(1);
    }

    add_event(listen_fd, EPOLLIN|EPOLLET);

    for (i = 0; i < THREAD_NUM; i++) {
        if (pthread_create(&thread_pool[i], NULL, accepter, NULL) == -1) {
            perror("pthread_create");
            exit(1);
        }
    }

    while (1) {
        nready = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);
        if (nready == -1) {
            if (errno == EINTR) {
                continue;
            } else {
                perror("epoll_wait");
                break;
            }
        }

        for (i = 0; i < nready; i++) {
            if (events[i].data.fd == listen_fd) {
                continue;
            }

            if (events[i].events & EPOLLIN) {
                pthread_t thread_id;
                int client_fd = events[i].data.fd;
                del_event(client_fd, EPOLLIN|EPOLLET);
                if (pthread_create(&thread_id, NULL, worker, (void*)&client_fd) == -1) {
                    perror("pthread_create");
                }
            }
        }
    }

    for (i = 0; i < THREAD_NUM; i++) {
        if (pthread_join(thread_pool[i], NULL) == -1) {
            perror("pthread_join");
        }
    }

    close(epoll_fd);

    return 0;
}

客户端代码：

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <errno.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/epoll.h>

#define MAX_EVENTS 1024
#define BUF_SIZE 1024

void setnonblocking(int fd) {
    int flags = fcntl(fd, F_GETFL, 0);
    fcntl(fd, F_SETFL, flags | O_NONBLOCK);
}

void add_event(int fd, int events) {
    struct epoll_event ev;
    ev.events = events;
    ev.data.fd = fd;
    epoll_ctl(epoll_fd, EPOLL_CTL_ADD, fd, &ev);
}

void del_event(int fd, int events) {
    struct epoll_event ev;
    ev.events = events;
    ev.data.fd = fd;
    epoll_ctl(epoll_fd, EPOLL_CTL_DEL, fd, &ev);
}

void mod_event(int fd, int events) {
    struct epoll_event ev;
    ev.events = events;
    ev.data.fd = fd;
    epoll_ctl(epoll_fd, EPOLL_CTL_MOD, fd, &ev);
}

int main(int argc, char *argv[]) {
    struct sockaddr_in server_addr;
    char *ip = "127.0.0.1";
    int port = 8888;
    int sockfd;
    struct epoll_event events[MAX_EVENTS];
    int nready;
    char buf[BUF_SIZE];
    int nread, nwrite;
    int i;

    if (argc > 1) {
        ip = argv[1];
    }

    if (argc > 2) {
        port = atoi(argv[2]);
    }

    sockfd = socket(AF_INET, SOCK_STREAM, 0);
    if (sockfd == -1) {
        perror("socket");
        exit(1);
    }

    setnonblocking(sockfd);

    memset(&server_addr, 0, sizeof(server_addr));
    server_addr.sin_family = AF_INET;
    server_addr.sin_addr.s_addr = inet_addr(ip);
    server_addr.sin_port = htons(port);

    if (connect(sockfd, (struct sockaddr*)&server_addr, sizeof(server_addr)) == -1) {
        if (errno != EINPROGRESS) {
            perror("connect");
            exit(1);
        }
    }

    epoll_fd = epoll_create(MAX_EVENTS);
    if (epoll_fd == -1) {
        perror("epoll_create");
        exit(1);
    }

    add_event(sockfd, EPOLLOUT);

    while (1) {
        nready = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);
        if (nready == -1) {
            if (errno == EINTR) {
                continue;
            } else {
                perror("epoll_wait");
                break;
            }
        }

        for (i = 0; i < nready; i++) {
            if (events[i].events & EPOLLOUT) {
                del_event(sockfd, EPOLLOUT);
                printf("connected\n");
                break;
            }
        }
    }

    while (1) {
        scanf("%s", buf);
        if (strcmp(buf, "quit") == 0) {
            break;
        }
        nwrite = send(sockfd, buf, strlen(buf), 0);
        if (nwrite == -1) {
            perror("send");
            break;
        }
        nread = recv(sockfd, buf, BUF_SIZE, 0);
        if (nread == -1) {
            if (errno == EAGAIN) {
                printf("recv EAGAIN\n");
                continue;
            } else {
                perror("recv");
                break;
            }
        } else if (nread == 0) {
            printf("server close\n");
            break;
        } else {
            printf("recv: %s\n", buf);
        }
    }

    close(sockfd);
    close(epoll_fd);

    return 0;
}

NIO多路复用epoll

NIO（Non-blocking I/O）是一种非阻塞的 I/O 模型，通过使用单线程管理多个并发的连接，实现了高效的 I/O 处理。而多路复用是 NIO 实现非阻塞的关键技术之一，其中 epoll 是 Linux 内核提供的一种高效的多路复用机制。

在 NIO 中，可以使用 select 或 epoll 函数实现多路复用。select 是一种较为简单的多路复用机制，它能够同时监听多个文件描述符，并在这些文件描述符中有可读、可写或异常事件发生时进行通知。而 epoll 是 select 的一种改进，它通过注册事件的方式，只关注发生事件的文件描述符，避免了遍历全部文件描述符的开销。

epoll 使用三个系统调用来操作事件：

1. epoll_create：创建一个 epoll 实例，返回一个文件描述符。
2. epoll_ctl：注册、修改或删除事件。
3. epoll_wait：等待事件发生，返回发生事件的文件描述符列表。

与 select 相比，epoll 有以下优势：
- 无需遍历全部文件描述符，只关注发生事件的文件描述符，提高了效率。
- 支持水平触发和边缘触发两种模式，可以根据需求选择适当的模式。
- 支持更大的并发连接数。

在 Redis 中，它使用 epoll 和多个 I/O 线程来处理客户端请求。epoll 负责监听客户端的连接和数据事件，并将事件交给 I/O 线程进行处理，这样可以充分发挥多核 CPU 的优势，提高了 Redis 服务器的性能。