muduo库TcpServer模块详解

news2025/5/22 23:48:01

Muduo库核心模块——TcpServer

Muduo库的TcpServer模块是一个基于Reactor模式的高性能TCP服务端实现，负责管理监听端口、接受新连接、分发IO事件及处理连接生命周期。

一、核心组件与职责

Acceptor
- 监听指定端口，接受新连接，通过epoll监听listenfd的可读事件（表示新连接到达）。
- 内部封装socket和bind操作，支持SO_REUSEADDR选项避免端口占用。
- 新连接到达时，调用TcpServer的回调创建TcpConnection对象。
EventLoop
- 主事件循环（运行在主线程），处理Acceptor的事件及定时任务。
- 采用“one loop per thread”模型，确保线程安全，所有操作通过EventLoop的任务队列跨线程调度。
EventLoopThreadPool
- IO线程池，管理多个子EventLoop，提升并发处理能力。
- 新连接通过轮询或哈希策略分配到子EventLoop，实现负载均衡。
TcpConnection
- 封装一条TCP连接，管理send/recv缓冲区、处理读写事件及连接状态（连接中/已断开）。
- 通过Channel注册到所属EventLoop，监听socket的可读/可写事件。
回调机制
- 用户可设置的关键回调：
  - ConnectionCallback：连接建立或关闭时触发。
  - MessageCallback：接收到数据时调用，传递数据缓冲区及时间戳。
  - WriteCompleteCallback：数据发送完成时通知。

二、工作流程

启动服务器
- 用户调用TcpServer::start()，启动EventLoopThreadPool，Acceptor开始监听。
- 主EventLoop进入循环，等待事件。
接受新连接
- Acceptor监听到新连接，创建connfd，构造TcpConnection对象。
- 为新连接分配IO线程（从EventLoopThreadPool中选择一个子EventLoop）。
- 将TcpConnection的socket注册到子EventLoop的epoll中。
处理IO事件
- 读事件：子EventLoop触发可读事件，TcpConnection读取数据至缓冲区，调用用户MessageCallback。
- 写事件：发送缓冲区中的数据，写完成后触发WriteCompleteCallback。
- 错误处理：处理EPOLLERR等异常，关闭连接并通知用户。
连接关闭
- 主动关闭：调用TcpConnection::shutdown()，等待数据发送完毕。
- 被动关闭：接收到EOF或错误，触发handleClose()，移除连接并销毁资源。

三、线程安全与资源管理

跨线程调度
- 通过EventLoop::runInLoop()确保回调在正确的线程执行，避免竞态条件。
- 示例：主线程接受连接后，通过ioLoop->runInLoop()将TcpConnection添加到子线程。
连接生命周期
- 使用shared_ptr<TcpConnection>管理连接对象，引用计数确保回调期间对象存活。
- 连接关闭时，通过TcpServer::removeConnection()从连接列表移除，并延迟销毁（防止析构在回调中触发）。
资源释放
- TcpServer维护ConnectionMap跟踪所有活跃连接，析构时自动关闭所有连接。

四、性能优化策略

零拷贝优化
- 使用readv/writev分散读写，减少内存拷贝。
- 应用层缓冲区设计，避免频繁系统调用。
LT模式与边缘触发
- 采用epoll的LT（水平触发）模式，确保数据未处理时会持续通知，简化编程。
- 非阻塞IO配合EAGAIN处理，避免线程阻塞。
高效线程模型
- 主线程仅处理新连接，IO线程分担数据读写，充分发挥多核优势。

五、示例代码

#include <muduo/net/TcpServer.h>
#include <muduo/net/EventLoop.h>

void onConnection(const muduo::net::TcpConnectionPtr& conn) {
    if (conn->connected()) {
        printf("New connection: %s\n", conn->peerAddress().toIpPort().c_str());
    } else {
        printf("Connection closed: %s\n", conn->peerAddress().toIpPort().c_str());
    }
}

void onMessage(const muduo::net::TcpConnectionPtr& conn,
               muduo::net::Buffer* buf,
               muduo::Timestamp time) {
    printf("Received %zd bytes: %s\n", buf->readableBytes(), buf->retrieveAllAsString().c_str());
}

int main() {
    muduo::net::EventLoop loop;
    muduo::net::InetAddress listenAddr(8888);
    muduo::net::TcpServer server(&loop, listenAddr, "ExampleServer");
    
    server.setConnectionCallback(onConnection);
    server.setMessageCallback(onMessage);
    server.setThreadNum(4);  // 4 IO threads
    server.start();
    
    loop.loop();  // Block here
}