Linux epoll ET 模式下的 Reactor 网络服务器实现

协议与业务逻辑

TCP 是流式协议，存在粘包问题。我们通过自定义的报文格式（长度 + 数据）来解决。protocol.hpp 中定义了请求和响应的序列化/反序列化逻辑，使用 JSON 作为有效载荷。

#pragma once
#include <iostream>
#include <string>
#include <unistd.h>
#include <memory>
#include <jsoncpp/json/json.h>

namespace protocol_ns {
const std::string SEP = "\r\n";

std::string Encode(const std::string &json_str) {
    int json_str_len = json_str.size();
    std::string proto_str = std::to_string(json_str_len);
    proto_str += SEP;
    proto_str += json_str;
    proto_str += SEP;
    return proto_str;
}

std::string Decode(std::string &inbuffer) {
    auto pos = inbuffer.find(SEP);
    if (pos == std::string::npos) return std::string();
    std::string len_str = inbuffer.substr(0, pos);
    if (len_str.empty()) return std::string();
    int packlen = std::stoi(len_str);
    int total = packlen + len_str.size() + 2 * SEP.size();
    if (inbuffer.size() < total) return std::string();
    std::string package = inbuffer.substr(pos + SEP.size(), packlen);
    inbuffer.erase(0, total);
    return package;
}

class Request {
public:
    Request() {}
    Request(int x, int y, char oper) : _x(x), _y(y), _oper(oper) {}
    bool Serialize(std::string* out) {
        Json::Value root;
        root["x"] = _x;
        root["y"] = _y;
        root["oper"] = _oper;
        Json::FastWriter writer;
        *out = writer.write(root);
        return true;
    }
    bool DeSerialize(const std::string& in) {
        Json::Value root;
        Json::Reader reader;
        bool res = reader.parse(in, root);
        if (!res) return false;
        _x = root["x"].asInt();
        _y = root["y"].asInt();
        _oper = root["oper"].asInt();
        return true;
    }
public:
    int _x;
    int _y;
    char _oper;
};

class Response {
public:
    Response() {}
    Response(int result, int code) : _result(result), _code(code) {}
    bool Serialize(std::string *out) {
        Json::Value root;
        root["result"] = _result;
        root["code"] = _code;
        Json::FastWriter writer;
        *out = writer.write(root);
        return true;
    }
    bool Deserialize(const std::string &in) {
        Json::Value root;
        Json::Reader reader;
        bool res = reader.parse(in, root);
        if (!res) return false;
        _result = root["result"].asInt();
        _code = root["code"].asInt();
        return true;
    }
public:
    int _result;
    int _code;
};
}

连接管理与事件处理

Connection 类维护了单个连接的上下文，包括输入输出缓冲区。HandlerConnection 则负责具体的 IO 回调逻辑，如接收数据填充缓冲区、发送数据清空缓冲区等。这里需要特别注意非阻塞 IO 下的错误处理，例如 EWOULDBLOCK 表示当前无数据可读或缓冲区满，应暂停操作而非报错。

#pragma once
#include <iostream>
#include <string>
#include <functional>
#include "Reactor.hpp"
#include "InetAddr.hpp"
#include <unistd.h>

class Connection;
class Reactor;
using func_t = std::function<void(Connection *)>;

class Connection {
public:
    Connection(int sock) : _sock(sock), _R(nullptr) {}
    int Sockfd() { return _sock; }
    void SetEvents(int events) { _events = events; }
    uint32_t Events() { return _events; }
    void Register(func_t recver, func_t sender, func_t excepter) {
        _recver = recver;
        _sender = sender;
        _excepter = excepter;
    }
    void SetSelf(Reactor *R) { _R = R; }
    void AppendInBuffer(const std::string &buff) { _inbuffer += buff; }
    std::string& Inbuffer() { return _inbuffer; }
    void AppendOutBuffer(const std::string &buff) { _outbuffer += buff; }
    std::string &Outbuffer() { return _outbuffer; }
    bool OutbufferEmpty() { return _outbuffer.empty(); }
    void OutbufferRemove(int n) { _outbuffer.erase(0, n); }
    void Close() {
        if (_sock >= 0) ::close(_sock);
    }
    ~Connection() {}
    func_t _recver;
    func_t _sender;
    func_t _excepter;
    Reactor *_R;
private:
    int _sock;
    std::string _inbuffer;
    std::string _outbuffer;
    InetAddr _addr;
    uint32_t _events;
};

Epoller 封装

Epoller 类是对 Linux epoll 接口的轻量级封装，负责添加、修改和删除事件监听。在 ET 模式下，必须确保一次性读取完所有数据，否则后续不会再次通知，直到有新数据到达。

#pragma once
#include <iostream>
#include <sys/epoll.h>
#include <unistd.h>
#include "Log.hpp"
#include "Comm.hpp"

static const int gsize = 128;

class Epoller {
private:
    bool EventMethodCore(int fd, u_int32_t events, int type) {
        struct epoll_event ev;
        ev.events = events;
        ev.data.fd = fd;
        int n = ::epoll_ctl(_epfd, type, fd, &ev);
        if (n < 0) {
            LOG(ERROR, "epoll_ctl error!\n");
            return false;
        }
        LOG(DEBUG, "epoll_ctl add %d success!\n", fd);
        return true;
    }
public:
    Epoller() {
        _epfd = ::epoll_create(gsize);
        if (_epfd < 0) {
            LOG(FATAL, "epoll create error!\n");
            exit(EPOLL_CREATE_ERROR);
        }
        LOG(FATAL, "epoll create success, epfd: %d\n", _epfd);
    }

    bool AddEvent(int fd, uint32_t events) {
        return EventMethodCore(fd, events, EPOLL_CTL_ADD);
    }

    bool ModEvent(int fd, uint32_t events) {
        return EventMethodCore(fd, events, EPOLL_CTL_MOD);
    }

    bool DelEvent(int fd) {
        return ::epoll_ctl(_epfd, EPOLL_CTL_DEL, fd, nullptr);
    }

    int Wait(struct epoll_event revs[], int num, int timeout) {
        int n = ::epoll_wait(_epfd, revs, num, timeout);
        return n;
    }

    ~Epoller() {
        if (_epfd >= 0) ::close(_epfd);
    }
private:
    int _epfd;
};

Reactor 主循环

Reactor 是整个框架的核心，它维护着所有连接的映射表，并不断调用 epoll_wait 等待事件。一旦有事件就绪，便根据事件类型（读/写）调用对应的回调函数。在 ET 模式下，如果一次读取未读完，需要重新注册可写事件以便后续发送响应。

#pragma once
#include <iostream>
#include <unordered_map>
#include "Epoller.hpp"
#include "Connection.hpp"

class Reactor {
    const static int gnum = 64;
public:
    Reactor() : _isrunning(false) {}

    void AddConnection(int fd, uint32_t events, func_t recver, func_t sender, func_t excepter) {
        Connection *conn = new Connection(fd);
        conn->SetEvents(events);
        conn->Register(recver, sender, excepter);
        conn->SetSelf(this);
        _epller.AddEvent(conn->Sockfd(), conn->Events());
        _connections.insert(std::make_pair(conn->Sockfd(), conn));
    }

    bool ConnectionIsExists(int sockfd) {
        auto iter = _connections.find(sockfd);
        return iter != _connections.end();
    }

    void EnableReadWrite(int sockfd, bool readable, bool writeable) {
        uint32_t events = (readable ? EPOLLIN : 0) | (writeable ? EPOLLOUT : 0) | EPOLLET;
        if (ConnectionIsExists(sockfd)) {
            _connections[sockfd]->SetEvents(events);
            _epller.ModEvent(sockfd, events);
        }
    }

    void RemoveConnection(int sockfd) {
        if (!ConnectionIsExists(sockfd)) return;
        _epller.DelEvent(sockfd);
        _connections[sockfd]->Close();
        delete _connections[sockfd];
        _connections.erase(sockfd);
    }

    void LoopOnce(int timeout) {
        int n = _epller.Wait(revs, gnum, timeout);
        for (int i = 0; i < n; i++) {
            int sockfd = revs[i].data.fd;
            uint32_t revents = revs[i].events;
            if (revents & EPOLLHUP) revents |= (EPOLLIN | EPOLLOUT);
            if (revents & EPOLLERR) revents |= (EPOLLIN | EPOLLOUT);

            if (revents & EPOLLIN) {
                if (ConnectionIsExists(sockfd) && (_connections[sockfd]->_recver != nullptr)) {
                    _connections[sockfd]->_recver(_connections[sockfd]);
                }
            }
            if (revents & EPOLLOUT) {
                if (ConnectionIsExists(sockfd) && (_connections[sockfd]->_sender != nullptr)) {
                    _connections[sockfd]->_sender(_connections[sockfd]);
                }
            }
        }
    }

    void Dispatcher() {
        _isrunning = true;
        int timeout = 3000;
        while (_isrunning) {
            LoopOnce(timeout);
            Debug();
        }
        _isrunning = false;
    }

    void Debug() {
        std::cout << "------------------------------------" << std::endl;
        for (auto& connection : _connections) {
            std::cout << "fd : " << connection.second->Sockfd() << ", ";
            uint32_t events = connection.second->Events();
            if ((events & EPOLLIN) && (events & EPOLLET)) std::cout << "EPOLLIN | EPOLLET, ";
            if ((events & EPOLLOUT) && (events & EPOLLET)) std::cout << "EPOLLOUT | EPOLLET";
            std::cout << std::endl;
        }
        std::cout << "------------------------------------" << std::endl;
    }

    ~Reactor() {}

private:
    std::unordered_map<int, Connection *> _connections;
    struct epoll_event revs[gnum];
    Epoller _epller;
    bool _isrunning;
};

主程序入口

最后，我们在 main.cc 中组装各个组件。创建一个 Reactor 实例，初始化 Listener 监听端口，并将监听套接字注册到 Reactor 中。启动后，Reactor 将进入无限循环，处理所有连接的生命周期。

#include <iostream>
#include <memory>
#include "Reactor.hpp"
#include "Connection.hpp"
#include "Listener.hpp"
#include "PackageParse.hpp"
#include "HandlerConnection.hpp"
#include "Log.hpp"

int main(int argc, char* argv[]) {
    if (argc != 2) {
        std::cout << "Usage: " << argv[0] << " port" << std::endl;
        return 0;
    }
    uint16_t port = std::stoi(argv[1]);
    EnableScreen();
    std::unique_ptr<Reactor> react = std::make_unique<Reactor>();
    HandlerConnection hc(PackageParse::Parse);
    Listener listener(port, hc);
    react->AddConnection(listener.Sockfd(), EPOLLIN | EPOLLET,
                         std::bind(&Listener::Accepter, &listener, std::placeholders::_1),
                         nullptr, nullptr);
    react->Dispatcher();
}

该实现展示了如何从零构建一个基于 C++ 的高性能网络服务器。通过合理运用 epoll ET 模式和面向对象的设计，能够有效支撑高并发场景下的通信需求。