C++仿 muduo 高并发服务器：Poller 与 Channel 模块联合解析

Poller 与 Channel 模块联合解析：仿 muduo 高并发服务器（Reactor 雏形）

本文基于仿 muduo 高并发服务器项目中的 Poller 与 Channel 模块源码及测试代码，梳理两个模块的职责、函数调用关系、事件处理流程与整体架构。

1. 模块概览

在 Reactor 模型中，两个最核心的基础组件是：

• Poller：对 epoll 的封装，负责监控所有 fd 的 I/O 事件。
• Channel：对一个 fd 的事件管理单元，保存需要监控的事件、实际就绪的事件，以及各种事件回调。

它们的关系：

Poller
└── _channels: fd -> Channel*
├── Channel(listenfd) // 监听套接字
├── Channel(connfd1) // 通信套接字
├── Channel(connfd2)
└── …

• Poller 通过 epoll_wait 拿到就绪事件。
• 根据就绪事件的 fd，在 _channels 中找到对应的 Channel。
• 把就绪事件类型设置到 Channel，加入活跃列表。
• 上层遍历活跃列表，调用 Channel::HandleEvent()。
• HandleEvent() 根据事件类型调用对应的回调函数。

这就是 Reactor 的“事件分发”雏形。

2. Channel 模块详解

2.1 类定义

class Poller;

class Channel {
private:
int _fd;
Poller *_poller;
uint32_t _events; // 当前需要监控的事件
uint32_t _revents; // 当前实际就绪的事件

using EventCallback = std::function<void()>;
EventCallback _read_callback;
EventCallback _write_callback;
EventCallback _error_callback;
EventCallback _close_callback;
EventCallback _event_callback;

public:
Channel(Poller *poller, int fd)
: _fd(fd), _events(0), _revents(0), _poller(poller) {}

int Fd() { return _fd; }
uint32_t Events() { return _events; }
void SetREvents(uint32_t events) { _revents = events; }

void SetReadCallback(const EventCallback &cb) { _read_callback = cb; }
void SetWriteCallback(const EventCallback &cb) { _write_callback = cb; }
void SetErrorCallback(const EventCallback &cb) { _error_callback = cb; }
void SetCloseCallback(const EventCallback &cb) { _close_callback = cb; }
void SetEventCallback(const EventCallback &cb) { _event_callback = cb; }

bool ReadAble() { return (_events & EPOLLIN); }
bool WriteAble() { return (_events & EPOLLOUT); }

void EnableRead() { _events |= EPOLLIN; Update(); }
void EnableWrite() { _events |= EPOLLOUT; Update(); }
void DisableRead() { _events &= ~EPOLLIN; Update(); }
void DisableWrite() { _events &= ~EPOLLOUT; Update(); }
void DisableAll() { _events = 0; Update(); }

void Remove();
void Update();

void HandleEvent();
};

2.2 核心成员

2.3 事件监控的增删改

void EnableRead() { _events |= EPOLLIN; Update(); }
void EnableWrite() { _events |= EPOLLOUT; Update(); }
void DisableRead() { _events &= ~EPOLLIN; Update(); }
void DisableWrite(){ _events &= ~EPOLLOUT; Update(); }
void DisableAll() { _events = 0; Update(); }

• 修改 _events 后，调用 Update()。
• Update() 定义在类外：

void Channel::Update() {
return _poller->UpdateEvent(this);
}

• UpdateEvent 内部根据是否已经添加过，决定 EPOLL_CTL_ADD 还是 EPOLL_CTL_MOD。

2.4 事件处理

void HandleEvent() {
if ((_revents & EPOLLIN) || (_revents & EPOLLRDHUP) || (_revents & EPOLLPRI)) {
if (_event_callback) _event_callback();
if (_read_callback) _read_callback();
}
if (_revents & EPOLLOUT) {
if (_event_callback) _event_callback();
if (_write_callback) _write_callback();
}
else if (_revents & EPOLLERR) {
if (_event_callback) _event_callback();
if (_error_callback) _error_callback();
}
else if (_revents & EPOLLHUP) {
if (_event_callback) _event_callback();
if (_close_callback) _close_callback();
}
}

• 读事件判断包含 EPOLLIN、EPOLLRDHUP、EPOLLPRI。
• 写事件、错误、挂断使用 else if，一次只处理一个“可能释放连接”的事件。
• 任意事件回调 _event_callback 在每种事件前都会调用，常用于刷新活跃度。

3. Poller 模块详解

3.1 类定义

#define MAX_EPOLLEVENTS 1024

class Poller {
private:
int _epfd;
struct epoll_event _evs[MAX_EPOLLEVENTS];
std::unordered_map<int, Channel*> _channels;

private:
void Update(Channel *channel, int op) {
int fd = channel->Fd();
struct epoll_event ev;
ev.data.fd = fd;
ev.events = channel->Events();
int ret = epoll_ctl(_epfd, op, fd, &ev);
if (ret < 0) {
ERR_LOG("EPOLLCTL FAILED!");
abort();
}
}

bool HasChannel(Channel *channel) {
auto it = _channels.find(channel->Fd());
return it != _channels.end();
}

public:
Poller() {
_epfd = epoll_create(256);
if (_epfd < 0) {
ERR_LOG("EPOLL CREATE FAILED!!");
abort();
}
}

void UpdateEvent(Channel *channel) {
bool ret = HasChannel(channel);
if (ret == false) {
_channels.insert(std::make_pair(channel->Fd(), channel));
return Update(channel, EPOLL_CTL_ADD);
}
return Update(channel, EPOLL_CTL_MOD);
}

void RemoveEvent(Channel *channel) {
auto it = _channels.find(channel->Fd());
if (it != _channels.end()) {
_channels.erase(it);
}
Update(channel, EPOLL_CTL_DEL);
}

void Poll(std::vector<Channel*> *active) {
int nfds = epoll_wait(_epfd, _evs, MAX_EPOLLEVENTS, –1);
if (nfds < 0) {
if (errno == EINTR) return;
ERR_LOG("EPOLL WAIT ERROR:%s\\n", strerror(errno));
abort();
}
for (int i = 0; i < nfds; i++) {
auto it = _channels.find(_evs[i].data.fd);
assert(it != _channels.end());
it->second->SetREvents(_evs[i].events);
active->push_back(it->second);
}
}
};

3.2 核心成员

3.3 核心接口

3.4 Poll 工作流程

void Poll(std::vector<Channel*> *active) {
int nfds = epoll_wait(_epfd, _evs, MAX_EPOLLEVENTS, –1);
// …
for (int i = 0; i < nfds; i++) {
auto it = _channels.find(_evs[i].data.fd);
assert(it != _channels.end());
it->second->SetREvents(_evs[i].events);
active->push_back(it->second);
}
}

4. 联合工作流程与函数调用关系

4.1 整体架构图（文字版）

main
│
├─ 创建 Poller
├─ 创建监听 Socket，CreateServer
├─ 创建监听 Channel，绑定到 listenfd
│ └─ SetReadCallback(Acceptor)
│ └─ EnableRead()
│ └─ Channel::Update()
│ └─ Poller::UpdateEvent()
│ └─ epoll_ctl(ADD, listenfd, EPOLLIN)
│
└─ while(1)
├─ Poller::Poll(&actives)
│ └─ epoll_wait()
│ └─ 返回就绪事件
│ └─ 对每个事件：查 _channels，SetREvents，加入 actives
│
└─ for each active Channel
└─ Channel::HandleEvent()
├─ 读事件 → _read_callback()
├─ 写事件 → _write_callback()
├─ 错误 → _error_callback()
└─ 挂断 → _close_callback()

4.2 监听 Channel 的读事件流程

4.3 通信 Channel 的读事件流程

4.4 关闭流程

5. 测试代码解析

5.1 服务端

int main() {
Poller poller;
Socket lst_sock;
lst_sock.CreateServer(8500);

Channel channel(&poller, lst_sock.Fd());
channel.SetReadCallback(std::bind(Acceptor, &poller, &channel));
channel.EnableRead();

while (1) {
std::vector<Channel*> actives;
poller.Poll(&actives);
for (auto &a : actives) {
a->HandleEvent();
}
}
}

• 创建 Poller。
• 创建监听套接字，绑定 8500。
• 为监听 fd 创建 Channel。
• 设置读回调为 Acceptor。
• 启动读监控。
• 主循环：Poll 拿活跃 Channel，逐个 HandleEvent。

5.2 Acceptor

void Acceptor(Poller *poller, Channel *lst_channel) {
int fd = lst_channel->Fd();
int newfd = accept(fd, NULL, NULL);
if (newfd < 0) return;

Channel *channel = new Channel(poller, newfd);
channel->SetReadCallback(std::bind(HandleRead, channel));
channel->SetWriteCallback(std::bind(HandleWrite, channel));
channel->SetCloseCallback(std::bind(HandleClose, channel));
channel->SetErrorCallback(std::bind(HandleError, channel));
channel->SetEventCallback(std::bind(HandleEvent, channel));
channel->EnableRead();
}

• accept 新连接。
• new Channel，设置各种回调。
• EnableRead 把新 fd 加入 epoll 监控。

5.3 通信回调

void HandleRead(Channel *channel) {
int fd = channel->Fd();
char buf[1024] = {0};
int ret = recv(fd, buf, 1023, 0);
if (ret <= 0) {
return HandleClose(channel);
}
std::cout << buf << std::endl;
channel->EnableWrite();
}

void HandleWrite(Channel *channel) {
int fd = channel->Fd();
const char *data = "天气不错!";
int ret = send(fd, data, strlen(data), 0);
if (ret < 0) {
return HandleClose(channel);
}
channel->DisableWrite();
}

void HandleClose(Channel *channel) {
std::cout << "close: " << channel->Fd() << std::endl;
channel->Remove();
delete channel;
}

• 读：recv，打印，启动写事件。
• 写：send，关闭写事件。
• 关闭：移除监控，删除 Channel。

5.4 客户端

int main() {
Socket cli_sock;
cli_sock.CreateClient(8500, "127.0.0.1");
while (1) {
std::string str = "hello";
cli_sock.Send(str.c_str(), str.size());
char buf[1024] = {0};
cli_sock.Recv(buf, 1023);
DBG_LOG("%s", buf);
sleep(1);
}
}

• 连接服务器。
• 循环发送 "hello"。
• 接收服务端响应，用 DBG_LOG 打印（所以会带日志前缀）。

6. 总结

• Poller：封装 epoll，管理 fd -> Channel 映射，负责 epoll_wait 并返回活跃 Channel。
• Channel：封装 fd 和事件回调，负责事件的分发处理。
• 调用关系：
  Poller::Poll → epoll_wait → 找到 Channel → Channel::HandleEvent → 调用具体回调。
• 测试流程：
  监听 Channel 读事件 → Acceptor → 创建通信 Channel → 读事件 → HandleRead → 写事件 → HandleWrite → 关闭 → HandleClose。

这两个模块是 Reactor 模式的基础，后续的 EventLoop、Connection、TcpServer 都会在此基础上构建。
理解它们的调用关系，就理解了高并发服务器事件驱动的核心。