基于 muduo 网络库核心代码分析 Reactor 模型

🔗 GitHub 链接：https://github.com/RareVoyager/cmuduo.git

​ 在高并发网络服务器的设计中，如何高效地处理海量连接始终是一个核心问题。传统的“一连接一线程”模型虽然直观，但当连接数持续增长时，线程数量也会随之膨胀，导致大量的内存占用和线程调度开销，难以满足现代互联网服务对高并发的需求。为了解决这一问题，Linux 提供了 select、poll、epoll 等 I/O 多路复用机制，使得单个线程能够同时监听大量连接的 I/O 事件。在此基础上，Reactor 模型进一步将事件监听、事件分发和业务处理进行解耦，形成了现代高性能网络框架的核心架构。

​ Reactor 模型并非只存在于理论之中，而是广泛应用于现代高性能网络系统，大量工业级的项目采用了Reactor模型进行开发。例如 Nginx 利用 Reactor 模型处理海量 HTTP 连接，Redis 通过单线程事件循环实现高并发访问，Netty 将 Reactor 思想引入 Java 网络编程，而 Node.js 的 EventLoop 本质上也是 Reactor 模型的一种实现。

1.什么是Reactor模型？

​ Reactor 本质上是一种事件驱动（Event Driven）的设计模式，其核心思想是将 I/O 事件的监听与业务逻辑的处理进行分离。

​ 在传统阻塞式网络编程中，程序通常需要为每个连接创建独立线程，并阻塞等待客户端数据到来。而 Reactor 模型则通过 I/O 多路复用技术统一监听多个连接，当某个连接上发生读、写或异常等事件时，再由 Reactor 将对应事件分发给预先注册好的事件处理器（Handler）进行处理。

从整体流程来看，Reactor 模型主要包含以下几个角色：

• Reactor：负责监听和分发事件
• Demultiplexer：事件分离器，通常由 select、poll、epoll 等 I/O 多路复用机制实现
• Event Handler：事件处理器，负责处理具体业务逻辑

其工作流程如下：

1. 应用程序向 Reactor 注册感兴趣的事件；
2. Reactor 通过 I/O 多路复用机制持续监听所有连接；
3. 当某个连接产生事件时，Demultiplexer 返回活跃事件集合；
4. Reactor 将事件分发给对应的 Event Handler；
5. Event Handler 执行具体业务逻辑。

通过这种方式，一个线程即可同时管理大量连接，从而显著提升服务器的并发处理能力。

具体可以看这篇文章 🔗 高性能网络模式：Reactor 和 Proactor

2. 从代码角度来理解什么是Reactor

cmuduo
├── base
├── net
└── ...

​ 代码目录遵从了muduo网络库的目录结构，muduo/base 目录主要存放日志、时间等基础设施代码, 实现相对简单 ，muduo/net下面是网络库的核心代码。因此本文将重点介绍src下的代码。

2.1 Channel 代码剖析

static const int KNoneEvent;
static const int KReadEvent;
static const int KWriteEvent;
EventLoop *loop_;
int fd_; // 文件描述符
int events_; // fd 感兴趣的事件
int revents_; // poller 具体发生的事件
int index_;
std::weak_ptr<void> tie_;
bool tied_;
ReadEventCallback readCallback_;
EventCallback writeCallback_;
EventCallback closeCallback_;
EventCallback errorCallback_;

​ channel 封装了文件描述符(fd) 、fd感兴趣的事件还有内核返回的发生的事件以及事件回调函数。
​ 同时Channel 知道自己所在的EventLoop是哪个(loop_)。
​ tie_ 和 tied_能帮助channel感知当前连接(TcpConnection)是否被销毁, 如果被销毁了则不需要继续执行连接, 避免出现野指针。
​ index_ 用来标识当前channel 的状态( 新创建、已添加、已删除)。

2.2 抽象层Poller

namespace cmuduo
{
	namespace net
	{
		class EventLoop;
		class Poller : base::noncopyable
		{
		public:
			using ChannelList = std::vector<Channel*>;
			Poller(EventLoop* loop);
			virtual ~Poller() = default;

			virtual cmuduo::base::TimeStamp poll(int timeoutMs, ChannelList* activeChannels) = 0;
			virtual void updateChannel(Channel* channel) = 0;
			virtual void removeChannel(Channel* channel) = 0;

			bool hasChannel(Channel* channel);

			// 通过该接口获取默认的IO多路复用的具体实现
			static Poller* newDefaultPoller(EventLoop* loop);

		protected:
			using ChannelMap = std::unordered_map<int, Channel*>;
			ChannelMap channels_;

		private:
			EventLoop* ownerLoop_;
		};
	}// namespace net
}// namespace cmuduo

Poller主要目的是利用面向对象编程中的多态，来对I/O多路复用进行封装。基类提供了四个虚方法，子类需要在使用I/O多路复用的基础上实现这四个方法。具体实现在EpollPoller中。

2.3 EventPoller

​ 要使用EpollPoller，首先需要创建一个epollfd_。后续所有epoll_ctl()和epoll_wait()操作都要基于这个epollfd_进行。

​ poll 方法会通过epoll_wait()活跃事件数量，再通过私有方法fillActiveChannel() 填写活跃 Channel 以及内核实际返回的事件 revents_。如果此时发现容量不够用了则触发扩容机制。这里的activeChannels 参数是传出参数，会被上层调用者EventLoop接收到活跃事件，并让Channel来处理活跃事件。

if (numEvents > 0)
{
	LOG_INFO("%d events happend\n", numEvents);
	fillActiveChannel(numEvents, activeChannels);
	if (numEvents == csize)
	{
		channels_.reserve(csize + csize / 2);
	}
}

​ updateChannel 以及 removeChannel 本质是操作文件描述符执行epoll_ctl() 操作。但是由于 Poller本身不持有具体的channel，他只包含了一个包含Channel的哈希表。 Key是fd，Value 是他对应的Channel*。

using ChannelMap = std::unordered_map<int, Channel*>;
ChannelMap channels_;

​ Poller提供了默认实现类newDefaultPoller，会实例化一个EpollPoller。

2.4 EventLoop 事件循环

​ 学习EventLoop，可以将前面的Channel 与 Poller 串联起来，能更清楚的理解Reactor网络模型。

using ChannelList = std::vector<Channel*>;
std::atomic_bool looping_;
// 标识退出loop
std::atomic_bool quit_;
// poller 返回发生事件的channel 的时间点
base::TimeStamp pollReturnTime_;
// 当前线程id
const pid_t threadId_;

// 当前EventLoop 保存的poller
std::unique_ptr<Poller> poller_;

// mainloop 获取一个新用户的channel，通过轮询算法选择一个subloop，通过该成员唤醒subloop处理相关操作
int wakeupFd_;
std::unique_ptr<Channel> wakeupChannel_;
ChannelList activeChannels_;
// Channel* currentActiveChannle;

// 当前loop 是否有回调操作
std::atomic_bool callingPendingFunctors_;
// 存储loop 需要执行的所有回调操作
std::vector<Functor> pendingFunctor_;
// 保护vector 线程安全
std::mutex mutex_;

​ 由于EventLoop可能在多线程环境下执行，因此要加锁mutex_。并且有可能跨线程执行，因此保存了线程id。接下来是相关的方法。

class EventLoop {
   public:
    // 开启事件循环
	void loop();
    // 退出事件循环
	void quit();
    // 当前线程执行cb
	void runInLoop(Functor cb);
	// cb放入队列,唤醒loop所在线程, 执行cb
	void queueInLoop(Functor cb);
    // 唤醒loop所在线程
	void wakeup();
    // 调用Poller 的方法
	void updateChannel(Channel* channel);
	void removeChannel(Channel* channel);
	bool hasChannel(Channel* channel);

	bool isInLoopThread() const;
   private:
    // 处理wakeup
	void handleRead();
	// 执行回调
	void doPendingFunctors();
};

​ 从EventLoop的成员变量和成员方法以及上述对EpollPoller类和Channel类 的理解，不难看出一个EventLoop持有一个Poller, 并且持有多个Channel。如果要更新Channel关心的事件，会经过 Channel发送更新/移除请求 —> EventLoop 转发 —> Poller 具体处理这样一个流程。并且Channel初始化时会知道自己所在的EventLoop具体是哪一个，Poller初始化时也会知道自己所在的EventLoop是哪一个。这样就保证了文件描述符的事件修改始终发生在正确的线程中，避免多个线程同时操作同一个 epoll 实例所带来的竞态问题。
​ 在muduo设计中，EventLoop 本质上就是 muduo 对 Reactor 的实现，它负责驱动整个事件循环。Channel的所有更新请求统一走update()逻辑。同时muduo遵循的逻辑是one loop per thread，也就是一个EventLoop在一个线程中。通过 isInLoopThread() 方法来判断是否在一个线程内。

​ loop()会调用Poller的poll()方法，来获取活跃 Channel 列表以及Poller::poll() 返回发生事件的Channel的时间点，然后让活跃的Channel::handleEvent(time)处理对应的回调函数。Channel::handleEvent(time)会首先判断tie_(一个TcpConnection对象, std::shared_ptr<>) 是否被析构，再决定是否要执行回调。执行回调时，会根据当前Channel具体发生的事件(revents) 来决定执行哪些回调函数。
​ quit() 用于退出事件循环，它会先将退出标志位quit设置为 true。如果是在 EventLoop 所属线程中调用，那么事件循环会在下一次判断 loop()的 `while (!quit)时退出；如果是在其他线程中调用，则会调用 **wakeup()**，通过向eventfd写入数据来唤醒正在epollwait()中阻塞的 EventLoop，使其能够及时发现quit已经变为true` 并退出循环。

​ 接下来要介绍wakeupChannel_的作用。他构造时接收了一个eventfd作为参数，并且在EventLoop的构造函数中主动设置了有关读的感兴趣事件和读回调。他的作用主要体现在跨线程唤醒或跨线程停止上。也就是执行相关回调操作或退出事件循环时，如果不在该线程执行，那么有可能程序会阻塞在loop() 的epoll_wait()。向 wakeupFd_ 对应的 eventfd 写入数据。wakeupChannel_ 只是封装这个 eventfd，负责监听它的可读事件，唤醒后，epoll_wait() 返回，wakeupChannel_ 变成活跃 Channel，EventLoop 会调用它的读回调 handleRead()，把 eventfd 中的数据读走，避免一直触发可读事件。之后 EventLoop 才会继续执行 doPendingFunctors() 或检查 quit_ 退出循环。

2.5 EventLoopThread 代码剖析

class EventLoopThread : base::noncopyable
{
	public:
		using ThreadInitCallback = std::function<void(EventLoop*)>;
		EventLoopThread(const ThreadInitCallback& cb = ThreadInitCallback() , const std::string& name = std::string());
		~EventLoopThread();
		EventLoop* startLoop();
	private:
		void threadFunc();
	private:
		EventLoop* loop_;
		Thread thread_;
		std::mutex mutex_;
		bool exiting_;
		std::string name_;
		ThreadInitCallback ;
		std::condition_variable cond_;
};

​ 主要就是对EventLoop 又封装了一层线程 thread(C++11 标准)，核心方法实现如下

EventLoop* EventLoopThread::startLoop(){
	thread_.start();
	// 线程间的通信
	EventLoop* loop = nullptr;
	{
	std::unique_lock<std::mutex> lock(mutex_);
		while (loop_ == nullptr){
			cond_.wait(lock);
		}
		loop = loop_;
	}
	return loop;
}

// 以下方法在新线程里面执行
void EventLoopThread::threadFunc(){
	EventLoop loop;
	if (callback_){
		callback_(&loop);
	}
	{
		std::unique_lock<std::mutex> lock(mutex_);
		loop_ = &loop;
		cond_.notify_one();
	}
	loop.loop();
	// 此时 底层服务器要关闭，不进行事件循环了。
	std::unique_lock<std::mutex>(mutex_);
	loop_ = nullptr;
}

void Thread::start(){
	started_ = true;
	sem_t sem;
	sem_init(&sem, false, 0);
	thread_ = std::make_shared<std::thread>([&]() {
		tid_ = CurrentThread::tid();
		sem_post(&sem);
		func_();
	});

	sem_wait(&sem);
	sem_destroy(&sem);
}

​ 在初始化Thread时，EventLoopThread::threadFunc()作为参数之一被传入到thread_ 。EventLoopThread::startLoop()的第一行代码 thread_.start(); 开启多线程，此时会调用EventLoopThread::threadFunc()，这里面的callback_ 参数是上层应用层的EventLoopThreadPool 线程池传入的参数。因为在子线程中 Thread::start() 具体逻辑如上所示。由于具体执行顺序未知，主线程中使用了条件变量来等待EventLoop的创建，子线程来做事件循环。也就是说使用EventLoopThread类时会自动创建一个子线程来处理事件循环。

​ EventLoopThread 的核心作用是：在一个独立线程中创建并运行一个 EventLoop，并把这个 EventLoop 的地址安全地返回给创建者线程。这也是 one loop per thread的核心思想。

2.6 EventLoopThreadPool代码剖析

class EventLoopThreadPool : base::noncopyable{
	public:
		using ThreadInitCallback = std::function<void(EventLoop*)>;
		EventLoopThreadPool(EventLoop* baseloop, const std::string& nameArg);
		~EventLoopThreadPool();

		void start(const ThreadInitCallback& cb = ThreadInitCallback());
		EventLoop* getNextLoop();
		std::vector<EventLoop*> getAllLoops();
    
		void setThreadNum(int numThreads){ numThreads_ = numThreads; }
		bool start() const{ return started_; }
		const std::string name() const{ return name_; }
		
	private:
		EventLoop* baseLoop_;
		std::string name_;
		bool started_;
		int numThreads_;
		int next_;
		std::vector<std::unique_ptr<EventLoopThread>> threads_;
		std::vector<EventLoop*> loops_;
};

​ 在成员变量中，第一个变量是baseloop，也就是常说的mainloop。如果启用了线程池那么这个baseloop仅仅负责接收新连接以及轮询分发到线程池中。如果是单线程模型那么baseloop不仅要负责接收新连接，同时也要负责处理相关事件。默认不启动多线程。需要用户手动设置线程数量，调用EventLoopThreadPool::setThreadNum()来设置线程数量。

​ 线程池的核心资源受到智能指针保护，因此构造和析构函数不需要做额外的操作。关键在EventLoopThreadPool::start() 和 EventLoopThreadPool::getNextLoop()。EventLoopThreadPool::start()是开启线程池的方法，根据创建的线程数量来创建EventLoopThread，并且赋予相应的初始化回调函数。将创建的线程和事件循环放到std::vector<> 容器中管理。EventLoopThreadPool::getNextLoop()是轮询算法的关键。每当有新连接到来时通过该算法获取一个事件循环，将新连接放到该事件循环中。

2.7 TcpConnection 代码剖析

class TcpConnection : base::noncopyable, public std::enable_shared_from_this<TcpConnection>{
	public:
		TcpConnection(EventLoop* loop,const std::string& nameArg,int sockfd,const base::InetAddress& localAddr,const base::InetAddress& peerAddr);
		~TcpConnection();
		// 建立连接
		void connectEstablished();
		void connectDestroyed();
		void shutdown();
		void send(const std::string& buf);

	private:
		enum StateE{
			kDisconnected,
			kConnecting,
			kConnected,
			kDisconnecting
		};
		void setState(StateE state);
		void handleRead(base::TimeStamp reciveTime);
		void handleWrite();
		void handleClose();
		void handleError();
		void shutdownInLoop();
		void sendInLoop(const void* data, size_t len);

	private:
		// 不是mainloop
		EventLoop* loop_;
		const std::string name_;
		std::atomic_int state_;
		bool reading_;

		std::unique_ptr<Socket> socket_;
		std::unique_ptr<Channel> channel_;

		const base::InetAddress localAddr_;
		const base::InetAddress peerAddr_;

		// 新连接回调
		ConnectionCallback connectionCallback_;
		// 读写消息回调
		MessageCallback messageCallback_;
		// 消息发送完成以后的回调
		WriteCompleteCallback writeCompleteCallback_;
		// 关闭连接回调
		CloseCallback closeCallback_;
		// 高水位回调
		HighWaterMarkCallback highWaterMarkCallback_;
		size_t highWaterMark_;

		// 接受数据的缓冲区
		Buffer inputBuffer_;
		// 发送数据的缓冲区
		Buffer outputBuffer_;
};

​ 每当有一个新的连接到达服务器时，TcpConnection会被创建，记录了所在的loop(多线程是轮询出来的，单线程是baseloop)，name_(连接名)，peerAddr_(对端地址),localAddr_(本机地址)。被创建时同时封装了Socket套接字以及Channel。TcpConnection 内部定义并封装了非监听连接 fd 的事件处理逻辑，例如读、写、关闭、错误处理；用户层真正关心的连接回调、消息回调等会被保存在 TcpConnection 中，并在合适时机触发。
​ 同时，TcpConnection内部维护了输入缓冲区和输出缓冲区，也就是自定义的 Buffer。输入缓冲区用于保存从 socket 读取到但尚未被业务层完整处理的数据；输出缓冲区用于保存暂时无法一次性写入 socket 的数据。由于 TCP 是字节流协议，并不保留消息边界，因此 Buffer 的存在可以帮助上层业务更方便地处理半包、粘包以及异步发送等问题。

​ TcpConnection 继承自 std::enable_shared_from_this<TcpConnection>，这使得它可以在已经由 std::shared_ptr 管理的前提下，通过 shared_from_this() 获取指向自身的 std::shared_ptr。这一点和前面提到的 Channel::tie() 有直接关系：TcpConnection会将自身的 shared_ptr 绑定到 Channel 中，Channel 在处理事件前可以通过 weak_ptr 判断 TcpConnection 是否仍然存活，从而避免连接已经销毁后继续执行回调导致悬空指针问题。

​ TcpConnection 的方法大致可以分为五类：连接建立与销毁、数据读取、数据发送、连接关闭、回调管理。

​ 首先是连接生命周期相关方法。TcpConnection::connectEstablished() 通常在新连接创建完成后被调用，它会将连接状态设置为已连接，并调用 channel_->tie(shared_from_this()) 将当前 TcpConnection 与对应的Channel绑定起来。随后开启读事件监听，并触发用户设置的连接回调 connectionCallback_。这一步表示连接已经正式交给 EventLoop 管理。
​ 与之对应的是TcpConnection::connectDestroyed()，它通常在连接销毁前调用。该方法会将连接状态设置为断开状态，取消 Channel 关注的所有事件，并将 Channel 从 Poller 中移除。同时它也会触发连接回调，通知上层连接已经断开。

​ 第二类是数据读取相关方法。TcpConnection::handleRead() 是 Channel 读事件触发后执行的内部回调。它会从 socket 中读取数据，并写入 inputBuffer_。如果读取成功，就调用用户设置的 messageCallback_，把当前连接、输入缓冲区和接收时间传给业务层；如果读取到 0，说明对端关闭连接，会进入关闭处理流程；如果读取出错，则会执行错误处理逻辑。

​ 第三类是数据发送相关方法。TcpConnection::send() 是上层业务最常用的发送接口。由于TcpConnection所属的 EventLoop可能运行在其他线程，所以 TcpConnection::send()通常会先判断当前线程是否是该连接所属的 loop 线程。如果是，就直接调用 TcpConnection::sendInLoop()；如果不是，就通过 TcpConnection::runInLoop() 将发送操作投递到正确的 EventLoop 线程中执行。
​ 对于 TcpConnection::sendInLoop(), 底层有一个高水位回调阈值以及高水位回调函数。高水位回调的核心作用是：当发送缓冲区的数据积压超过某个阈值时，通知上层应用：当前连接发送压力过大，需要采取措施。数据发送流程大概是：先尝试直接 write() 到 socket。如果一次没写完，剩余数据会追加到 outputBuffer_。如果对端接收慢、网络拥塞，或者你发送太快，outputBuffer_ 会越来越大。这段代码就是在判断：追加剩余数据以后，输出缓冲区是否会第一次超过高水位线。以下代码就是在判断：追加剩余数据以后，输出缓冲区是否会第一次超过高水位线。

size_t oldLen = outputBuffer_.readableBytes();

if (oldLen + remaining > highWaterMark_ &&
    oldLen < highWaterMark_ &&
    highWaterMarkCallback_)
{
    loop_->queueInLoop([this, oldLen, remaining]() {
        highWaterMarkCallback_(shared_from_this(), oldLen + remaining);
    });
}

​ 这里有两个判断条件很关键。oldLen + remaining > highWaterMark_ 表示：这次追加数据后，缓冲区大小会超过高水位线。oldLen < highWaterMark_ 表示：追加之前还没有超过高水位线。所以它只在“从未超过变成超过”的那一刻触发一次，避免缓冲区已经处于高水位状态时，每次继续追加都反复触发回调。这个回调通常用于做流量控制，例如暂停继续读取上游数据、停止生产新消息、记录慢连接日志、主动关闭异常连接等。高水位回调就是 TcpConnection 的发送缓冲区压力告警机制，用来告诉上层：数据发送速度跟不上生产速度，outputBuffer_ 已经积压到危险范围了。

​ TcpConnection::sendInLoop() 负责真正的数据写入。它会优先尝试直接向 socket 写数据。如果一次性写完，就可以直接返回；如果 socket 缓冲区已满，或者只写入了一部分数据，剩余数据会被追加到 outputBuffer_ 中，并让 Channel 关注可写事件。等 socket 再次可写时，EventLoop 会触发 TcpConnection::handleWrite()，继续发送 outputBuffer_ 中剩余的数据。

​ 第四类是连接关闭相关方法。TcpConnection::shutdown() 通常用于主动关闭写端，它同样会保证真正的关闭操作在所属 EventLoop 线程中执行。内部会调用 TcpConnection::shutdownInLoop()，在没有待发送数据时关闭 socket 的写端。这样可以保证已经进入输出缓冲区的数据尽量发送完成后再关闭连接。

​ 第五类是事件处理和回调管理方法。TcpConnection::handleWrite() 用于处理 socket 可写事件，继续发送输出缓冲区中的数据。当输出缓冲区清空后，会取消 Channel 对写事件的关注，避免 epoll 一直返回可写事件。如果设置了写完成回调 writeCompleteCallback_，则会在数据完全写出后通知上层。
​ TcpConnection::handleClose() 用于处理连接关闭事件。它会将连接状态改为断开，取消 TcpConnection::的所有事件，并触发关闭回调 closeCallback_。这个关闭回调一般不是用户直接设置的，而是由 TcpServer 设置，用来把该连接从连接表中移除。
​ TcpConnection::handleError() 用于处理 socket 错误事件。它通常会通过 getsockopt() 获取具体的错误码，方便日志记录和问题定位。

2.8 TcpServer 代码剖析

class TcpServer : base::noncopyable
{
public:
    void start();

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

private:
    EventLoop* loop_;
    std::unique_ptr<Acceptor> acceptor_;
    std::shared_ptr<EventLoopThreadPool> threadPool_;
    ConnectionMap connectionMap_;
};

​ 前面介绍的 Channel、Poller、EventLoop 和 TcpConnection 更多属于网络库内部实现，而 TcpServer 则是 muduo 提供给用户直接使用的服务器类。因此用户通常只需要：

EventLoop loop;

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

server.setConnectionCallback(...);
server.setMessageCallback(...);
server.setThreadNum(4);

server.start();

loop.loop();

​ 即可启动一个高并发 TCP 服务器，而无需关心底层 epoll、Channel、EventLoop 等细节。从结构上看，TcpServer 主要管理三个核心组件：

• Acceptor：负责监听并接收新连接
• EventLoopThreadPool：负责管理 subLoop 线程池
• TcpConnection：负责管理已建立连接

它们之间的关系如下：

TcpServer
    │
    ├── Acceptor
    │      │
    │      └── accept()
    │
    ├── EventLoopThreadPool
    │      │
    │      └── subLoop
    │
    └── TcpConnection

其中 mainLoop 只负责监听和接收连接，而具体的连接读写事件则交给 subLoop 处理。

---

TcpServer::start()

void TcpServer::start()
{
    if (started_++ == 0)
    {
        threadPool_->start(threadInitCallback_);

        loop_->runInLoop([this]()
        {
            acceptor_->listen();
        });
    }
}

启动过程分为两个阶段。第一步启动线程池：

threadPool_->start(threadInitCallback_);

根据用户设置的线程数量创建对应数量的 EventLoopThread，并让每个线程都运行独立的 EventLoop。

第二步启动监听：

acceptor_->listen();

listen() 内部会执行：

socket.listen();
channel.enableReading();

从而让监听 socket 注册到 epoll 中。此后当客户端连接到来时，Acceptor 对应的 Channel 会收到读事件通知。

---

TcpServer::newConnection()

当有新的客户端连接到来时，成功返回一个新的连接 fd。随后会触发：

TcpServer::newConnection()

其核心流程如下：

新连接到来
        ↓
accept()
        ↓
获取 connfd
        ↓
getNextLoop()
        ↓
选择一个 subLoop
        ↓
创建 TcpConnection
        ↓
设置各种回调
        ↓
connectEstablished()

其中最重要的是：

EventLoop* ioLoop = threadPool_->getNextLoop();

这里使用轮询算法选择一个 subLoop。从而实现负载均衡。

接下来创建连接对象，并设置各种用户回调：

conn->setConnectionCallback(connectionCallback_);
conn->setMessageCallback(messageCallback_);
conn->setWriteCompleteCallback(writeCompleteCallback_);
conn->setCloseCallback(
    [this](const TcpConnectionPtr& conn)
    {
        removeConnection(conn);
    });

最后将连接注册到对应的 subLoop：

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

至此，一个新的 TCP 连接正式建立完成。

---

TcpServer::removeConnection()

当客户端关闭连接时：

EPOLLHUP
      ↓
TcpConnection::handleClose()
      ↓
closeCallback_
      ↓
TcpServer::removeConnection()

TcpServer 会将对应连接从：

connectionMap_

中删除。

connectionMap_.erase(conn->name());

然后调用：

TcpConnection::connectDestroyed()

注销 Channel、移除 epoll 监听并释放资源。

整个连接生命周期如下：

accept()
    ↓
newConnection()
    ↓
TcpConnection
    ↓
handleRead()
    ↓
handleWrite()
    ↓
handleClose()
    ↓
removeConnection()
    ↓
connectDestroyed()

至此，一个连接的完整生命周期结束。

用户想使用muduo，核心在于编写相关的四个回调函数。其中最重要的就算messageCallback。

2.9 Acceptor 代码剖析

class Acceptor : base::noncopyable
{
public:
    using NewConnectionCallback =
        std::function<void(int sockfd,
                           const base::InetAddress&)>;

private:
    EventLoop* loop_;
    Socket acceptSocket_;
    Channel acceptChannel_;
    NewConnectionCallback newConnectionCallback_;
    bool listenning_;
};

​ Acceptor 是 muduo 中专门负责接收新连接的组件，可以将其理解为 Reactor 模型中的 连接接收器（Connection Acceptor）。它运行在 mainLoop 中，只负责监听和接收客户端连接，并不会处理具体的网络读写业务。

​ 从成员变量可以看出，Acceptor 内部主要由三个核心对象组成：

• EventLoop：负责事件驱动
• Socket：负责监听 socket 的创建与管理
• Channel：负责监听 socket 的读事件

其中：

Socket acceptSocket_;

封装了监听套接字。

在构造函数中会完成：socket()、bind()等操作，并根据用户配置设置。

Channel acceptChannel_;则负责将监听 socket 注册到 EventLoop 对应的 Poller 中。

---

Acceptor::listen()

当服务器启动时，最终会调用：

Acceptor::listen()

其核心逻辑如下：

listenning_ = true;

acceptSocket_.listen();

acceptChannel_.enableReading();

第一步调用系统的：listen()，使 socket 进入监听状态。第二步让：acceptChannel_关注读事件。对于监听 socket 而言，可读事件并不代表有数据到来，而是代表有新的客户端连接到来。因此当新的 TCP 连接建立时，epoll 会返回对应事件。

Acceptor::handleRead()

当监听 socket 变为可读状态后：

epoll_wait() -> Channel::handleEvent() -> Acceptor::handleRead()

此时 Acceptor 会执行：accept() 获取新的连接 fd：

int connfd = acceptSocket_.accept(&peerAddr);

并获得客户端地址信息。随后不会直接创建 TcpConnection，而是触发 newConnectionCallback_,将新连接交给 TcpServer 处理：

if (newConnectionCallback_)
{
    newConnectionCallback_(connfd, peerAddr);
}

而这个回调实际上绑定的是：TcpServer::newConnection()

​ Acceptor 的职责非常单一：只负责监听并接收新连接，不负责连接建立后的任何业务逻辑处理。 这种设计使 muduo 的模块职责更加清晰，也符合 Reactor 模型中“事件接收与事件处理分离”的设计思想。