muduo 網路庫源碼剖析 (二)：TcpServer、Acceptor 與 TcpConnection 核心類詳解

muduo 伺服器端架構總覽

muduo 採用 **Reactor 模式** + **one loop per thread** 線程模型。伺服器端三大核心類：

• TcpServer：用戶介面類，管理完整生命週期
• Acceptor：監聽埠、接受新連線、回調 TcpServer
• TcpConnection：管理單一 TCP 連線的讀寫、狀態機

TcpServer：用戶層入口

class TcpServer : noncopyable {
public:
    using ConnectionCallback = std::function;
    using MessageCallback = std::function;
    using WriteCompleteCallback = std::function;
    
    TcpServer(EventLoop* loop, const InetAddress& listenAddr, const string& name);
    ~TcpServer();
    
    void start();  // 啟動監聽
    
    // 回調設定
    void setConnectionCallback(const ConnectionCallback& cb) { connectionCallback_ = cb; }
    void setMessageCallback(const MessageCallback& cb) { messageCallback_ = cb; }
    void setWriteCompleteCallback(const WriteCompleteCallback& cb) { writeCompleteCallback_ = cb; }
    
private:
    void newConnection(int sockfd, const InetAddress& peerAddr);  // Acceptor 回調
    
    EventLoop* loop_;                    // 主 Reactor 循環
    const string name_;
    std::unique_ptr acceptor_; // 監聽器
    ConnectionCallback connectionCallback_;
    MessageCallback messageCallback_;
    WriteCompleteCallback writeCompleteCallback_;
    std::map connections_;  // 連線管理
    int nextConnId_;                     // 連線 ID 生成器
};

關鍵流程：start() → newConnection()

void TcpServer::start() {
    // 必須在 loop 線程中執行
    loop_->runInLoop(std::bind(&Acceptor::listen, acceptor_.get()));
}

void TcpServer::newConnection(int sockfd, const InetAddress& peerAddr) {
    loop_->assertInLoopThread();
    
    // 1. 建立 TcpConnection
    char buf[32];
    snprintf(buf, sizeof(buf), "#%d", nextConnId_++);
    string connName = name_ + buf;
    
    InetAddress localAddr(sockets::getLocalAddr(sockfd));
    TcpConnectionPtr conn(new TcpConnection(loop_, connName, sockfd, localAddr, peerAddr));
    
    // 2. 註冊用戶回調
    conn->setConnectionCallback(connectionCallback_);
    conn->setMessageCallback(messageCallback_);
    conn->setWriteCompleteCallback(writeCompleteCallback_);
    
    // 3. 設定關閉回調 (從 connections_ 移除)
    conn->setCloseCallback(std::bind(&TcpServer::removeConnection, this, _1));
    
    // 4. 加入管理表
    connections_[connName] = conn;
    
    // 5. 從線程池取一個 I/O 線程，在其 EventLoop 中建立連線
    EventLoop* ioLoop = threadPool_->getNextLoop();
    ioLoop->runInLoop(std::bind(&TcpConnection::connectEstablished, conn));
}

Acceptor：連線接受器

class Acceptor : noncopyable {
public:
    using NewConnectionCallback = std::function;
    
    Acceptor(EventLoop* loop, const InetAddress& listenAddr, bool reuseport);
    ~Acceptor();
    
    void listen();  // 開始監聽
    void setNewConnectionCallback(const NewConnectionCallback& cb) { newConnectionCallback_ = cb; }
    
private:
    void handleRead();  // Channel 讀事件回調
    
    EventLoop* loop_;
    Socket acceptSocket_;           // 監聽 socket 封裝
    Channel acceptChannel_;         // 向 Poller 註冊讀事件
    NewConnectionCallback newConnectionCallback_;
    bool listening_;
};

監聽啟動流程

void Acceptor::listen() {
    loop_->assertInLoopThread();
    listening_ = true;
    acceptSocket_.listen();                    // 系統 listen()
    acceptChannel_.enableReading();            // 向 Poller 註冊 EPOLLIN
}

讀事件處理：接受新連線並回調

void Acceptor::handleRead() {
    loop_->assertInLoopThread();
    InetAddress peerAddr;
    int connfd = acceptSocket_.accept(&peerAddr);  // non-blocking accept
    if (connfd >= 0) {
        if (newConnectionCallback_) {
            newConnectionCallback_(connfd, peerAddr);  // 回調 TcpServer::newConnection
        } else {
            sockets::close(connfd);  // 無回調則關閉
        }
    }
}

TcpConnection：連線狀態機與 I/O 處理

class TcpConnection : noncopyable, public std::enable_shared_from_this {
public:
    using Ptr = std::shared_ptr;
    
    TcpConnection(EventLoop* loop, const string& name, int sockfd,
                  const InetAddress& localAddr, const InetAddress& peerAddr);
    ~TcpConnection();
    
    // 狀態機
    enum State { kConnecting, kConnected, kDisconnecting, kDisconnected };
    
    // 核心操作
    void connectEstablished();  // 在 I/O 線程中調用
    void connectDestroyed();    // 連線銷毀
    void send(const void* data, size_t len);
    void send(Buffer* buf);
    void shutdown();            // 發送 FIN
    
    // 回調設定
    void setConnectionCallback(const ConnectionCallback& cb) { connectionCallback_ = cb; }
    void setMessageCallback(const MessageCallback& cb) { messageCallback_ = cb; }
    void setWriteCompleteCallback(const WriteCompleteCallback& cb) { writeCompleteCallback_ = cb; }
    void setCloseCallback(const CloseCallback& cb) { closeCallback_ = cb; }
    
    const string& name() const { return name_; }
    const InetAddress& localAddress() const { return localAddr_; }
    const InetAddress& peerAddress() const { return peerAddr_; }
    bool connected() const { return state_ == kConnected; }
    
private:
    void handleRead(Timestamp receiveTime);   // Channel 讀事件
    void handleWrite();                       // Channel 寫事件
    void handleClose();                       // Channel 關閉事件
    void handleError();                       // Channel 錯誤事件
    
    EventLoop* loop_;
    const string name_;
    State state_;
    Socket socket_;                           // 封裝 sockfd
    Channel channel_;                         // 向 Poller 註冊事件
    InetAddress localAddr_;
    InetAddress peerAddr_;
    Buffer inputBuffer_;                      // 接收緩衝
    Buffer outputBuffer_;                     // 發送緩衝
    
    ConnectionCallback connectionCallback_;
    MessageCallback messageCallback_;
    WriteCompleteCallback writeCompleteCallback_;
    CloseCallback closeCallback_;
};

連線建立：connectEstablished()

void TcpConnection::connectEstablished() {
    loop_->assertInLoopThread();
    assert(state_ == kConnecting);
    state_ = kConnected;
    
    // 向 Poller 註冊讀事件
    channel_.setReadCallback(std::bind(&TcpConnection::handleRead, this, _1));
    channel_.enableReading();
    
    // 觸發用戶連線建立回調
    connectionCallback_(shared_from_this());
}

讀事件處理：handleRead()

void TcpConnection::handleRead(Timestamp receiveTime) {
    loop_->assertInLoopThread();
    int savedErrno = 0;
    ssize_t n = inputBuffer_.readFd(channel_.fd(), &savedErrno);
    
    if (n > 0) {
        // 有資料到達，觸發用戶訊息回調
        messageCallback_(shared_from_this(), &inputBuffer_, receiveTime);
    } else if (n == 0) {
        // 對端關閉 (收到 FIN)
        handleClose();
    } else {
        errno = savedErrno;
        handleError();
    }
}

完整伺服器端事件流向

用戶代碼                    muduo 內部
─────────────────────────────────────────────────────────────
TcpServer::start()
    → loop->runInLoop(Acceptor::listen)
        → Acceptor::listen()
            → acceptSocket.listen()
            → acceptChannel.enableReading()
                → EventLoop::updateChannel()
                    → Poller::updateChannel()  [向 epoll 註冊 EPOLLIN]

[等待連線...]

Poller::poll() 發現 acceptChannel 可讀
    → EventLoop::loop()
        → Channel::handleEvent()
            → Acceptor::handleRead()
                → acceptSocket.accept() 獲取 connfd
                → newConnectionCallback_(connfd, peerAddr)
                    → TcpServer::newConnection()
                        → new TcpConnection(ioLoop, ...)
                        → ioLoop->runInLoop(bind(&TcpConnection::connectEstablished))
                            → EventLoop::queueInLoop()
                                → EventLoop::doPendingFunctors()
                                    → TcpConnection::connectEstablished()
                                        → channel.enableReading()  [向 I/O 線程 Poller 註冊]
                                        → connectionCallback_(conn)  [用戶 onConnection]

[連線建立完成，進入訊息處理迴路]

設計精華總結