8. TcpServer

约 3413 字大约 11 分钟

2026-04-16

TcpServer 简介

TcpServer 类可以理解为 muduo 对外提供的服务器总入口，它负责把“监听新连接、管理连接、分发连接、关闭连接”这些事情串起来。

如果把 muduo 服务器看成一套流水线，那么：

Acceptor 负责在主线程监听新连接；
EventLoopThreadPool 负责把连接分配给多个 subLoop；
TcpConnection 负责一条具体连接的读写和事件处理；
TcpServer 则负责把这些模块组织起来，对外提供一个统一的服务器编程接口。

它的核心职责可以概括为：

启动服务器监听；
接受新连接；
把新连接分配给某个 EventLoop；
创建并保存 TcpConnection；
在连接关闭时把连接从服务器中移除；
对外提供连接回调、消息回调、写完成回调等接口。

thread(main)
 └── TcpServer
      ├── mainLoop
      │    ├── Poller
      │    │    ├── wakeupChannel
      │    │    └── acceptChannel
      │    └── Acceptor
      │         └── Socket(listenfd)
      │
      ├── EventLoopThreadPool
      │    ├── subThread1 -> subLoop1 -> Poller -> Channel(connfd...)
      │    ├── subThread2 -> subLoop2 -> Poller -> Channel(connfd...)
      │    └── subThread3 -> subLoop3 -> Poller -> Channel(connfd...)
      │
      └── connections_
           ├── TcpConnection1
           ├── TcpConnection2
           └── TcpConnection3

TcpServer 类重要的成员变量

1. 重要成员

EventLoop *loop_：服务器使用的 baseLoop，通常运行在主线程。
TcpServer 本身的大部分管理逻辑都依赖这个主事件循环来执行。
const std::string ipPort_：监听地址的字符串形式，比如 "0.0.0.0:8000"。
主要用于日志输出和连接命名。
const std::string name_：服务器名字。
用于标识当前 TcpServer 对象，日志里经常会用到。
std::unique_ptr<Acceptor> acceptor_：负责监听新连接的接收器。
它运行在 mainLoop 中，专门负责处理 listenfd 上的可读事件。
std::shared_ptr<EventLoopThreadPool> threadPool_：事件循环线程池。
用来管理多个 subLoop，实现 one loop per thread 的连接分发模型。
ConnectionCallback connectionCallback_：有新连接建立时的回调。
通常由用户设置，用来处理连接建立和连接断开时的通知。
MessageCallback messageCallback_：当连接上收到消息时的回调。
这个回调通常在 TcpConnection::handleRead() 中触发。
WriteCompleteCallback writeCompleteCallback_：当消息发送完成后的回调。
常用于发送完一批数据后做后续处理。
ThreadInitCallback threadInitCallback_：线程池中每个 EventLoop 初始化时的回调。
用来在 subLoop 启动时做一些自定义初始化工作。
int numThreads_：线程池中的线程数量。
用来控制 EventLoopThreadPool 启动多少个工作线程。
std::atomic_int started_：记录服务器是否已经启动。
通过原子变量避免多次调用 start() 导致重复启动。
int nextConnId_：新连接的自增编号。
用于生成连接名，例如 xxx-127.0.0.1:8000#1、#2、#3。
ConnectionMap connections_：保存当前服务器中所有活跃的连接。
key 是连接名，value 是对应的 TcpConnectionPtr。

TcpServer.h

EventLoop *loop_; // baseloop 用户自定义的loop

const std::string ipPort_;
const std::string name_;

std::unique_ptr<Acceptor> acceptor_; // 运行在mainloop 任务就是监听新连接事件

std::shared_ptr<EventLoopThreadPool> threadPool_; // one loop per thread

ConnectionCallback connectionCallback_;       //有新连接时的回调
MessageCallback messageCallback_;             // 有读写事件发生时的回调
WriteCompleteCallback writeCompleteCallback_; // 消息发送完成后的回调

ThreadInitCallback threadInitCallback_; // loop线程初始化的回调
int numThreads_;//线程池中线程的数量。
std::atomic_int started_;
int nextConnId_;
ConnectionMap connections_; // 保存所有的连接

TcpServer 类重要的成员方法

1. 构造 / 析构

TcpServer.h

TcpServer(EventLoop *loop,
          const InetAddress &listenAddr,
          const std::string &nameArg,
          Option option = kNoReusePort);
~TcpServer();

TcpServer.cc

static EventLoop *CheckLoopNotNull(EventLoop *loop)
{
    if (loop == nullptr)
    {
        LOG_FATAL("%s:%s:%d mainLoop is null!\n", __FILE__, __FUNCTION__, __LINE__);
    }
    return loop;
}

TcpServer::TcpServer(EventLoop *loop,
                     const InetAddress &listenAddr,
                     const std::string &nameArg,
                     Option option)
    : loop_(CheckLoopNotNull(loop))
    , ipPort_(listenAddr.toIpPort())
    , name_(nameArg)
    , acceptor_(new Acceptor(loop, listenAddr, option == kReusePort))
    , threadPool_(new EventLoopThreadPool(loop, name_))
    , connectionCallback_()
    , messageCallback_()
    , nextConnId_(1)
    , started_(0)
{
    acceptor_->setNewConnectionCallback(
        std::bind(&TcpServer::newConnection, this, std::placeholders::_1, std::placeholders::_2));
}

TcpServer::~TcpServer()
{
    for(auto &item : connections_)
    {
        TcpConnectionPtr conn(item.second);
        item.second.reset();
        conn->getLoop()->runInLoop(
            std::bind(&TcpConnection::connectDestroyed, conn));
    }
}

构造函数主要做了几件事：

检查 loop 是否为空；
记录监听地址字符串；
创建 Acceptor，用于监听新连接；
创建 EventLoopThreadPool，用于管理多个 subLoop；
设置 Acceptor 的新连接回调为 TcpServer::newConnection()。

提示

为什么 Acceptor 要绑定 TcpServer::newConnection()？

因为 Acceptor 只负责“接到新连接”，但不知道接下来怎么管理这个连接。所以 TcpServer 需要接管这个 connfd，把它包装成 TcpConnection 并分配给某个线程。

析构函数负责释放服务器中还没销毁的所有连接：

遍历 connections_ 让每个连接在它所属的 EventLoop 中执行 connectDestroyed() 从而安全地关闭连接

2. `setThreadInitCallback(...)`

TcpServer.h

void setThreadInitCallback(const ThreadInitCallback &cb) { threadInitCallback_ = cb; }

这个函数用于设置线程初始化回调。

当 EventLoopThreadPool 启动每个子线程时，可以执行这段回调，用于：

打日志
设置线程名
初始化某些线程相关资源

3. `setConnectionCallback(...)`

TcpServer.h

void setConnectionCallback(const ConnectionCallback &cb) { connectionCallback_ = cb; }

这个函数用于设置连接建立/关闭回调。

当一个 TcpConnection 建立或者关闭时，上层用户可以收到通知。这通常是服务器框架对外暴露的基础回调之一。

4. `setMessageCallback(...)`

TcpServer.h

void setMessageCallback(const MessageCallback &cb) { messageCallback_ = cb; }

这个函数用于设置消息到达回调。

当某个连接上收到数据时，TcpConnection 会把读到的数据交给这个回调。它通常是业务处理最核心的入口。

5. `setWriteCompleteCallback(...)`

TcpServer.h

void setWriteCompleteCallback(const WriteCompleteCallback &cb) { writeCompleteCallback_ = cb; }

这个函数用于设置写完成回调。

当一批数据已经完全发送出去后，可以通过这个回调通知上层。

6. `setThreadNum(int numThreads)`

TcpServer.h

void setThreadNum(int numThreads);

TcpServer.cc

void TcpServer::setThreadNum(int numThreads)
{
    int numThreads_=numThreads;
    threadPool_->setThreadNum(numThreads_);
}

这个函数用于设置底层 subLoop 的线程数量。

TcpServer 可以通过线程池把不同连接分配到不同的 EventLoop 中处理，从而实现并发处理能力。

7. `start()`

TcpServer.h

void start();

TcpServer.cc

void TcpServer::start()
{
    if (started_.fetch_add(1) == 0)    // 防止一个TcpServer对象被start多次
    {
        threadPool_->start(threadInitCallback_);    // 启动底层的loop线程池
        loop_->runInLoop(std::bind(&Acceptor::listen, acceptor_.get()));
    }
}

start() 是启动服务器监听的入口。

它做了什么

用 started_ 保证只启动一次；
启动线程池；
在 baseLoop 里调用 Acceptor::listen()，开始监听新连接。

提示

为什么要用 runInLoop()？

因为 Acceptor 属于 mainLoop 所在的线程，监听操作应该在正确的事件循环线程中执行。 runInLoop() 可以保证这个函数在合适的线程中运行。

8. `newConnection(int sockfd, const InetAddress &peerAddr)`

TcpServer.h

void newConnection(int sockfd, const InetAddress &peerAddr);

TcpServer.cc

// 有一个新用户连接，acceptor会执行这个回调操作，负责将mainLoop接收到的请求连接(acceptChannel_会有读事件发生)通过回调轮询分发给subLoop去处理
void TcpServer::newConnection(int sockfd, const InetAddress &peerAddr)
{
   // 轮询算法 选择一个subLoop 来管理connfd对应的channel
    EventLoop *ioLoop = threadPool_->getNextLoop();
    char buf[64] = {0};
    snprintf(buf, sizeof buf, "-%s#%d", ipPort_.c_str(), nextConnId_);
    ++nextConnId_;
    std::string connName = name_ + buf;

    LOG_INFO("TcpServer::newConnection [%s] - new connection [%s] from %s\n",
             name_.c_str(), connName.c_str(), peerAddr.toIpPort().c_str());
    
    sockaddr_in local;
    ::memset(&local, 0, sizeof(local));
    socklen_t addrlen = sizeof(local);
    if(::getsockname(sockfd, (sockaddr *)&local, &addrlen) < 0)
    {
        LOG_ERROR("sockets::getLocalAddr");
    }

    InetAddress localAddr(local);
    TcpConnectionPtr conn(new TcpConnection(ioLoop,
                                            connName,
                                            sockfd,
                                            localAddr,
                                            peerAddr));
    connections_[connName] = conn;

    conn->setConnectionCallback(connectionCallback_);
    conn->setMessageCallback(messageCallback_);
    conn->setWriteCompleteCallback(writeCompleteCallback_);

    conn->setCloseCallback(
        std::bind(&TcpServer::removeConnection, this, std::placeholders::_1));

    ioLoop->runInLoop(
        std::bind(&TcpConnection::connectEstablished, conn));
}

这个函数是在 Acceptor 接收到新连接后调用的。

它的作用

从线程池中选一个 subLoop；
生成新连接名字；
获取本地地址；
创建 TcpConnection；
保存到 connections_；
给连接绑定各种回调；
设置关闭连接时的回调；
在目标 EventLoop 中执行 connectEstablished()。

它完成了“新连接接入 -> 封装成 TcpConnection -> 分发到某个 subLoop”的全过程。

9. `removeConnection(const TcpConnectionPtr &conn)`

TcpServer.h

void removeConnection(const TcpConnectionPtr &conn);

TcpServer.cc

void TcpServer::removeConnection(const TcpConnectionPtr &conn)
{
    loop_->runInLoop(
        std::bind(&TcpServer::removeConnectionInLoop, this, conn));
}

这个函数用于发起连接移除请求。

TcpConnection 销毁时，可能不是在 baseLoop 线程中执行的，所以这里先把删除操作投递回 baseLoop，由 removeConnectionInLoop() 统一处理。

10. `removeConnectionInLoop(const TcpConnectionPtr &conn)`

TcpServer.h

void removeConnectionInLoop(const TcpConnectionPtr &conn);

TcpServer.cc

void TcpServer::removeConnectionInLoop(const TcpConnectionPtr &conn)
{
    LOG_INFO("TcpServer::removeConnectionInLoop [%s] - connection %s\n",
             name_.c_str(), conn->name().c_str());

    connections_.erase(conn->name());
    EventLoop *ioLoop = conn->getLoop();
    ioLoop->queueInLoop(
        std::bind(&TcpConnection::connectDestroyed, conn));
}

这个函数是真正执行连接移除的地方。

它做了什么

把连接从 connections_ 中移除；
取得连接所属的 EventLoop；
在该 EventLoop 中调用 TcpConnection::connectDestroyed()。

提示

为什么要这样设计？

因为连接的销毁和状态清理必须在正确的线程里做，不能随便跨线程直接删除。

所以这里通过 queueInLoop() 把销毁动作交给连接所属的 loop 执行。

TcpServer 和 Acceptor、EventLoopThreadPool、TcpConnection 的关系

Acceptor：负责接收新连接；
EventLoopThreadPool：负责选择哪个 subLoop 来处理连接；
TcpConnection：负责一条具体连接的生命周期和 IO；
TcpServer：把这些模块串起来，对外提供完整服务器能力。

TcpServer 的整体工作流程

1. 创建服务器对象

TcpServer server(loop, addr, "server");

此时会创建：

Acceptor
EventLoopThreadPool
各种回调保存区

2. 设置回调

通常会设置：

setConnectionCallback()
setMessageCallback()
setWriteCompleteCallback()
setThreadInitCallback()

3. 启动服务器

调用：

server.start();

后：

启动线程池；
让 Acceptor 开始监听新连接。

4. 接收新连接

当有客户端连接进来时：

Acceptor 接收到 connfd
回调到 TcpServer::newConnection()
创建 TcpConnection
分配给某个 subLoop

5. 管理连接生命周期

当连接关闭时：

TcpConnection 触发关闭回调
TcpServer::removeConnection() 被调用
连接从 connections_ 中移除
在所属 loop 中销毁连接

TcpServer 就像整个服务器的“总经理”：

接待员 Acceptor 负责收新客人；
调度员 EventLoopThreadPool 负责给客人分配服务窗口；
具体服务员 TcpConnection 负责和客人交互；
TcpServer 负责把所有人安排好、保存好、清理好。

总结

TcpServer 的核心职责

对外提供服务器入口
管理监听和连接
接收新连接并分配给子线程
保存所有活跃连接
处理连接关闭与销毁
提供用户可定制的回调接口

TcpServer 是 muduo 的服务器总控类，它负责把 Acceptor、EventLoopThreadPool 和 TcpConnection 串成一个完整的 TCP 服务器。

TcpServer 源码

TcpServer.h

#pragma once

/**
 * 用户使用muduo编写服务器程序
 **/

#include <functional>
#include <string>
#include <memory>
#include <atomic>
#include <unordered_map>

#include "EventLoop.h"
#include "Acceptor.h"
#include "InetAddress.h"
#include "noncopyable.h"
#include "EventLoopThreadPool.h"
#include "Callbacks.h"
#include "TcpConnection.h"
#include "Buffer.h"

// 对外的服务器编程使用的类
class TcpServer
{
public:
    using ThreadInitCallback = std::function<void(EventLoop *)>;

    enum Option
    {
        kNoReusePort,//不允许重用本地端口
        kReusePort,//允许重用本地端口
    };

    TcpServer(EventLoop *loop,
              const InetAddress &listenAddr,
              const std::string &nameArg,
              Option option = kNoReusePort);
    ~TcpServer();

    void setThreadInitCallback(const ThreadInitCallback &cb) { threadInitCallback_ = cb; }
    void setConnectionCallback(const ConnectionCallback &cb) { connectionCallback_ = cb; }
    void setMessageCallback(const MessageCallback &cb) { messageCallback_ = cb; }
    void setWriteCompleteCallback(const WriteCompleteCallback &cb) { writeCompleteCallback_ = cb; }

    // 设置底层subloop的个数
    void setThreadNum(int numThreads);
    /**
     * 如果没有监听, 就启动服务器(监听).
     * 多次调用没有副作用.
     * 线程安全.
     */
    void start();

private:
    void newConnection(int sockfd, const InetAddress &peerAddr);
    void removeConnection(const TcpConnectionPtr &conn);
    void removeConnectionInLoop(const TcpConnectionPtr &conn);

    using ConnectionMap = std::unordered_map<std::string, TcpConnectionPtr>;

    EventLoop *loop_; // baseloop 用户自定义的loop

    const std::string ipPort_;
    const std::string name_;

    std::unique_ptr<Acceptor> acceptor_; // 运行在mainloop 任务就是监听新连接事件

    std::shared_ptr<EventLoopThreadPool> threadPool_; // one loop per thread

    ConnectionCallback connectionCallback_;       //有新连接时的回调
    MessageCallback messageCallback_;             // 有读写事件发生时的回调
    WriteCompleteCallback writeCompleteCallback_; // 消息发送完成后的回调

    ThreadInitCallback threadInitCallback_; // loop线程初始化的回调
    int numThreads_;//线程池中线程的数量。
    std::atomic_int started_;
    int nextConnId_;
    ConnectionMap connections_; // 保存所有的连接
};

TcpServer.cc

#include <functional>
#include <string.h>

#include "TcpServer.h"
#include "Logger.h"
#include "TcpConnection.h"

static EventLoop *CheckLoopNotNull(EventLoop *loop)
{
    if (loop == nullptr)
    {
        LOG_FATAL("%s:%s:%d mainLoop is null!\n", __FILE__, __FUNCTION__, __LINE__);
    }
    return loop;
}

TcpServer::TcpServer(EventLoop *loop,
                     const InetAddress &listenAddr,
                     const std::string &nameArg,
                     Option option)
    : loop_(CheckLoopNotNull(loop))
    , ipPort_(listenAddr.toIpPort())
    , name_(nameArg)
    , acceptor_(new Acceptor(loop, listenAddr, option == kReusePort))
    , threadPool_(new EventLoopThreadPool(loop, name_))
    , connectionCallback_()
    , messageCallback_()
    , nextConnId_(1)
    , started_(0)
{
    // 当有新用户连接时，Acceptor类中绑定的acceptChannel_会有读事件发生，执行handleRead()调用TcpServer::newConnection回调
    acceptor_->setNewConnectionCallback(
        std::bind(&TcpServer::newConnection, this, std::placeholders::_1, std::placeholders::_2));
}

TcpServer::~TcpServer()
{
    for(auto &item : connections_)
    {
        TcpConnectionPtr conn(item.second);
        item.second.reset();    // 把原始的智能指针复位 让栈空间的TcpConnectionPtr conn指向该对象 当conn出了其作用域 即可释放智能指针指向的对象
        // 销毁连接
        conn->getLoop()->runInLoop(
            std::bind(&TcpConnection::connectDestroyed, conn));
    }
}

// 设置底层subloop的个数
void TcpServer::setThreadNum(int numThreads)
{
    int numThreads_=numThreads;
    threadPool_->setThreadNum(numThreads_);
}

// 开启服务器监听
void TcpServer::start()
{
    if (started_.fetch_add(1) == 0)    // 防止一个TcpServer对象被start多次
    {
        threadPool_->start(threadInitCallback_);    // 启动底层的loop线程池
        loop_->runInLoop(std::bind(&Acceptor::listen, acceptor_.get()));
    }
}

// 有一个新用户连接，acceptor会执行这个回调操作，负责将mainLoop接收到的请求连接(acceptChannel_会有读事件发生)通过回调轮询分发给subLoop去处理
void TcpServer::newConnection(int sockfd, const InetAddress &peerAddr)
{
   // 轮询算法 选择一个subLoop 来管理connfd对应的channel
    EventLoop *ioLoop = threadPool_->getNextLoop();
    char buf[64] = {0};
    snprintf(buf, sizeof buf, "-%s#%d", ipPort_.c_str(), nextConnId_);
    ++nextConnId_;  // 这里没有设置为原子类是因为其只在mainloop中执行 不涉及线程安全问题
    std::string connName = name_ + buf;

    LOG_INFO("TcpServer::newConnection [%s] - new connection [%s] from %s\n",
             name_.c_str(), connName.c_str(), peerAddr.toIpPort().c_str());
    
    // 通过sockfd获取其绑定的本机的ip地址和端口信息
    sockaddr_in local;
    ::memset(&local, 0, sizeof(local));
    socklen_t addrlen = sizeof(local);
    if(::getsockname(sockfd, (sockaddr *)&local, &addrlen) < 0)
    {
        LOG_ERROR("sockets::getLocalAddr");
    }

    InetAddress localAddr(local);
    TcpConnectionPtr conn(new TcpConnection(ioLoop,
                                            connName,
                                            sockfd,
                                            localAddr,
                                            peerAddr));
    connections_[connName] = conn;
    // 下面的回调都是用户设置给TcpServer => TcpConnection的，至于Channel绑定的则是TcpConnection设置的四个，handleRead,handleWrite... 这下面的回调用于handlexxx函数中
    conn->setConnectionCallback(connectionCallback_);
    conn->setMessageCallback(messageCallback_);
    conn->setWriteCompleteCallback(writeCompleteCallback_);

    // 设置了如何关闭连接的回调
    conn->setCloseCallback(
        std::bind(&TcpServer::removeConnection, this, std::placeholders::_1));

    ioLoop->runInLoop(
        std::bind(&TcpConnection::connectEstablished, conn));
}

void TcpServer::removeConnection(const TcpConnectionPtr &conn)
{
    loop_->runInLoop(
        std::bind(&TcpServer::removeConnectionInLoop, this, conn));
}

void TcpServer::removeConnectionInLoop(const TcpConnectionPtr &conn)
{
    LOG_INFO("TcpServer::removeConnectionInLoop [%s] - connection %s\n",
             name_.c_str(), conn->name().c_str());

    connections_.erase(conn->name());
    EventLoop *ioLoop = conn->getLoop();
    ioLoop->queueInLoop(
        std::bind(&TcpConnection::connectDestroyed, conn));
}