epoll + 线程池c++例程

好的,我来为您介绍一个使用epoll和线程池的C++服务器例程:

#include <iostream>
#include <vector>
#include <unordered_map>
#include <thread>
#include <mutex>
#include <queue>
#include <functional>
#include <unistd.h>
#include <fcntl.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <netinet/in.h>

const int MAX_EVENTS = 10;
const int PORT = 8080;

class ThreadPool {
public:
    ThreadPool(int numThreads) : stop(false) {
        for(int i = 0; i < numThreads; ++i) {
            workers.emplace_back([this]() {
                while(true) {
                    std::function<void()> task;
                    {
                        std::unique_lock<std::mutex> lock(this->queue_mutex);
                        this->condition.wait(lock, [this] { return this->stop || !this->tasks.empty(); });
                        if(this->stop && this->tasks.empty()) return;
                        task = std::move(this->tasks.front());
                        this->tasks.pop();
                    }
                    task();
                }
            });
        }
    }

    void enqueue(std::function<void()> task) {
        {
            std::unique_lock<std::mutex> lock(queue_mutex);
            if(stop) throw std::runtime_error("enqueue on stopped ThreadPool");
            tasks.push(task);
        }
        condition.notify_one();
    }

    ~ThreadPool() {
        {
            std::unique_lock<std::mutex> lock(queue_mutex);
            stop = true;
        }
        condition.notify_all();
        for(std::thread &worker: workers)
            worker.join();
    }

private:
    std::vector<std::thread> workers;
    std::queue<std::function<void()>> tasks;
    std::mutex queue_mutex;
    std::condition_variable condition;
    bool stop;
};

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

int main() {
    int serverFd = socket(AF_INET, SOCK_STREAM, 0);
    setNonBlocking(serverFd);

    struct sockaddr_in serverAddr;
    memset(&serverAddr, 0, sizeof(serverAddr));
    serverAddr.sin_family = AF_INET;
    serverAddr.sin_addr.s_addr = INADDR_ANY;
    serverAddr.sin_port = htons(PORT);

    bind(serverFd, (struct sockaddr*)&serverAddr, sizeof(serverAddr));
    listen(serverFd, SOMAXCONN);

    int epollFd = epoll_create1(0);
    struct epoll_event event;
    event.data.fd = serverFd;
    event.events = EPOLLIN | EPOLLET;
    epoll_ctl(epollFd, EPOLL_CTL_ADD, serverFd, &event);

    ThreadPool pool(4);

    std::unordered_map<int, std::string> clients;

    while(true) {
        struct epoll_event events[MAX_EVENTS];
        int n = epoll_wait(epollFd, events, MAX_EVENTS, -1);
        for(int i = 0; i < n; ++i) {
            if(events[i].data.fd == serverFd) {
                while(true) {
                    struct sockaddr_in clientAddr;
                    socklen_t clientAddrLen = sizeof(clientAddr);
                    int clientFd = accept(serverFd, (struct sockaddr*)&clientAddr, &clientAddrLen);
                    if(clientFd == -1) break;
                    setNonBlocking(clientFd);
                    event.data.fd = clientFd;
                    event.events = EPOLLIN | EPOLLET;
                    epoll_ctl(epollFd, EPOLL_CTL_ADD, clientFd, &event);
                }
            } else {
                pool.enqueue([&clients, epollFd, events, i]() {
                    char buf[1024];
                    int clientFd = events[i].data.fd;
                    int n = read(clientFd, buf, sizeof(buf));
                    if(n <= 0) {
                        epoll_ctl(epollFd, EPOLL_CTL_DEL, clientFd, NULL);
                        close(clientFd);
                        clients.erase(clientFd);
                    } else {
                        clients[clientFd] += std::string(buf, n);
                        event.data.fd = clientFd;
                        event.events = EPOLLOUT | EPOLLET;
                        epoll_ctl(epollFd, EPOLL_CTL_MOD, clientFd, &event);
                    }
                });
            }
        }
    }

    return 0;
}

这个例程展示了如何使用epoll和线程池来实现一个简单的C++服务器。主要特点如下:

1. 使用epoll实现事件驱动,处理多个客户端连接。

2. 使用线程池来处理客户端请求,提高并发处理能力。

3. 使用非阻塞IO,配合epoll的边缘触发模式。

4. 使用unordered_map存储客户端数据。

5. 处理客户端连接、读取和写入事件。

这个例程可以作为构建更复杂服务器的基础。在实际应用中,你可能需要添加更多的错误处理、协议解析等功能。