c++怎么实现一个Reactor网络模型_C++网络编程中Reactor事件模型实现-C++-PHP中文网

c++怎么实现一个Reactor网络模型_C++网络编程中Reactor事件模型实现

下次还敢

发布： 2025-10-31 16:39:02

原创

283人浏览过

Reactor模型通过事件循环监听多个socket，使用epoll实现I/O多路复用，核心组件包括事件分发器、反应器和事件处理器，结合示例展示了高并发服务器中事件注册与回调处理机制。

c++怎么实现一个reactor网络模型_c++网络编程中reactor事件模型实现

在C++网络编程中，Reactor模型是一种高效的I/O多路复用设计模式，适用于高并发的服务器程序。它通过一个事件循环监听多个文件描述符（如socket），当某个描述符就绪时（可读、可写或出错），触发对应的事件处理器进行处理。

Reactor模型核心组件

一个典型的Reactor模型包含以下几个关键部分：

EventDemultiplexer（事件多路分发器）：通常是epoll（Linux）、kqueue（BSD/macOS）或select/poll，用于监听多个socket的事件。
Reactor（反应器）：运行事件循环，调用事件分发器等待事件，并将就绪的事件分发给注册的事件处理器。
EventHandler（事件处理器）：用户定义的回调对象，包含handle_read、handle_write等接口，由Reactor在事件发生时调用。
ConcreteEventHandler：具体实现，比如处理客户端连接、接收数据等。

使用epoll实现简易Reactor

下面是一个基于Linux epoll 的简化版Reactor模型实现，展示基本结构和流程。

天工大模型

中国首个对标ChatGPT的双千亿级大语言模型

115

查看详情

// reactor.h

#include <sys/epoll.h>
#include <functional>
#include <unordered_map>
#include <vector>

class Reactor {
public:
    using EventCallback = std::function<void(int)>;

    Reactor();
    ~Reactor();

    // 注册可读事件
    void register_read(int fd, EventCallback cb);
    // 注册可写事件
    void register_write(int fd, EventCallback cb);
    // 移除事件
    void remove(int fd);

    // 事件循环
    void run();

private:
    int epoll_fd_;
    std::unordered_map<int, EventCallback> read_callbacks_;
    std::unordered_map<int, EventCallback> write_callbacks_;
};

登录后复制

// reactor.cpp

#include "reactor.h"
#include <iostream>
#include <unistd.h>
#include <fcntl.h>

Reactor::Reactor() {
    epoll_fd_ = epoll_create1(0);
    if (epoll_fd_ == -1) {
        perror("epoll_create1");
        exit(1);
    }
}

Reactor::~Reactor() {
    close(epoll_fd_);
}

void Reactor::register_read(int fd, EventCallback cb) {
    read_callbacks_[fd] = cb;

    struct epoll_event ev;
    ev.events = EPOLLIN | EPOLLET;  // 边缘触发
    ev.data.fd = fd;
    epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev);
}

void Reactor::register_write(int fd, EventCallback cb) {
    write_callbacks_[fd] = cb;

    struct epoll_event ev;
    ev.events = EPOLLOUT | EPOLLET;
    ev.data.fd = fd;
    epoll_ctl(epoll_fd_, EPOLL_CTL_MOD, fd, &ev);
}

void Reactor::remove(int fd) {
    read_callbacks_.erase(fd);
    write_callbacks_.erase(fd);
    epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr);
    close(fd);
}

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

    while (true) {
        int n = epoll_wait(epoll_fd_, events, MAX_EVENTS, -1);
        if (n == -1) {
            perror("epoll_wait");
            break;
        }

        for (int i = 0; i < n; ++i) {
            int fd = events[i].data.fd;
            uint32_t events_mask = events[i].events;

            if (events_mask & EPOLLIN) {
                auto it = read_callbacks_.find(fd);
                if (it != read_callbacks_.end()) {
                    it->second(fd);  // 调用读回调
                }
            }
            if (events_mask & EPOLLOUT) {
                auto it = write_callbacks_.find(fd);
                if (it != write_callbacks_.end()) {
                    it->second(fd);  // 调用写回调
                }
                // 写事件通常处理完后移除，避免反复触发
                epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr);
                write_callbacks_.erase(fd);
            }
            if (events_mask & (EPOLLERR | EPOLLHUP)) {
                remove(fd);
            }
        }
    }
}

登录后复制

简单TCP服务器示例

使用上面的Reactor创建一个回显服务器：

立即学习“C++免费学习笔记（深入）”；

#include "reactor.h"
#include <sys/socket.h>
#include <netinet/in.h>
#include <cstring>

int create_server_socket(int port) {
    int sockfd = socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0);
    if (sockfd == -1) {
        perror("socket");
        exit(1);
    }

    int opt = 1;
    setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));

    sockaddr_in addr{};
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = INADDR_ANY;
    addr.sin_port = htons(port);

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

    if (listen(sockfd, SOMAXCONN) == -1) {
        perror("listen");
        exit(1);
    }

    return sockfd;
}

int main() {
    Reactor reactor;
    int server_fd = create_server_socket(8888);

    // 监听新连接
    reactor.register_read(server_fd, [&reactor](int fd) {
        while (true) {
            sockaddr_in client_addr{};
            socklen_t len = sizeof(client_addr);
            int client_fd = accept4(fd, (struct sockaddr*)&client_addr, &len, SOCK_NONBLOCK);
            if (client_fd == -1) break;

            printf("Client connected: %d\n", client_fd);

            // 注册客户端可读事件
            reactor.register_read(client_fd, [&reactor](int cfd) {
                char buf[1024];
                ssize_t n = read(cfd, buf, sizeof(buf));
                if (n > 0) {
                    write(cfd, buf, n);  // 回显
                } else {
                    reactor.remove(cfd); // 连接关闭或出错
                }
            });
        }
    });

    printf("Server running on port 8888...\n");
    reactor.run();

    return 0;
}

登录后复制