C++ 仿 muduo 库高并发服务器 TcpServer 模块实现

• 作用：从 Reactor 线程池（Sub Loops）。
• 用途：管理一组 IO 线程，每个线程运行一个 EventLoop。当 _baseloop 接收到新连接后，会通过 _pool 选择一个 IO 线程，将新连接交给它管理。

std::unordered_map<uint64_t, PtrConnection> _conns

• 作用：连接容器。
• 用途：保存服务器当前所有活跃连接的智能指针 (shared_ptr)。
• 关键点：这是服务器持有连接生命周期的地方。只要连接在这个 map 中，它的引用计数就不为 0，就不会被析构。当连接关闭时，从这里移除，引用计数归零，连接自动销毁。

回调函数对象 (_connected_callback, _message_callback, _closed_callback, _event_callback)

• 作用：保存用户注册的业务逻辑。
• 用途：当新连接创建时，这些回调会被传递给 Connection 对象。

成员函数

void NewConnection(int fd)

• 触发时机：当 Acceptor 检测到新连接请求并成功执行 accept 后调用。
• 逻辑流程：
  1. _next_id++：生成新连接 ID。
  2. _pool.NextLoop()：从线程池中轮询选出一个 EventLoop（负载均衡）。
  3. new Connection(...)：创建新连接对象，绑定到选定的 Loop 和 fd 上。
  4. 配置连接：将用户设置的各种回调函数（OnMessage, OnClose 等）传递给这个新连接。
  5. 设置内部关闭回调：这是关键一步，当连接断开时，Connection 对象会回调这个函数，通知 TcpServer 从 _conns 中移除自己。
  6. 启动超时销毁：如果 _enable_inactive_release 为真，开启连接的定时器。
  7. 初始化：调用 conn->Established()，正式开启读事件监控。
  8. 存储：_conns.insert(...)，将连接放入 map 中管理。

void RemoveConnection(const PtrConnection& conn)

• 触发时机：当 Connection 检测到连接断开（对端关闭或错误）并执行清理流程时调用。
• 作用：它只是一个跳板，为了线程安全，它调用 _baseloop.RunInLoop 将实际的删除操作调度到主线程执行。

void RemoveConnectionInLoop(const PtrConnection& conn)

• 执行环境：主线程 (_baseloop)。
• 逻辑：根据连接 ID 从 _conns map 中删除对应的 Connection 对象。
• 结果：shared_ptr 引用计数减 1。如果此时没有其他地方引用该连接（通常情况），连接对象就会被析构，从而释放 Socket 资源。

void RunAfterInLoop(const Functor task, int delay)

• 作用：在主循环中添加定时任务的辅助函数。

TcpServer(int port)

• 构造函数：
  1. 初始化端口。
  2. 初始化 _acceptor，传入 _baseloop 和端口。
  3. 初始化 _pool，传入 _baseloop。
  4. 关键绑定：_acceptor.SetAcceptCallback(...)，将 TcpServer::NewConnection 绑定为接受连接的回调。
  5. _acceptor.Listen()：立即开始监听端口（此时还没有开始事件循环）

void SetThreadCount(int count)

• 作用：设置 IO 线程的数量。
• 说明：
  • count = 0：单线程模式。所有连接的 IO 操作都在主线程 (_baseloop) 中处理。
  • count > 0：多线程模式。主线程只负责 Accept，count 个子线程负责处理连接 IO。

void Start()

• 作用：启动服务器。
• 逻辑：
  1. _pool.Create()：创建并启动所有 IO 线程。
  2. _baseloop.Start()：主线程开始进入 epoll_wait 事件循环，正式开始工作。这是一个死循环，函数调用后不会返回。

源代码

class TcpServer {
private:
    uint64_t _next_id; // 自动增长的连接 ID
    int _port;
    int _timeout; // 超时时间
    bool _enable_inactive_release; // 是否启用非活跃销毁
    EventLoop _baseloop; // 主线程的 EventLoop 对象，负责监听事件的处理
    Acceptor _acceptor; // 监听套接字管理对象
    LoopThreadPool _pool; // 从属 EventLoop 线程池
    std::unordered_map<uint64_t, PtrConnection> _conns; // 保存管理所有连接对应的 shared_ptr 对象

private:
    using ConnectedCallback = std::function<void(const PtrConnection&)>;
    using MessageCallback = std::function<void(const PtrConnection&, Buffer*)>;
    using ClosedCallback = std::function<void(const PtrConnection&)>;
    using AnyEventCallback = std::function<void(const PtrConnection&)>;
    using Functor = std::function<void()>;

    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _closed_callback;
    AnyEventCallback _event_callback;

    // 为新连接构造一个 Connnection 进行管理
    void NewConnection(int fd) {
        _next_id++;
        PtrConnection conn(new Connection(_pool.NextLoop(), _next_id, fd));
        conn->SetMessageCallback(_message_callback);
        conn->SetClosedCallback(_closed_callback);
        conn->SetConnectedCallback(_connected_callback);
        conn->SetAnyEventCallback(_event_callback);
        conn->SetSrvClosedCallback(std::bind(&TcpServer::RemoveConnection, this, conn));
        if (_enable_inactive_release) conn->EnableInactiveRelease(_timeout); // 启动非活跃超时销毁
        conn->Established(); // 就绪初始化
        _conns.insert(std::make_pair(_next_id, conn));
    }

    void RemoveConnectionInLoop(const PtrConnection& conn) {
        uint64_t id = conn->Id();
        auto it = _conns.find(id);
        if (it != _conns.end()) {
            _conns.erase(it);
        }
    }

    // 从_conns 中移除连接信息
    void RemoveConnection(const PtrConnection& conn) {
        _baseloop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop, this, conn));
    }

    void RunAfterInLoop(const Functor task, int delay) {
        _next_id++;
        _baseloop.TimerAdd(_next_id, delay, task);
    }

public:
    TcpServer(int port) : _port(port), _next_id(0), _enable_inactive_release(false), _acceptor(&_baseloop, port), _pool(&_baseloop) {
        _acceptor.SetAcceptCallback(std::bind(&TcpServer::NewConnection, this, std::placeholders::_1));
        _acceptor.Listen(); // 将监听套接字挂在 baseloop 上
    }

    // 设置线程池数量
    void SetThreadCount(int count) {
        _pool.SetThreadCount(count);
    }

    void SetConnectedCallback(const ConnectedCallback& cb) {
        _connected_callback = cb;
    }

    void SetMessageCallback(const MessageCallback& cb) {
        _message_callback = cb;
    }

    void SetClosedCallback(const ClosedCallback& cb) {
        _closed_callback = cb;
    }

    void SetAnyEventCallback(const AnyEventCallback& cb) {
        _event_callback = cb;
    }

    // 添加定时任务
    void RunAfter(const Functor task, int delay) {
        _baseloop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop, this, task, delay));
    }

    // 启动非活跃销毁
    void EnableInactiveRelease(int timeout) {
        _timeout = timeout;
        _enable_inactive_release = true;
    }

    void Start() {
        _pool.Create();
        _baseloop.Start();
    }
};