反应堆模型

Reactor 是一个基于epoll的高性能、低延迟、高并发的模型。

设计思路

• Reactor模式核心：使用单线程事件循环处理所有I/O事件，通过epoll管理文件描述符。
• 非阻塞IO：所有socket设置为非阻塞模式，配合epoll的边沿触发（ET）模式，确保高效事件处理。
• 事件分发：每个文件描述符关联一个事件处理器（EventHandler），由Reactor统一管理事件注册和回调。
• 资源管理：使用RAII确保资源自动释放，避免泄漏。

代码示例

#include <sys/epoll.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <fcntl.h>
#include <unistd.h>
#include <cstring>
#include <unordered_map>
#include <memory>
#include <stdexcept>
#include <iostream>
#include <vector>
#include <string>
#include <cerrno>

// 设置文件描述符为非阻塞模式
void set_non_blocking(int fd) {
    int flags = fcntl(fd, F_GETFL, 0);
    if (flags == -1) {
        throw std::runtime_error("fcntl F_GETFL failed");
    }
    if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -1) {
        throw std::runtime_error("fcntl F_SETFL failed");
    }
}

// 事件处理器接口
class EventHandler {
public:
    virtual ~EventHandler() = default;
    virtual void handle_event(uint32_t events) = 0;
};

class Reactor;

// Acceptor类，处理新连接
class Acceptor : public EventHandler {
public:
    Acceptor(Reactor& reactor, int port);
    ~Acceptor();
    void handle_event(uint32_t events) override;

private:
    Reactor& reactor_;
    int server_fd_;
};

// Connection类，处理客户端连接
class Connection : public EventHandler {
public:
    Connection(Reactor& reactor, int fd);
    ~Connection();
    void handle_event(uint32_t events) override;

private:
    Reactor& reactor_;
    int fd_;
    std::string output_buffer_;
    std::string input_buffer_;
};

// Reactor核心类
class Reactor {
public:
    Reactor();
    ~Reactor();

    void add_fd(int fd, uint32_t events, EventHandler* handler);
    void update_fd(int fd, uint32_t events);
    void del_fd(int fd);
    void run();
    void stop();

private:
    int epoll_fd_;
    std::unordered_map<int, EventHandler*> handlers_;
    bool running_ = true;

    friend class Acceptor;
    friend class Connection;
};

// Acceptor实现
Acceptor::Acceptor(Reactor& reactor, int port) : reactor_(reactor), server_fd_(-1) {
    server_fd_ = socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0);
    if (server_fd_ == -1) {
        throw std::runtime_error("socket creation failed");
    }

    int opt = 1;
    if (setsockopt(server_fd_, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) == -1) {
        close(server_fd_);
        throw std::runtime_error("setsockopt failed");
    }

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

    if (bind(server_fd_, (struct sockaddr*)&addr, sizeof(addr)) == -1) {
        close(server_fd_);
        throw std::runtime_error("bind failed");
    }

    if (listen(server_fd_, SOMAXCONN) == -1) {
        close(server_fd_);
        throw std::runtime_error("listen failed");
    }

    reactor_.add_fd(server_fd_, EPOLLIN | EPOLLET, this);
}

Acceptor::~Acceptor() {
    if (server_fd_ != -1) {
        close(server_fd_);
    }
}

void Acceptor::handle_event(uint32_t events) {
    if (events & EPOLLIN) {
        while (true) {
            struct sockaddr_in client_addr;
            socklen_t len = sizeof(client_addr);
            int client_fd = accept(server_fd_, (struct sockaddr*)&client_addr, &len);
            if (client_fd == -1) {
                if (errno == EAGAIN || errno == EWOULDBLOCK) {
                    break; // 所有新连接已处理
                } else {
                    std::cerr << "accept error: " << strerror(errno) << std::endl;
                    break;
                }
            }

            try {
                set_non_blocking(client_fd);
                reactor_.add_fd(client_fd, EPOLLIN | EPOLLET | EPOLLRDHUP, new Connection(reactor_, client_fd));
            } catch (const std::exception& e) {
                std::cerr << "Failed to add client: " << e.what() << std::endl;
                close(client_fd);
            }
        }
    }
}

// Connection实现
Connection::Connection(Reactor& reactor, int fd) : reactor_(reactor), fd_(fd) {}

Connection::~Connection() {
    if (fd_ != -1) {
        close(fd_);
    }
}

void Connection::handle_event(uint32_t events) {
    if (events & (EPOLLERR | EPOLLHUP | EPOLLRDHUP)) {
        reactor_.del_fd(fd_);
        return;
    }

    if (events & EPOLLIN) {
        char buffer[4096];
        ssize_t nread;
        while ((nread = read(fd_, buffer, sizeof(buffer))) > 0) {
            input_buffer_.append(buffer, nread);
        }

        if (nread == -1) {
            if (errno != EAGAIN && errno != EWOULDBLOCK) {
                reactor_.del_fd(fd_);
                return;
            }
        } else if (nread == 0) {
            reactor_.del_fd(fd_);
            return;
        }

        // 业务逻辑：将输入直接作为输出返回
        output_buffer_ = input_buffer_;
        input_buffer_.clear();
        reactor_.update_fd(fd_, EPOLLOUT | EPOLLET);
    }

    if (events & EPOLLOUT) {
        size_t total = output_buffer_.size();
        ssize_t nwritten = write(fd_, output_buffer_.data(), total);
        if (nwritten == -1) {
            if (errno != EAGAIN) {
                reactor_.del_fd(fd_);
            }
            return;
        }
        output_buffer_.erase(0, nwritten);
        if (output_buffer_.empty()) {
            reactor_.update_fd(fd_, EPOLLIN | EPOLLET);
        }
    }
}

// Reactor实现
Reactor::Reactor() {
    epoll_fd_ = epoll_create1(0);
    if (epoll_fd_ == -1) {
        throw std::runtime_error("epoll_create1 failed");
    }
}

Reactor::~Reactor() {
    stop();
    for (auto& pair : handlers_) {
        close(pair.first);
        delete pair.second;
    }
    close(epoll_fd_);
}

void Reactor::add_fd(int fd, uint32_t events, EventHandler* handler) {
    epoll_event ev;
    ev.events = events;
    ev.data.fd = fd;
    if (epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) == -1) {
        delete handler;
        throw std::runtime_error("epoll_ctl add failed: " + std::string(strerror(errno)));
    }
    handlers_[fd] = handler;
}

void Reactor::update_fd(int fd, uint32_t events) {
    epoll_event ev;
    ev.events = events;
    ev.data.fd = fd;
    if (epoll_ctl(epoll_fd_, EPOLL_CTL_MOD, fd, &ev) == -1) {
        throw std::runtime_error("epoll_ctl mod failed: " + std::string(strerror(errno)));
    }
}

void Reactor::del_fd(int fd) {
    auto it = handlers_.find(fd);
    if (it == handlers_.end()) return;

    delete it->second;
    handlers_.erase(it);
    epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr);
    close(fd);
}

void Reactor::run() {
    const int MAX_EVENTS = 64;
    epoll_event events[MAX_EVENTS];

    while (running_) {
        int nfds = epoll_wait(epoll_fd_, events, MAX_EVENTS, -1);
        if (nfds == -1) {
            if (errno == EINTR) continue;
            throw std::runtime_error("epoll_wait failed");
        }

        for (int i = 0; i < nfds; ++i) {
            int fd = events[i].data.fd;
            auto it = handlers_.find(fd);
            if (it != handlers_.end()) {
                it->second->handle_event(events[i].events);
            }
        }
    }
}

void Reactor::stop() {
    running_ = false;
}

int main() {
    try {
        Reactor reactor;
        Acceptor acceptor(reactor, 8080);
        reactor.run();
    } catch (const std::exception& e) {
        std::cerr << "Error: " << e.what() << std::endl;
        return 1;
    }
    return 0;
}