面试：从零实现一个Mutli-reactor服务器——项目

最近也是在学习Reactor的原理，也看了很多相关的项目。出于学习的心态，我就尝试学习陈硕的Tcp网络编程库（Muduo）。这篇文章将带你从零开始实现一个Mutli-reactor服务器。

一、Mutli-reactor原理

1.1 什么是Mutli-reactor？

在高性能网络编程中，Reactor模式是一种常用的事件驱动架构模式，它允许服务器高效地处理大量并发连接。而Multi-Reactor模式是对经典Reactor模式的一种扩展和优化，通过将事件分派与事件处理分离，并利用多个Reactor实例来提高系统性能和可扩展性。在这种模式下，通常有一个主线程（或少量线程）负责监听新连接请求，并将这些连接分配给工作线程池中的不同Reactor实例进行处理。

当然，这种模型也有明显缺点，连接建立、IO 事件读取以及事件分发完全有单线程处理；比如当某个连接通过系统调用正在读取数据，此时相对于其他事件来说，完全是阻塞状态，新连接无法处理、其他连接的 IO、查询 IO 读写以及事件分发都无法完成。但我们可以通过学习Mutli-reactor编程模型拓展自己对于Reactor的了解。下图是简单的Mutli-reactor示例。

1.2 如何实现一个简单的Mutli-reactor模型？

关键点

示例代码

下面的代码，简单介绍了如何实现一个基本的Mutli-reactor服务器（丐版）

#include <iostream>
#include <thread>
#include <vector>
#include <sys/epoll.h>
#include <unistd.h>
#include <fcntl.h>
#include <netinet/in.h>

#define MAX_EVENTS 10 // 最大事件数
#define PORT 8080 // 监听端口

// 将文件描述符设置为非阻塞模式
void set_nonblock(int fd) {
int flags = fcntl(fd, F_GETFL, 0); // 获取当前文件状态标志
fcntl(fd, F_SETFL, flags | O_NONBLOCK); // 设置为非阻塞模式
}

// 处理客户端连接
void handle_client(int client_fd) {
char buffer[1024]; // 缓冲区用于存储读取的数据
// 循环读取并处理来自客户端的数据
while(true) {
// 从客户端读取数据到缓冲区
ssize_t read_bytes = read(client_fd, buffer, sizeof(buffer));
if(read_bytes > 0) { // 如果读取到数据
std::cout << "Received: " << std::string(buffer, read_bytes) << std::endl; // 打印接收到的数据
write(client_fd, buffer, read_bytes); // 将数据回传给客户端（echo）
} else { // 没有数据或发生错误时关闭连接
close(client_fd);
break;
}
}
}

// 子Reactor处理逻辑，负责处理具体的客户端连接
void sub_reactor(int listen_fd) {
// 创建epoll实例
int epoll_fd = epoll_create1(0);
struct epoll_event event;
event.events = EPOLLIN | EPOLLET; // 监听可读事件和边缘触发模式
event.data.fd = listen_fd; // 关联监听socket
epoll_ctl(epoll_fd, EPOLL_CTL_ADD, listen_fd, &event); // 将监听socket加入epoll监控列表

struct epoll_event events[MAX_EVENTS]; // 用于存储发生的事件
while(true) {
// 等待事件发生（阻塞直到有事件）
int n = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);
for(int i = 0; i < n; ++i) {
if(events[i].data.fd == listen_fd) { // 如果是新的连接请求
sockaddr_in client_addr;
socklen_t client_len = sizeof(client_addr);
int client_fd = accept(listen_fd, (sockaddr*)&client_addr, &client_len); // 接受新连接
if(client_fd == -1) continue;
set_nonblock(client_fd); // 设置为非阻塞模式
event.events = EPOLLIN | EPOLLET; // 设置监听事件类型
event.data.fd = client_fd; // 关联新的客户端socket
epoll_ctl(epoll_fd, EPOLL_CTL_ADD, client_fd, &event); // 将新的客户端socket加入epoll监控列表
} else { // 如果是已有连接的数据到达
handle_client(events[i].data.fd); // 处理客户端数据
epoll_ctl(epoll_fd, EPOLL_CTL_DEL, events[i].data.fd, nullptr); // 从epoll中移除该socket
close(events[i].data.fd); // 关闭socket
}
}
}
}

int main() {
int listen_fd = socket(AF_INET, SOCK_STREAM, 0); // 创建监听socket
set_nonblock(listen_fd); // 设置为非阻塞模式

sockaddr_in server_addr;
server_addr.sin_family = AF_INET; // 使用IPv4地址族
server_addr.sin_port = htons(PORT); // 绑定端口号
server_addr.sin_addr.s_addr = htonl(INADDR_ANY); // 绑定所有可用的网络接口

bind(listen_fd, (sockaddr*)&server_addr, sizeof(server_addr)); // 绑定socket到指定的IP和端口
listen(listen_fd, SOMAXCONN); // 开始监听

// 主Reactor循环在主线程中运行
int epoll_fd = epoll_create1(0); // 创建epoll实例
struct epoll_event event;
event.events = EPOLLIN; // 监听可读事件
event.data.fd = listen_fd; // 关联监听socket
epoll_ctl(epoll_fd, EPOLL_CTL_ADD, listen_fd, &event); // 将监听socket加入epoll监控列表

std::vector<std::thread> threads;
const int num_threads = std::thread::hardware_concurrency(); // 获取硬件支持的最大并发线程数
for(int i = 0; i < num_threads; ++i) {
threads.emplace_back(sub_reactor, listen_fd); // 启动多个子Reactor线程
}

struct epoll_event events[MAX_EVENTS]; // 用于存储发生的事件
while(true) {
int n = epoll_wait(epoll_fd, events, MAX_EVENTS, -1); // 等待事件发生
for(int i = 0; i < n; ++i) {
if(events[i].data.fd == listen_fd) { // 如果是新的连接请求
sockaddr_in client_addr;
socklen_t client_len = sizeof(client_addr);
int client_fd = accept(listen_fd, (sockaddr*)&client_addr, &client_len); // 接受新连接
if(client_fd == -1) continue;
set_nonblock(client_fd); // 设置为非阻塞模式

// 在实际应用中，这里应该将新连接分配给一个子Reactor进行处理
// 为了简化示例，我们只是打印一条消息
std::cout << "New connection accepted" << std::endl;
}
}
}

for(auto& t : threads) {
if(t.joinable()) t.join(); // 等待所有子Reactor线程结束
}

return 0;
}

• set_nonblock函数：将文件描述符设置为非阻塞模式，这样当尝试读写操作而没有数据可读或写时不会阻塞。

• handle_client函数：处理每个客户端的具体逻辑，包括接收数据、打印接收到的数据以及将数据回传给客户端（echo）。

• sub_reactor函数：这是子Reactor的工作逻辑，主要负责处理具体客户端的I/O事件。首先创建一个epoll实例，并将监听套接字添加到epoll的监控列表中。然后进入一个无限循环，等待并处理事件。如果事件对应于一个新的连接请求，则接受该连接，并将其添加到epoll监控列表中；如果是已有的连接上有数据到达，则调用handle_client函数处理数据。

• main函数：

  • 创建监听套接字，并设置为非阻塞模式。
  • 绑定套接字到特定的IP地址和端口，并开始监听。
  • 创建一个epoll实例并将监听套接字添加到其中。
  • 根据系统支持的最大并发线程数启动多个子Reactor线程。
  • 进入一个无限循环，等待并处理事件。在这个示例中，主要是简单地打印出“新连接已接受”的消息。在实际应用中，需要实现将新连接分配给某个子Reactor的逻辑（这一部分的代码逻辑将在后续的代码中实现）。

下面的代码实现了多个并发client客户端向Mutli-reactor服务器发送报文：

#include <iostream>
#include <thread>
#include <vector>
#include <cstring>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <unistd.h>

void client_task(int client_id, const char* ip, int port) {
int sock = 0;
struct sockaddr_in serv_addr;
const char* hello = "Hello from client";
char buffer[1024] = {0};

// 创建socket
if ((sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
std::cerr << "Client " << client_id << ": Socket creation error\\n";
return;
}

serv_addr.sin_family = AF_INET;
serv_addr.sin_port = htons(port);

// 将IP地址从文本转换为二进制形式
if (inet_pton(AF_INET, ip, &serv_addr.sin_addr) <= 0) {
std::cerr << "Client " << client_id << ": Invalid address/ Address not supported\\n";
return;
}

// 连接到服务器
if (connect(sock, (struct sockaddr*)&serv_addr, sizeof(serv_addr)) < 0) {
std::cerr << "Client " << client_id << ": Connection failed\\n";
return;
}

// 发送消息给服务器
send(sock, hello, strlen(hello), 0);
std::cout << "Client " << client_id << ": Hello message sent\\n";

// 接收服务器的响应
read(sock, buffer, 1024);
std::cout << "Client " << client_id << ": Server response: " << buffer << std::endl;

close(sock); // 关闭socket
}

int main() {
const int num_clients = 100; // 要启动的客户端数量
const char* server_ip = "127.0.0.1"; // 服务器IP地址
int server_port = 8080; // 服务器端口

std::vector<std::thread> clients;
for(int i = 0; i < num_clients; ++i) {
clients.emplace_back(client_task, i + 1, server_ip, server_port);
}

// 等待所有客户端线程完成
for(auto& t : clients) {
if(t.joinable()) t.join();
}

return 0;
}

上面的代码主要逻辑为：开启多个client客户端连接到服务器，并发送hello报文、接受hello报文。

实现效果如下：

可以看到我们有100个客户端向服务器发送hello报文，并且也全部接收到了。

1.3 Multi-Reactor有什么用？

采用Multi-Reactor模型的优势在于：

• 高并发处理能力：通过使用多个Reactor实例，可以在一定程度上避免单个Reactor成为瓶颈，从而支持更高的并发连接数。
• 良好的扩展性：随着服务器硬件资源的增加（如CPU核心数），可以通过增加Reactor的数量来充分利用新增资源，增强系统性能。
• 简化复杂度管理：将事件分派与事件处理分离，有助于降低代码复杂度，便于维护和扩展。

1.4其他的Reactor模型

主要的Reactor模型主要有单线程Reactor模型、 线程池Reactor模型、主从Reactor模型。而我们主要要实现的是线程池Reactor模型（Multi-Reactor）。下面是它们之间的比较。

您也可以通过高性能网络编程之 Reactor 网络模型（彻底搞懂）_reactor网络模型-CSDN博客这篇文章进行学习。

1.4.1 单线程Reactor模型

描述：

• 在单线程Reactor模型中，所有的事件分派（监听新连接、读写操作等）和业务处理都在同一个线程内完成。
• 这种模式实现简单，适用于客户端数量不多或I/O操作耗时较短的情况。

优缺点：

• 优点： 实现简单直接，易于理解和维护；由于所有操作都在一个线程中进行，因此不存在线程切换和同步问题。
• 缺点： 性能受限于单个CPU核心的能力，难以应对高并发场景；如果某个处理过程耗时较长，会影响其他任务的及时响应。

1.4.2 线程池Reactor模型（我们目前要实现的）

描述：

• 此模型在基本的Reactor基础上增加了线程池机制，即当Reactor接收到一个事件后，它不会直接处理该事件，而是将事件交给线程池中的一个工作线程来执行。
• 这样可以避免长时间运行的任务阻塞Reactor主线程，提高系统的整体效率。

优缺点：

• 优点： 提升了系统对不同负载的适应能力，特别是对于那些可能包含耗时业务逻辑的请求；通过合理配置线程池大小，可以在一定程度上优化资源利用率。
• 缺点： 需要额外管理线程池，增加了系统复杂度；若线程池配置不当，可能会导致资源浪费或者请求排队等待的问题。

1.4.3 主从Reactor模型

描述：

• 主从Reactor模型是为了解决单Reactor模型在高并发情况下性能瓶颈而设计的一种改进方案。主Reactor负责监听新的连接请求，并将已建立的连接分配给多个从Reactor实例处理。
• 每个从Reactor独立地处理其分配到的连接上的读写事件，从而实现了并行处理。

优缺点：

• 优点： 能够充分利用多核处理器的优势，提升系统的吞吐量；通过分离新连接接受与已有连接处理的过程，减轻了主Reactor的压力。
• 缺点： 设计和实现比单Reactor模型更加复杂；需要精心设计以确保负载均衡，避免某些从Reactor过载。

二、Epoll类的创建  

 关于select、poll、epoll的原理可以参考我这篇文章：

面试：Epoll、Poll、Select学习记录

当你深入到C++开发的世界中尝试实现Multi-Reactor模型时，你会发现epoll这个强大的工具确实是不可或缺的。但是，直接使用epoll进行事件监听和处理有时就像是在走钢丝——不仅需要手动记录各种事件，还得自己管理计时器，尤其是在非阻塞状态下操作的时候，这简直就像是一场噩梦。

所以，为什么不让我们来简化这一切呢？设想一下，如果我们能够创建一个叫做EPollPoller的类来帮我们搞定这一切，那该有多棒！通过这个类的构造函数（实际上就是调用epoll_create），我们可以轻松地创建出epoll句柄。更酷的是，我们可以对外提供一个更新接口（也就是update窗口），这样外界就可以很方便地通过类似epoll_ctl的方式来添加他们想要监听的事件了。

想象一下，你不再需要担心那些复杂的细节，只需要专注于你的业务逻辑。当有新的事件到来时，比如可读事件发生，我们的EPollPoller就会调用相应的回调函数，然后你就能知道“嘿，这里有消息等你处理呢！”这种设计不仅大大简化了代码的复杂度，还让整个开发过程变得更加流畅愉快。

总之，通过引入这样一个EPollPoller类，我们不仅可以使Multi-Reactor模型的实现变得更加简单直观，同时也为后续的功能扩展留下了充足的空间！

2.1 EPollPoller类介绍

2.1.1 Poller类

在创建 EventLoop 的时候，确实需要初始化一个新的 Poller 实例来管理与监听相关的 Channel。每个 Channel 包含了一个 socket 端口及其相关联的回调函数集合，用于处理各种事件（如连接、读取、写入等）。接下来，我们对这些接口和机制进行更加清晰的描述：

接口说明

• virtual Timestamp poll(int timeout, ChannelList* channels) = 0; 这是一个纯虚函数，其实现由派生类提供。该方法执行一次轮询操作，检查是否有任何已注册的通道发生了事件，并返回发生事件的通道列表以及当前的时间戳。它允许框架通过统一接口访问不同的底层I/O多路复用机制（例如 epoll, poll）。

• virtual void updateChannel(Channel* channel) = 0; 和 virtual void removeChannel(Channel* channel) = 0; 这两个纯虚函数分别用于更新（添加或修改）和移除监听的通道。它们确保了可以根据应用程序的需求动态地调整正在监听的文件描述符集，从而实现高效且灵活的事件驱动架构。

• bool hasChannel(Channel* channel) const; 此方法用于检查给定的通道是否已经被注册到 Poller 中。这对于避免重复注册或者错误删除非常有用。

• static Poller* newDefaultPoller(EventLoop* loop); 这是一个静态工厂方法，旨在无需直接指定具体类名的情况下创建并返回一个具体的 Poller 实例（可能是某个派生类实例）。这种方法增强了代码的灵活性和可维护性，因为它允许轻松切换不同类型的 Poller 实现而无需修改客户端代码。

内部机制

在 Poller 类内部，保存了一个指向 EventLoop 的指针 ownerLoop_，这表明了该 Poller 属于哪个事件线程（即 EventLoop）。这种设计对于 Multi-Reactor 模型至关重要。在 Multi-Reactor 架构中，通常有一个主 Reactor 负责接收新连接并将这些连接分配给子 Reactor（Sub-Reactor）。每个 Sub-Reactor 都需要知道它负责监听哪些事件。

此外这个Poller在主Reactor的实现形式是Aceptor（接受新连接并分配给Sub-Reactor，后续介绍）

因此，ownerLoop_ 让 Sub-Reactor 明白自己应该监听哪些事件。一旦通过 poll 方法得知有新的事件到达时，Sub-Reactor 就会执行对应的事件回调函数，从而实现高效的事件处理流程。这种机制确保了系统可以有效地处理大量的并发连接，同时保持良好的响应速度和资源利用率。

详细的代码介绍见下：

// Poller.h
#ifndef POLLER_H
#define POLLER_H

#include "noncopyable.h"
#include "Channel.h"
#include "Timestamp.h"

#include <vector>
#include <unordered_map>
class Poller:noncopyable{
public:
using ChannelList=std::vector<Channel*>;
//构造函数，将次poller绑定到一个EventLoop（线程循环）中
Poller(EventLoop* loop);
// 析函数，释放所有channel。并且具有继承的性质，所以设置为虚函数
virtual ~Poller()=default;
//对外暴露的poll结构，主要用于调用epoll_wait或者poll_wait
// 接受超时事件，并返回发生事件的ChannelList，返回的时间为poll函数被调用的时间点
virtual Timestamp poll(int timeout,ChannelList* channels)=0;
// 注册新的事件（使用Channel包装的事件）
virtual void updateChannel(Channel* channel)=0;
// 删除已注册的事件（使用Channel包装的事件），在Channel被关闭或者被销��时被调用，并在Poller中删除该channel并关闭epoll的句��
virtual void removeChannel(Channel* channel)=0;
// 检测当前是否有事件(使用Channel包装的事件)监听
bool hasChannel(Channel* channel)const;
// 创建并返回一个 Poller 类型的对象指针
// 为什么要设置为static：静态方法可以不依赖于类的任何实例成员变量或实例方法。这意味着不需要创建 Poller 类的对象就可以调用这个方法。
static Poller* newDefaultPoller(EventLoop* loop);

protected:
using ChannelMap=std::unordered_map<int, Channel*>;
// 监听的fd与channel键值对，方便移除增加监听事件（channel）
ChannelMap channels_;
private:
EventLoop* ownerLoop_;// 定义Poller所属的事件循环EventLoop
};

#endif // POLLER_H

// 实现Poller.cc
#include "Poller.h"

//构造函数传入所属的eventloop即可，让poller知道自己所属与哪个线程
// 为什么要知道自己属于那个线程呢
// 因为在后续的multi-reactor中会有个线程池的成员
// 在线程池中所有的线程都被分配了一个poller，线程不断侦听这个poller，管理事件
Poller::Poller(EventLoop* eventLoop)
: ownerLoop_(eventLoop){

}
// 查找这个Channel是否有目标channel
bool Poller::hasChannel(Channel* channel)const{
auto it=channels_.find(channel->fd());
return it!=channels_.end()&&it->second==channel;
}

2.1.2 EPollPoller类

实现细节

为了利用 epoll 高效地管理大量文件描述符，我们在初始化 EPollPoller 时会调用 epoll_create 函数来创建一个 epoll 实例。这里，kInitEventListSize 参数用于指定该 epoll 实例初始监听列表的大小，即它预分配的空间大小，以优化性能并减少动态扩展的需求。

核心方法：poll

poll 方法是 EPollPoller 类中对 Poller 抽象类所定义的纯虚函数 poll 的具体实现。它的主要职责是通过调用 epoll_wait 来等待文件描述符上的事件，并处理这些事件。具体来说，其工作流程如下：

 在这个EventLoop代码我们可以看到调用poll的方法（EventLoop后续详细介绍）：

void EventLoop::loop(){
looping_=true;
quit_=false;

while(!quit_){
activeChannels_.clear();
// 调用Poller的poll函数获取感兴的事件
pollReturnTime_=poller_->poll(kPollTimeMs, &activeChannels_);
// 处理感兴趣的事件
for(Channel* channel : activeChannels_){
currentActiveChannel_ = channel;
// 处理channel上发生的事件
channel->handleEvent(pollReturnTime_);
}
// 执行当前EventLoop事件循环需要处理的回调操作
/**
* IO thread：mainLoop accept fd 打包成 chennel 分发给 subLoop
* mainLoop实现注册一个回调，交给subLoop来执行，wakeup subLoop 之后，让其执行注册的回调操作
* 这些回调函数在 std::vector<Functor> pendingFunctors_; 之中
*/
doPendingFunctors();
}
looping_=false;
}

关键成员变量

• int epollfd_;：这是由 epoll_create 创建的文件描述符，用于后续的所有 epoll 操作。

• std::vector<epoll_event> events_;：这是一个动态数组，用来存储从 epoll_wait 调用中获取到的事件信息。其初始大小由 kInitEventListSize 确定，但可以根据需要自动扩展。

同时updateChannel、removeChannel是为了方便增加、删除channel监听的事件节点。

EPollPoller详细代码如下：

// EPollPoller

#ifndef EPOLLPOLLER_H
#define EPOLLPOLLER_H

#include <vector>
#include <sys/epoll.h>
#include <unistd.h>

#include "Poller.h"
#include "Timestamp.h"

class EPollPoller:public Poller{
public:
EPollPoller(EventLoop* loop);
~EPollPoller() override;

Timestamp poll(int timeout, ChannelList* activeChannels) override;
void updateChannel(Channel* channel) override;
void removeChannel(Channel* channel) override;

private:
// 默认监听数量
static const int kInitEventListSize = 16;
// 填写活跃的连接
void fillActiveChannels(int numEvents,ChannelList* activeChannels)const;
// 更新channel通道，本质是调用了epoll_ctl
void update(int operation,Channel* channel);

int epollfd_; //epoll_create在内核创建空间返回的fd
std::vector<epoll_event> events_;
};

#endif

// EPollPoller.cc

#include "EventLoop.h"
#include "Poller.h"
#include <unistd.h>
#include <sys/eventfd.h>
#include <fcntl.h>
// 防止一个线程创建多个EventLoop (thread_local)
__thread EventLoop *t_loopInThisThread = nullptr;
// 定义默认的Poller IO复用接口的超时时间
const int kPollTimeMs=10000;

// eventfd复用
int createEventfd(){
int eventfd = ::eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
if(eventfd < 0){
perror("eventfd");
abort();
}
return eventfd;
}

EventLoop::EventLoop():
looping_(false),
quit_(false),
callingPendingFunctors_(false),
threadId_(CurrentThread::tid()),
poller_(Poller::newDefaultPoller(this)),
timerQueue_(new TimerQueue(this)),
wakeupFd_(createEventfd()),
wakeupChannel_(new Channel(this,wakeupFd_)),
currentActiveChannel_(nullptr){
// 让mainLoop和wakeupChannel在同一个线程中执行
if(t_loopInThisThread){
std::cerr<< "Another EventLoop" << t_loopInThisThread << " exists in this thread " << threadId_<<std::endl;
}else{
t_loopInThisThread = this;
}

// 设置wakeupfd的事件类型以及发生事件的回调函数
wakeupChannel_->setReadCallback(std::bind(&EventLoop::handleRead, this));
// 每一个EventLoop都监听wakeChannel的EPOLLIN事件
wakeupChannel_->enableReading();
}

EventLoop::~EventLoop(){
//移除所有感兴趣的事件
wakeupChannel_->disableAll();
// 将channel从EentLoop中删除
wakeupChannel_->remove();
::close(wakeupFd_);
//指向EventLoop的指针至空
t_loopInThisThread = nullptr;
}

void EventLoop::loop(){
looping_=true;
quit_=false;

while(!quit_){
activeChannels_.clear();
// 调用Poller的poll函数获取感兴的事件
pollReturnTime_=poller_->poll(kPollTimeMs, &activeChannels_);
// 处理感兴趣的事件
for(Channel* channel : activeChannels_){
currentActiveChannel_ = channel;
// 处理channel上发生的事件
channel->handleEvent(pollReturnTime_);
}
// 执行当前EventLoop事件循环需要处理的回调操作
/**
* IO thread：mainLoop accept fd 打包成 chennel 分发给 subLoop
* mainLoop实现注册一个回调，交给subLoop来执行，wakeup subLoop 之后，让其执行注册的回调操作
* 这些回调函数在 std::vector<Functor> pendingFunctors_; 之中
*/
doPendingFunctors();
}
looping_=false;
}

void EventLoop::quit(){
quit_=true;
/**
* TODO:生产者消费者队列派发方式和muduo的派发方式
* 有可能是别的线程调用quit(调用线程不是生成EventLoop对象的那个线程)
* 比如在工作线程(subLoop)中调用了IO线程(mainLoop)
* 这种情况会唤醒主线程
*/
if(isInLoopThread()){
wakeup();
}
}

void EventLoop::runInLoop(Functor cb){
// 每个EventLoop都保存创建自己的线程tid
// 我们可以通过CurrentThread::tid()获取当前执行线程的tid然后和EventLoop保存的进行比较
if(isInLoopThread()){
cb();
}
// 在非当前eventLoop线程中执行回调函数，需要唤醒evevntLoop所在线程
else{
queueInLoop(cb);
}
}

void EventLoop::queueInLoop(Functor cb){
{
std::unique_lock<std::mutex> lock(mutex_);
pendingFunctors_.emplace_back(cb);
}
// 唤醒相应的，需要执行上面回调操作的loop线程
/**
* TODO:
* std::atomic_bool callingPendingFunctors_; 标志当前loop是否有需要执行的回调操作
* 这个 || callingPendingFunctors_ 比较有必要，因为在执行回调的过程可能会加入新的回调
* 则这个时候也需要唤醒，否则就会发生有事件到来但是仍被阻塞住的情况
*/
if(!isInLoopThread()||callingPendingFunctors_){
//唤醒loop所在线程
wakeup();
}
}

void EventLoop::wakeup(){
uint64_t one = 1;
ssize_t n = ::write(wakeupFd_, &one, sizeof one);
if(n!=sizeof(one)){
perror("EventLoop::wakeup");
abort();
}
}

void EventLoop::handleRead(){
uint64_t one = 1;
ssize_t n = ::read(wakeupFd_, &one, sizeof(one));
// 忽略读取的字节数
if(n!=sizeof(one)){
perror("EventLoop::wakeup");
abort();
}
}

void EventLoop::updateChannel(Channel* channel){
poller_->updateChannel(channel);
}

void EventLoop::removeChannel(Channel* channel){
poller_->removeChannel(channel);
}

bool EventLoop::hasChannel(Channel* channel){
return poller_->hasChannel(channel);
}

void EventLoop::doPendingFunctors(){
std::vector<Functor> pendingFunctors;
callingPendingFunctors_ = true;
/**
* TODO:
* 如果没有生成这个局部的 functors
* 则在互斥锁加持下，我们直接遍历pendingFunctors
* 其他线程这个时候无法访问，无法向里面注册回调函数，增加服务器时延
*/
{
std::unique_lock<std::mutex> lock(mutex_);
pendingFunctors_.swap(pendingFunctors);
}

for(const auto& cb : pendingFunctors){
cb();
}

callingPendingFunctors_ = false;
}

2.1.3 总结

外部如何创建Poller管理我们的poller呢？

那当然是通过Poller中的newDefaultPoller方法啦。通过设置系统环境变量，我们可以自己选择生成那种io复用方法。当然，目前默认生成的都是EPollPoller！

#include "Poller.h"
#include "EPollPoller.h"

#include <stdlib.h>

// 通过环境变量选择那种触发方式
Poller* Poller::newDefaultPoller(EventLoop* eventLoop){
if(getenv("MUDUO_USE_POLL")){
return nullptr; //生成poll实例
}else{
return new EPollPoller(eventLoop);//生成epoll实例
}
}

最后补充新连接到来时候的调用流程：

三、核心代码的实现

在上一章，我们画了一个简易的调用流程图，如下：

那么，我们后续将会不断地基于这张图，补充完整的调用接口。

3.1 Channel的实现

再上一章我们实现了EPollPoller。我们讨论到了想用一个Channel模块管理我们的新连接的监听事件，并且能够对发生的监听事件通过Channel存放的相应的回调函数，完成相应的监听事件操作。

那么我们预期的这个Channel要完成哪些操作呢？

• 存放监听事件的fd（句柄，用于poller监听）
• 感兴趣的事件标志（用于向poller表面我希望监听那些事件）
• 具体发生时间的标志（poller中发生的具体的事件）
• 回调函数（不同事件的回调函数）

3.1.1 Channel类

3.1.1.1 Channel的构造函数

构造函数有两个参数的输入，第一个是EventLoop（循环事件——也可暂时视为Sub-Reactor或Main-Reactor），表明该Channel属于哪个reactor要监听的端口和相关事件，第二个参数就是该Channel监听的socket。代码如下：

// Channel.cc
Channel::Channel(EventLoop *loop, int fd)
: loop_(loop),
fd_(fd),
events_(0),
revents_(0),
index_(-1),
tied_(false)
{
}

// EventLoop
// 创建事件fd
int createEventfd()
{
int evfd = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
if (evfd < 0)
{
std::cerr << "eventfd error: " << std::endl;
}
return evfd;
}

EventLoop::EventLoop() :
looping_(false),
quit_(false),
callingPendingFunctors_(false),
threadId_(CurrentThread::tid()),
poller_(Poller::newDefaultPoller(this)),
timerQueue_(new TimerQueue(this)),
wakeupFd_(createEventfd()), //这里创建fd
wakeupChannel_(new Channel(this, wakeupFd_)), //在这里创建新的Channel，并把刚创建的fd作为输入
currentActiveChannel_(nullptr){}

3.1.1.2 事件更新类接口

外部可以通过调用这些接口，从而调用epoll的updatechannel接口，最终调用epoll_ctl更新监听事件

// Channel.h
// 设置fd相应的事件状态，update()其本质调用epoll_ctl
void enableReading() { events_ |= kReadEvent; update(); }
void disableReading() { events_ &= ~kReadEvent; update(); }
void enableWriting() { events_ |= kWriteEvent; update(); }
void disableWriting() { events_ &= ~kWriteEvent; update(); }
void disableAll() { events_ &= kNoneEvent; update(); }
// Channel.cc
void Channel::update()
{
//TODO:Channel::update()
// 通过该channel所属的EventLoop，调用poller对应的方法，注册fd的events事件
loop_->updateChannel(this);
}

// loop调用 EventLoop.cc

void EventLoop::updateChannel(Channel *channel)
{
poller_->updateChannel(channel);
}

// 随后调用updateChannel，使用epollpoller的update接口，最终调用epoll_ctl，完成对相关监听事件的更新
void EPollPoller::updateChannel(Channel *channel)
{
// TODO:__FUNCTION__
// 获取参数channel在epoll的状态
const int index = channel->index();

// 未添加状态和已删除状态都有可能会被再次添加到epoll中
if (index == kNew || index == kDeleted)
{
// 添加到键值对
if (index == kNew)
{
int fd = channel->fd();
channels_[fd] = channel;
}
else // index == kAdd
{
}
// 修改channel的状态，此时是已添加状态
channel->set_index(kAdded);
// 向epoll对象加入channel
update(EPOLL_CTL_ADD, channel);
}
// channel已经在poller上注册过
else
{
// 没有感兴趣事件说明可以从epoll对象中删除该channel了
if (channel->isNoneEvent())
{
update(EPOLL_CTL_DEL, channel);
channel->set_index(kDeleted);
}
// 还有事件说明之前的事件删除，但是被修改了
else
{
update(EPOLL_CTL_MOD, channel);
}
}
}
// 输入对监听事件的操作operation，和对应的事件channel
void EPollPoller::update(int operation, Channel *channel)
{
epoll_event event;
::memset(&event, 0, sizeof(event));

int fd = channel->fd();
event.events = channel->events();
event.data.fd = fd;
event.data.ptr = channel;

if (::epoll_ctl(epollfd_, operation, fd, &event) < 0)
{
if (operation == EPOLL_CTL_DEL)
{
LOG_ERROR << "epoll_ctl() del error:" << errno;
}
else
{
LOG_FATAL << "epoll_ctl add/mod error:" << errno;
}
}
}

完整的Channel代码

完整的Channel代码如下：

// Channel.h
#ifndef CHANNEL_H
#define CHANNEL_H

#include <functional>
#include <memory>
#include <sys/epoll.h>

#include "noncopyable.h"
#include "Timestamp.h"
#include "Logging.h"

// 前置声明，不引用头文件防止暴露太多信息
class EventLoop;
class Timestamp;

/**
* Channel 理解为通道，封装了sockfd和其感兴趣的event，如EPOLLIN,EPOLLOUT
* 还绑定了poller返回的具体事件
*/
class Channel : noncopyable
{
public:
using EventCallback = std::function<void()>;
using ReadEventCallback = std::function<void(Timestamp)>;

Channel(EventLoop *loop, int fd);
~Channel();

// fd得到poller通知以后，处理事件的回调函数
void handleEvent(Timestamp receiveTime);

// 设置回调函数对象
// 使用右值引用，延长了cb对象的生命周期，避免了拷贝操作
void setReadCallback(ReadEventCallback cb) { readCallback_ = std::move(cb); }
void setWriteCallback(EventCallback cb) { writeCallback_ = std::move(cb); }
void setCloseCallback(EventCallback cb) { closeCallback_ = std::move(cb); }
void setErrorCallback(EventCallback cb) { errorCallback_ = std::move(cb); }

// TODO:防止当 channel 执行回调函数时被被手动 remove 掉
void tie(const std::shared_ptr<void>&);

int fd() const { return fd_; } // 返回封装的fd
int events() const { return events_; } // 返回感兴趣的事件
void set_revents(int revt) { revents_ = revt; } // 设置Poller返回的发生事件

// 设置fd相应的事件状态，update()其本质调用epoll_ctl
void enableReading() { events_ |= kReadEvent; update(); }
void disableReading() { events_ &= ~kReadEvent; update(); }
void enableWriting() { events_ |= kWriteEvent; update(); }
void disableWriting() { events_ &= ~kWriteEvent; update(); }
void disableAll() { events_ &= kNoneEvent; update(); }

// 返回fd当前的事件状态
bool isNoneEvent() const { return events_ == kNoneEvent; }
bool isWriting() const { return events_ & kWriteEvent; }
bool isReading() const { return events_ & kReadEvent; }

/**
* for Poller
* const int kNew = -1; // fd还未被poller监视
* const int kAdded = 1; // fd正被poller监视中
* const int kDeleted = 2; // fd被移除poller
*/
int index() { return index_; }
void set_index(int idx) { index_ = idx; }

// one lool per thread
EventLoop* ownerLoop() { return loop_; }
void remove();

private:
void update();
void handleEventWithGuard(Timestamp receiveTime);

/**
* const int Channel::kNoneEvent = 0;
* const int Channel::kReadEvent = EPOLLIN | EPOLLPRI;
* const int Channel::kWriteEvent = EPOLLOUT;
*/
static const int kNoneEvent;
static const int kReadEvent;
static const int kWriteEvent;

EventLoop *loop_; // 当前Channel属于的EventLoop
const int fd_; // fd, Poller监听对象
int events_; // 注册fd感兴趣的事件
int revents_; // poller返回的具体发生的事件
int index_; // 在Poller上注册的情况

std::weak_ptr<void> tie_; // 弱指针指向TcpConnection(必要时升级为shared_ptr多一份引用计数，避免用户误删)
bool tied_; // 标志此 Channel 是否被调用过 Channel::tie 方法

// 因为 channel 通道里面能够获知fd最终发生的具体的事件revents
// 保存事件到来时的回调函数
ReadEventCallback readCallback_;
EventCallback writeCallback_;
EventCallback closeCallback_;
EventCallback errorCallback_;
};

#endif // CHANNEL_H
// Channel.cc
#include "Channel.h"
#include "EventLoop.h"

const int Channel::kNoneEvent = 0;
const int Channel::kReadEvent = EPOLLIN | EPOLLPRI;
const int Channel::kWriteEvent = EPOLLOUT;

Channel::Channel(EventLoop *loop, int fd)
: loop_(loop),
fd_(fd),
events_(0),
revents_(0),
index_(-1),
tied_(false)
{
}

//TODO:析构操作和断言，判断是否是在当前线程
Channel::~Channel()
{
}

// 在TcpConnection建立得时候会调用
void Channel::tie(const std::shared_ptr<void> &obj)
{
// weak_ptr 指向 obj
tie_ = obj;
tied_ = true;
}

/**
* 当改变channel所表示fd的events事件后，update负责在poller里面更改fd相应的事件epoll_ctl
*
*/
void Channel::update()
{
//TODO:Channel::update()
// 通过该channel所属的EventLoop，调用poller对应的方法，注册fd的events事件
loop_->updateChannel(this);
}

// 在channel所属的EventLoop中，把当前的channel删除掉
void Channel::remove()
{
//TODO:Channel::remove()
loop_->removeChannel(this);
}

// fd得到poller通知以后，去处理事件
void Channel::handleEvent(Timestamp receiveTime)
{
/**
* 调用了Channel::tie得会设置tid_=true
* 而TcpConnection::connectEstablished会调用channel_->tie(shared_from_this());
* 所以对于TcpConnection::channel_ 需要多一份强引用的保证以免用户误删TcpConnection对象
*/
if (tied_)
{
// 变成shared_ptr增加引用计数，防止误删
std::shared_ptr<void> guard = tie_.lock();
if (guard)
{
handleEventWithGuard(receiveTime);
}
// guard为空情况，说明Channel的TcpConnection对象已经不存在了
}
else
{
handleEventWithGuard(receiveTime);
}
}

// 根据相应事件执行回调操作
void Channel::handleEventWithGuard(Timestamp receiveTime)
{
// 对端关闭事件
// 当TcpConnection对应Channel，通过shutdown关闭写端，epoll触发EPOLLHUP
if ((revents_ & EPOLLHUP) && !(revents_ & EPOLLIN))
{
// 确认是否拥有回调函数
if (closeCallback_)
{
closeCallback_();
}
}

// 错误事件
if (revents_ & (EPOLLERR))
{
LOG_ERROR << "the fd = " << this->fd();
if (errorCallback_)
{
errorCallback_();
}
}

// 读事件
if (revents_ & (EPOLLIN | EPOLLPRI))
{
LOG_DEBUG << "channel have read events, the fd = " << this->fd();
if (readCallback_)
{
LOG_DEBUG << "channel call the readCallback_(), the fd = " << this->fd();
readCallback_(receiveTime);
}
}

// 写事件
if (revents_ & EPOLLOUT)
{
if (writeCallback_)
{
writeCallback_();
}
}
}

3.1.2 总结

3.2 EventLoop的实现

凑凑字数，后续补充

3.3 Acceptor的实现

凑凑字数，后续补充

四、其余代码的实现

凑凑字数，后续补充

五、实现一个server

凑凑字数，后续补充

参考