【Linux网络】基于Reactor反应堆模式的高并发服务器

文章目录

• 1. Reactor 模式
• • 1.1 核心组件
  • 1.2 工作流程（经典单线程Reactor）
• 2. 设计思路
• 3. 事件处理器框架
• • 3.1 封装epoll
  • 3.2 Connection基类
  • 3.3 Listener派生类框架
  • 3.3 Channel派生类框架
• 4. 设计服务器
• 5. 补充细节
• • 5.1 读事件处理
  • 5.2 写事件处理
  • 5.3 异常事件处理
• 6. 主函数运行服务器
• 7. 总结

这篇文章我们将基于
Reactor反应堆模式，使用ET + 非阻塞实现一个epoll服务器

1. Reactor 模式

Reactor 模式是处理高并发网络 I/O 的一种经典事件驱动设计模式。它的核心思想是 通过一个或多个事件循环来统一监听和分发所有 I/O 事件，由一个或少数线程来处理大量并发连接，从而避免为每个连接创建独立线程带来的巨大资源开销。

我们先通过一个生活化的比喻来建立直观感受。

想象一下一家餐厅的服务模式：

• 传统多线程模式 (Thread-Per-Connection)：好比一个顾客配一个专属服务员。从点菜、下单到上菜，这个服务员全程只为这一桌服务。如果后厨做菜慢，服务员也只能等着，无法服务其他顾客。顾客多了，餐厅就需要雇佣大量服务员，人力成本极高。
• Reactor 模式：更像一家高效运营的现代餐厅。它有一个（或几个）接待员 (Reactor)。接待员的工作不是直接服务顾客，而是站在餐厅中央，时刻关注所有顾客的状态 (事件)。
  • 当新顾客进门时 (连接事件)，接待员引导其入座。
  • 当某桌顾客举手示意已看好菜单 (可读事件)，接待员通知一个空闲的服务员 (Handler) 过去点菜。
  • 当后厨的菜做好了 (数据就绪)，接待员再通知服务员去上菜 (可写事件)。

这样，通过少数角色（接待员）高效协调多个服务资源，用很少的服务员就能应对大量的顾客，这就是 Reactor 模式的精髓。

1.1 核心组件

1.2 工作流程（经典单线程Reactor）

2. 设计思路

当我们在读取的时候，我们怎么保证我们读到的数据，未来给每一个fd，都要维护起来？

搞清楚这个问题之前，我们需要先说明白：为什么需要为每个fd单独维护数据？

在基于 epoll的事件驱动模型中，程序在一个主循环中等待事件，当某个连接（fd）可读时，才去读取数据。这带来两个问题：

如果不为每个 fd 单独维护接收缓冲区，数据就会混杂或丢失，无法正确重组消息。

所以，我们要怎么维护每一个fd呢？

每个客户端连接对应一个fd。为了高效处理数据，我们可以为每个fd维护独立的buffer，用于暂存未处理完的数据。

问题：listensocket和普通socket，有区别吗？

监听套接字（listensockfd）和普通连接套接字（normal fd）在本质上都是文件描述符，它们在 epoll的事件驱动模型中被统一看待，但其“就绪”后所对应的操作截然不同。只不过listensockfd将来在‘recv’的时候，用的是accept！！两者统一了！！

因此，不是所有fd都需要维护缓冲区buffer，因为我们的监听套接字fd的读事件只需要接受（accept）新的客户端连接，并不会真的读取数据，而已经建立连接的普通套接字fd才需要进行读写数据，这个时候才需要维护自己的缓冲区。

所以，对于不同的套接字fd有不同的行为，这个时候我们就可以采用继承与多态的做法，我们定义一个基类 Connection，在基类中定义监听套接字listensockfd和普通套接字sockfd都需要的成员变量，再分别定义listensockfd需要的派生类Listener，和sockfd需要的派生类Channel，在对应的派生类中定义各自需要的成员变量，这样我们就可以通过多态来实现不同套接字的不同行为

问题：服务器如何管理/派发就绪的事件？ （如新连接到达、 socket 可读或可写）

我们知道，当有文件描述符就绪时，内核会将就绪事件填充到用户提供的 events 数组中，并返回就绪事件的数量。但是我们服务器如何将就绪的事件分发给对应的事件处理器去处理呢？

在前面我们已经打算通过继承和多态来处理不同套接字的不同行为，也就是说，当读事件就绪（新连接到来），我们就要通过Listener派生类来处理就绪的读事件（accept新连接），当普通读写事件就绪，就需要通过Channel派生类来处理就绪的读写事件（recv/send数据到对应fd的缓冲区中）

所以，服务器要想将就绪事件派发给不同的处理器处理，就需要建立服务器和Connection基类之间的关系（多态行为需要基类指针或引用）。因为服务器只能在 events 数组中拿到最直接的文件描述符fd，不能直接拿到对应fd封装的Connection基类。

那么这个时候我们服务器就可以选择哈希表这个容器，来建立需要关心的文件描述符fd和对应的Connection基类之间的关系，这样我们就能派发就绪事件给对应的事件处理器

问题：Listener派生类accept接受新连接时，如何将之后返回的通信套接字sockfd（也就是与客户端连接建立成功之后的普通读写套接字）添加到服务器中的哈希表中呢？

为了服务器后续管理，我们需要将所有内核关心的fd添加到哈希表中建立映射。但是哈希表中存储的对象是服务器需要管理的，并不是对应具体事件处理器管理的，那么Listener派生类在接受新连接之后，得到的普通读写套接字，要怎么添加到服务器内管理的哈希表中呢？

这里我们可以在Connection基类中定义一个回指指针，指向服务器，这样我们就可以通过回指指针调用服务器中的接口来添加fd到哈希表中

同样，后面在普通读写事件时，如果发送缓冲区的数据满了，那么此时就不能再向发送缓冲区写入数据，这个时候我们就需要开启对写事件的关心，这就需要通知内核帮我们关心写事件，所以这个时候也可以通过回指指针来调用服务器中的接口从而向内核中添加对写事件的关心。

所以我们需要在Connection基类中定义一个回指指针，而不是只在Listener派生类中定义，因为都需要通过回指指针来调用服务器中的接口

3. 事件处理器框架

3.1 封装epoll

我们先来把epoll相关的函数封装一下

#pragma once

#include <unistd.h>
#include <sys/epoll.h>
#include "Common.hpp"
#include "Log.hpp"

using namespace LogModule;

class Epoller
{
public:
Epoller() : _epfd(–1)
{
// 创建epoll模型
_epfd = epoll_create(256);
if (_epfd < 0)
{
LOG(LogLevel::FATAL) << "epoll_create error…";
exit(EPOLL_CREATE_ERR);
}
LOG(LogLevel::INFO) << "epoll_create success, _epfd: " << _epfd;
}

void ModEventHelper(int sockfd, uint32_t events, int op)
{
struct epoll_event ev;
ev.data.fd = sockfd;
ev.events = events;
int n = epoll_ctl(_epfd, op, sockfd, &ev);
if (n < 0)
{
LOG(LogLevel::FATAL) << "epoll_ctl error…";
return;
}
LOG(LogLevel::INFO) << "epoll_ctl success…, sockfd: " << sockfd;
}

void AddEvent(int sockfd, uint32_t events)
{
ModEventHelper(sockfd, events, EPOLL_CTL_ADD);
}

void ModEvent(int sockfd, uint32_t events)
{
ModEventHelper(sockfd, events, EPOLL_CTL_MOD);
}

void DelEvent(int sockfd)
{
int m = epoll_ctl(_epfd, EPOLL_CTL_DEL, sockfd, nullptr); // 删除不需要再关心任何事件直接填空指针
if (m > 0)
{
LOG(LogLevel::INFO) << "epoll_ctl remove sockfd success: " << sockfd;
}
}

int WaitEvent(struct epoll_event evs[], int maxnum, int timeout)
{
int n = epoll_wait(_epfd, evs, maxnum, timeout);
if(n < 0)
{
LOG(LogLevel::WARNING) << "epoll_wait error…";
}
else if(n == 0)
{
LOG(LogLevel::WARNING) << "epoll_wait timeout…";
}
else
{
LOG(LogLevel::DEBUG) << "有事件就绪了…，就绪事件的数量n: " << n;
// 事件派发逻辑我们在服务器中实现
}
return n;
}

~Epoller()
{
if (_epfd >= 0)
{
close(_epfd);
}
}

private:
int _epfd;
};

对于epoll相关函数我们在上一篇文章中已经详细介绍过，并介绍了如何使用，这里就不再过多介绍

3.2 Connection基类

在基类中我们定义派生类都需要的成员变量和函数，但是读写函数以及错误处理都会设置为纯虚函数，派生类中需要怎么处理就自己去重写

#pragma once

#include <iostream>

class Reactor;
// 基类
class Connection
{
public:
Connection():_events(0), _owner(nullptr)
{}

virtual void Recver() = 0;
virtual void Sender() = 0;
virtual void Excepter() = 0;
virtual int GetSockfd() = 0;

void SetEvent(const uint32_t& events)
{
_events = events;
}

uint32_t GetEvent()
{
return _events;
}

void SetOwner(Reactor* owner)
{
_owner = owner;
}

Reactor* GetOwner()
{
return _owner;
}

~Connection(){}
private:
uint32_t _events;
Reactor* _owner;
};

3.3 Listener派生类框架

Listener派生类 只需要关心读事件，并且只需要acceot接受连接，并不需要去从缓冲区中读数据上来，同时我们需要ET模式的非阻塞epoll服务器，那就还需要在构造函数中设置非阻塞，那么我们可以将之前介绍的设置非阻塞的方法放在 Common.hpp 里（前面写网络编程时封装的一个公有接口文件，如：错误码封装、不可拷贝的基类等）

#pragma once

#include <memory>
#include "Epoller.hpp"
#include "Socket.hpp"
#include "Common.hpp"
#include "Connection.hpp"

using namespace SocketModule;

// Listener 专门进行获取新连接
class Listener : public Connection
{
public:
Listener(int port):_port(port), _listensockfd(std::make_unique<TcpSocket>())
{
_listensockfd->BuildTcpSocketMethod(_port);
SetEvent(EPOLLIN | EPOLLET); // ET模式
SetNonBlock(_listensockfd->Fd()); // 设置非阻塞
}

void Recver() override
{}

void Sender() override
{}

void Excepter() override
{}

int GetSockfd() override
{
return _listensockfd->Fd();
}

~Listener() {}
private:
int _port;
std::unique_ptr<Socket> _listensockfd;
};

在 Listener派生类 中只需要重写读和获取fd两个接口，另外两个不需要用到。至于具体的读事件处理我们后面再完善

3.3 Channel派生类框架

Channel派生类 需要处理普通的读写事件，因此 Channel派生类 的所有普通读写套接字sockfd都需要自己的缓冲区

#pragma once

#include <string>
#include "InetAddr.hpp"
#include "Log.hpp"
#include "Connection.hpp"

// 普通sockfd的封装
class Channel : public Connection
{
public:
Channel(int sockfd, const InetAddr& client):_sockfd(sockfd), _client(client)
{
SetNonBlock(_sockfd); // 设置非阻塞
}

void Recver() override
{}

void Sender() override
{}

void Excepter() override
{}

int GetSockfd() override
{
return _sockfd;
}

~Channel() {}
private:
int _sockfd;
std::string _inbuffer;
std::string _outbuffer;
InetAddr _client;
};

同样关于具体读写以及异常处理我们后面再处理

4. 设计服务器

在服务器中，首先肯定需要创建一个epoll模型，所以需要一个Epoller类型的指针，当然还有我们一直都在用的运行标志位，还有我们前面说过的哈希表，用来管理获取到的fd，同时还要管理就绪的事件，这里我们直接采用静态数组

#pragma once

#include <unordered_map>
#include "Connection.hpp"

class Reactor
{
static const int evs_num = 128;
private:
bool IsConnectionEmpty()
{
return _connections.empty();
}

// 一次循环
int LoopOnce(int timeout)
{
return _epoller_ptr->WaitEvent(_evs, evs_num, timeout);
}

public:
Reactor():_epoller_ptr(std::make_unique<Epoller>()), _isrunning(false)
{}

void Loop()
{
if(IsConnectionEmpty())
return;

_isrunning = true;
int timeout = –1;
while(_isrunning)
{
int n = LoopOnce(timeout);
Dispatcher(n);
}
_isrunning = false;
}

void Stop()
{
_isrunning = false;
}
~Reactor() {}
private:
// 1. epoll模型
std::unique_ptr<Epoller> _epoller_ptr;
// 2. 是否启动
bool _isrunning;
// 3. 管理所有的connection，本质是管理未来所有我获取到的fd
// fd : Connection
std::unordered_map<int, std::shared_ptr<Connection>> _connections;
// 4. 就绪的所有事件
struct epoll_event _evs[evs_num];
};

在LoopOnce中我们通过epoll_wait进行阻塞式的等待事件就绪（timeout 设置为-1）

注意：这里我们阻塞的目的是高效地等待。让出CPU，避免空转，仅在有事可做时被唤醒。我们之前说的实现非阻塞是指IO事件处理器（负责处理 epoll_wait返回的就绪事件）必须设置为非阻塞。通过 fcntl设置 O_NONBLOCK标志，将 socket 设为非阻塞模式。目的是不延误。确保处理单个fd的I/O时不会阻塞整个事件循环，从而快速服务所有连接。

在LoopOnce中事件就绪返回之后，我们就可以将事件派发给不同的事件处理器去处理

事件派发器代码如下：

// 事件派发器
void Dispatcher(int n)
{
for(int i = 0; i < n; i++)
{
int sockfd = _evs[i].data.fd; // 就绪的fd
uint32_t events = _evs[i].events; // 就绪的事件
// 1. 将所有的异常处理，统一转化成IO错误 2. 所有的IO异常，统一转换成为一个异常处理函数
if (events & EPOLLERR) // 出错
events |= (EPOLLIN | EPOLLOUT); // 将所有的异常处理，统一转化成IO错误
if (events & EPOLLHUP) // 对端退出
events |= (EPOLLIN | EPOLLOUT); // 将所有的异常处理，统一转化成IO错误

// 读事件就绪, 用不用区分是否异常?不用，未来会在IO处理器内部统一处理
if(events & EPOLLIN)
{
// 读事件就绪，还用不用区分是listenfd还是普通socketfd？不用，这里会多态调用自己的事件处理器
if(IsConnectionExist(sockfd))
_connections[sockfd]->Recver();
}

if(events & EPOLLOUT)
{
// 写事件就绪
if(IsConnectionExist(sockfd))
_connections[sockfd]->Sender();
}
}
}

在事件派发器中我们将所有的异常处理，统一转化成IO错误。为什么这么做？因为我们不仅仅在检测事件就绪时，文件描述符会出错，并且对端退出也会出错，然而我们在进行IO时也会出错，这些错误我们当然可以一个一个处理，但是我们如果统一转化为IO错误，最后在进行统一处理不就要高效简洁一点嘛

另外，我们这里读事件就绪后，就不再需要去判断是新连接到来，还是普通读事件就绪了，因为我们直接可以多态调用自己的事件处理器来处理

服务器需要将我们想要关心的fd和events写入到内核中，让内核帮我们关心，同时还需要添加到哈希表中管理，所以肯定需要增删接口来完成这些工作

void AddConnection(const std::shared_ptr<Connection>& conn)
{
if(IsConnectionExist(conn->GetSockfd()))
{
LOG(LogLevel::WARNING) << "conn is exists: " << conn->GetSockfd();
return;
}

// 1. conn对应的fd和他要关心的事件，写透到内核中！
int sockfd = conn->GetSockfd();
uint32_t events = conn->GetEvent();
_epoller_ptr->AddEvent(sockfd, events);

// 2. 设置当前conn的拥有者回指指针
conn->SetOwner(this);

// 3. 添加到哈希表中管理
_connections[sockfd] = conn;
}

void DelConnection(int sockfd)
{
if(!IsConnectionExist(sockfd))
{
LOG(LogLevel::WARNING) << "connection is not exist";
return;
}

// 从内核中移除关心fd
_epoller_ptr->DelEvent(sockfd);

// 从哈希表中移除，不再管理
_connections.erase(sockfd);

// 关闭fd
close(sockfd);

LOG(LogLevel::INFO) << "client quit: " << sockfd;
}

5. 补充细节

5.1 读事件处理

Listener派生类——接受新连接

新连接就绪了，但是你不能保证只有一个新连接，因此我们需要一次性将所有连接读取上来进行连接

这里我们需要将之前封装的Socket稍微修改一下，我们需要通过错误码errno判断具体的错误原因，因为有可能将底层所有连接读取完时，错误码表示的意思并非是出错了，而是暂时数据还没有就绪，这个时候就需要继续等待新连接到来了，还有可能是暂时被信号中断了，OS优先去执行其他任务了，这个时候我们后续需要继续循环执行。

#define ACCEPT_DONE –1
#define ACCEPT_CONTINUE –2
#define ACCEPT_ERR –3
int Accept(InetAddr *client) override
{
struct sockaddr_in peer;
socklen_t len = sizeof(peer);
int fd = ::accept(_sockfd, (struct sockaddr*)&peer, &len);
if (fd < 0)
{
if(errno == EAGAIN || errno == EWOULDBLOCK)
{
return –1; // 底层没有新连接
}
else if(errno == EINTR)
{
return –2; // 被信号中断，继续读取
}
else
{
LOG(LogLevel::WARNING) << "accept error";
return –3; // 出错
}
}
client->SetAddr(peer);
return fd;
}

这里我们只需要不断循环的接受新连接，直到将所有连接获取完毕，因为我们一开始在构造函数就设置了非阻塞，所以并不会阻塞整个进程的执行

void Recver() override
{
InetAddr client;
// 新连接就绪了，你能保证只有 一个连接到来吗？ 我们要一次把所有的连接全部获取上来
// while, ET, sockfd设置为非阻塞！！ —- listensock本身设置为非阻塞
while(true)
{
int sockfd = _listensockfd->Accept(&client);
if(sockfd == ACCEPT_DONE)
break; // 这个时候我们可以保证新连接都获取完了
else if(sockfd == ACCEPT_CONTINUE)
continue;
else if(sockfd == ACCEPT_ERR)
break; // 真正出错就退出循环不需要再继续
else
{
// 此时连接建立成功，拿到的是普通读写套接字
// fd默认事件状态（缓冲区为空时）：读事件不就绪，写事件就绪
// 我们需要将普通读写套接字交给内核帮我们关心读事件，写事件默认就绪不需要关心
std::shared_ptr<Connection> conn = std::make_shared<Channel>(sockfd, client);
conn->SetEvent(EPOLLIN | EPOLLET); // ET模式

GetOwner()->AddConnection(conn);
}
}
}

当你连接建立成功，拿到普通读写套接字sockfd时，就需要将sockfd的读事件交给内核帮我们关心，因为fd的默认事件状态，是读事件不就绪，写事件就绪，同时还需要将sockfd添加到哈希表中方便服务器未来管理

Channel派生类——普通读事件

这里我们同样需要不断循环，来把本轮的数据全部读取到用户接受缓冲区上

// 问题1: 怎么保证我把本轮数据读取完毕？ while 循环 — 本层只解决IO问题 — done
// 问题2：即便是你把本轮数据读完，你怎么知道数据就有完整的报文，如果不完整呢？如果是多个报文呢？粘报问题？反序列化 — 引入协议的
void Recver() override
{
char buffer[SIZE];
while(true)
{
buffer[0] = 0;
ssize_t n = recv(_sockfd, buffer, sizeof(buffer)–1, 0);
if(n > 0)
{
buffer[n] = 0;
_inbuffer += buffer;
}
else if(n == 0)
{
Excepter();// 异常处理
return;
}
else
{
if(errno == EAGAIN || errno == EWOULDBLOCK)
{
break;
}
else if(errno == EINTR) // 信号中断
{
continue;
}
else
{
Excepter();// 异常处理
return;
}
}
}

if(!_inbuffer.empty())
{
// 回调交给上层处理
}

if(!_outbuffer.empty())
{
Sender();
}
}

之前我们说过，把所有异常处理统一转换为IO错误，那么在IO中，我们可以再统一交给异常函数去处理。但是，即便是你把本轮数据读完，你怎么知道数据就有完整的报文，如果不完整呢？如果是多个报文呢？出现粘报问题呢？

所以这里我们需要把用户接受缓冲区的数据交给上层的协议层，来进行对报文的解析拿到一条完整的报文，然后再反序列化得到结构化的数据，此时我们就可以在协议层中进行回调交给应用层来处理数据，应用层处理完数据后将结果回调到协议层，最后协议层再对结果进行反序列化和封装报文，再次回调回来写入用户发送缓冲区中

我们之前就已经自己定制过协议，同时还简单实现了应用层的计算器，所以我们这里可以再引入之前自己写的定制协议和网络计算器。

我们先在基类中新增一个回调注册函数

#pragma once

#include <iostream>
#include <memory>
#include "Epoller.hpp"
#include "Socket.hpp"
#include "Common.hpp"
#include "InetAddr.hpp"

class Reactor;

using handler_t = std::function<std::string (std::string &)>;

// 基类
class Connection
{
public:
Connection():_events(0), _owner(nullptr)
{}

virtual void Recver() = 0;
virtual void Sender() = 0;
virtual void Excepter() = 0;
virtual int GetSockfd() = 0;

void SetEvent(const uint32_t& events)
{
_events = events;
}

uint32_t GetEvent()
{
return _events;
}

void SetOwner(Reactor* owner)
{
_owner = owner;
}

Reactor* GetOwner()
{
return _owner;
}

void RegisterHandler(handler_t handler)
{
_handler = handler;
}

~Connection(){}
private:
uint32_t _events;
Reactor* _owner;
public:
handler_t _handler;
};

另外，在Listener派生类接受新连接时，将回调函数注册（回调函数的编写可以在Main.cc主函数中）

void Recver() override
{
InetAddr client;
// 新连接就绪了，你能保证只有 一个连接到来吗？ 我们要一次把所有的连接全部获取上来
// while, ET, sockfd设置为非阻塞！！ —- listensock本身设置为非阻塞
while(true)
{
int sockfd = _listensockfd->Accept(&client);
if(sockfd == ACCEPT_DONE)
break; // 这个时候我们可以保证新连接都获取完了
else if(sockfd == ACCEPT_CONTINUE)
continue;
else if(sockfd == ACCEPT_ERR)
break; // 真正出错就退出循环不需要再继续
else
{
// 此时连接建立成功，拿到的是普通读写套接字
// fd默认事件状态（缓冲区为空时）：读事件不就绪，写事件就绪
// 我们需要将普通读写套接字交给内核帮我们关心读事件，写事件默认就绪不需要关心
std::shared_ptr<Connection> conn = std::make_shared<Channel>(sockfd, client);
conn->SetEvent(EPOLLIN | EPOLLET); // ET模式
if(_handler != nullptr)
conn->RegisterHandler(_handler);
GetOwner()->AddConnection(conn);
}
}
}

我们之前自己写的协议包含了服务器和客户端两个部分，这里我们只需要实现服务器即可，所以我们可以将之前的客户端获取应答报文以及构建请求报文接口删除，同时修改一下获取请求报文。

在协议中我们可以一次处理一条完整的报文，不断解析完整报文的部分我们可以放在回调函数中

代码如下：

std::string Execute(std::string &json_package)
{
// 将json字符串反序列化
Request req;
bool ok = req.Deserialize(json_package);
if (!ok)
return std::string();
// 回调到应用层进行处理
Response res = _func(req);

// 将结果序列化
std::string jsonstr = res.Serialize();

// 封装报文
std::string message = Encode(jsonstr);

return message;
}

现在我们就可以将用户接受缓冲区中的数据回调交给协议层处理

void Recver() override
{
char buffer[SIZE];
while(true)
{
buffer[0] = 0;
ssize_t n = recv(_sockfd, buffer, sizeof(buffer)–1, 0);
if(n > 0)
{
buffer[n] = 0;
_inbuffer += buffer;
}
else if(n == 0)
{
Excepter();// 异常处理
return;
}
else
{
if(errno == EAGAIN || errno == EWOULDBLOCK)
{
break;
}
else if(errno == EINTR) // 信号中断
{
continue;
}
else
{
Excepter();// 异常处理
return;
}
}
}

LOG(LogLevel::DEBUG) << "Channel: Inbuffer:\\n" << _inbuffer;

if(!_inbuffer.empty())
{
// 回调交给上层处理
_outbuffer += _handler(_inbuffer);
}

if(!_outbuffer.empty())
{
Sender();
}
}

5.2 写事件处理

void Sender() override
{
while(true)
{
ssize_t n = send(_sockfd, _outbuffer.c_str(), _outbuffer.size(), 0);
if(n > 0)
{
_outbuffer.erase(0, n);
// 发送缓冲区为空时退出循环
if(_outbuffer.empty())
break;
}
else if(n == 0)// 对端关闭
{
Excepter(); // 异常处理
return;
}
else
{
if(errno == EAGAIN || errno == EWOULDBLOCK)
{
break;
}
else if(errno == EINTR) // 信号中断
{
continue;
}
else
{
Excepter();// 异常处理
return;
}
}
}

// 1. 数据发送完毕
// 2. 发送条件不具备
if(!_outbuffer.empty())
{
// 此时用户发送缓冲区还有数据，说明内核缓冲区满了，即发送条件不具备
// 此时开启对写事件的关心
GetOwner()->EnableReadWrite(_sockfd, true, true);
}
else
{
// 到这里数据发送完毕
GetOwner()->EnableReadWrite(_sockfd, true, false);
}
}

同样，我们需要不断循环将数据写入内核缓冲区，如果循环结束，我们就需要判断是数据发送完毕还是内核缓冲区已经满了，不能在写了，后者我们就需要让内核帮我们关注写事件。

另外这两种情况下我们都需要让内核帮我们关注读事件，因为我们是ET模式，只要内核缓冲区状态发生变化，就会通知我们读事件就绪，我们必须在读事件就绪时就一次性全部读上来。在这里我们将读取到用户接受缓冲区里的数据回调交给协议层处理，然后把结果回调回来写到用户发送缓冲区中，说明用户接收缓冲区又有空间了，那么我们就需要关注读事件是否就绪。因为只要用户接受缓冲区没满，就需要关注读事件。

所以我们还需要在服务器中实现一个使能读写的接口，让内核帮我们关注读写事件

void EnableReadWrite(int sockfd, bool enableread, bool enablewrite)
{
if (!IsConnectionExist(sockfd))
{
LOG(LogLevel::WARNING) << "EnableReadWrite, conn is not exists, sockfd: " << sockfd;
return;
}

uint32_t new_event = EPOLLET | (enableread ? EPOLLIN : 0) | (enablewrite ? EPOLLOUT : 0);
_connections[sockfd]->SetEvent(new_event);
_epoller_ptr->ModEvent(sockfd, new_event);
}

5.3 异常事件处理

所以错误都被归总在了异常处理这里

void Excepter() override
{
// 所有的异常，都被我归一到了这个函数内部！！
GetOwner()->DelConnection(_sockfd);
}

直接从内核移除关心，然后从哈希表中移除管理，最后关闭文件描述符，而这些我们已经DelConnection这个接口中实现，直接调用即可

6. 主函数运行服务器

至此，我们从下至上实现了一个高效的epoll服务器

主函数中我们需要从上而下构建不同模块的对象

代码如下：

#include <iostream>
#include <string>
#include "Reactor.hpp"
#include "Listener.hpp"
#include "Channel.hpp"
#include "Log.hpp"
#include "Common.hpp"
#include "Protocol.hpp"
#include "NetCal.hpp"

int main(int argc, char *argv[])
{
if (argc != 2)
{
std::cout << "Usage: " << argv[0] << " port" << std::endl;
exit(USAGE_ERR);
}
Enable_Console_Log_Strategy();
uint16_t port = std::stoi(argv[1]);

// 构建一个业务模块
std::shared_ptr<Cal> cal = std::make_shared<Cal>();

// 构建协议对象
std::shared_ptr<Protocol> protocol = std::make_shared<Protocol>([&cal](Request &req) -> Response
{ return cal->Execute(req); });
// 构建Listener对象
std::shared_ptr<Connection> conn = std::make_shared<Listener>(port);
conn->RegisterHandler([&protocol](std::string &inbuffer) -> std::string { // 这个匿名函数就是要被Channel回调的
LOG(LogLevel::DEBUG) << "进入到匿名函数中…";
std::string response_str;
while (true)
{
std::string package;
if (!protocol->Decode(inbuffer, &package))
break;
// 我敢保证，我的packge一定是一个完整的请求，是字节流的
response_str += protocol->Execute(package);
}
LOG(LogLevel::DEBUG) << "结束匿名函数…: " << response_str;
return response_str;
});
// 构建一个reactor模块
std::unique_ptr<Reactor> R = std::make_unique<Reactor>();
R->AddConnection(conn);

R->Loop();

return 0;
}

Cal对象：这是应用层业务逻辑的核心，负责执行具体的计算任务。

Protocol对象：这是应用层协议解析器，负责两件关键事情：

• Decode：将接收到的原始字节流（inbuffer）根据预定规则（如长度、分隔符）拆分成一个个完整的请求包（package）。
• Execute：它本身不处理业务，而是将拆分好的请求包传递给业务逻辑模块（cal） 进行处理，并获取返回结果。这样做就实现了网络层与业务层的解耦。

Listener对象：它封装了监听套接字，负责在指定端口上接受新的客户端连接。它是一个特殊的Connection，即派生类。

注册回调函数：这是事件驱动模型的关键。这里为一个匿名函数（Lambda表达式） 注册到了监听器的数据到达事件上。当这个连接上有数据可读时（用户接收缓冲区上有数据时），Reactor就会自动调用这个函数。

Reactor对象：它是整个服务器的调度中心。服务器内部使用了 epoll 这样的 I/O 多路复用技术，在一个或少量线程中同时监控所有连接（包括监听套接字和已连接的客户端套接字）上的事件（如新连接到达、数据可读、可写等）。

Loop()方法：这是一个无限循环。在每次循环中，Reactor会阻塞在 epoll_wait这类调用上。一旦有任何被监控的连接上发生了它关心的事件（比如监听套接字上有新的连接请求，或某个客户端连接上有数据到达），epoll_wait就会返回。Reactor随后会根据事件的类型，通过哈希表找到对应的 Connection对象。如果是数据可读，就会多态调用Channel派生类中的Recver()，将内核缓冲区中的数据读取到用户接受缓冲区中，并触发其事先注册好的回调函数。

我们来运行测试一下，由于我们只实现了服务器，并没有实现客户端，所以我们这里可以使用之前网络计算机的客户端来进行测试

运行结果：

至此，我们实现了单 Reactor 单线程模型

7. 总结

在实践中，根据如何处理接收连接（Accept）和I/O读写（Read/Write）操作，Reactor模式演化出三种主要模型，其性能和处理能力依次提升。

模式对比与优劣分析

了解不同模型后，我们再从宏观视角对比一下Reactor模式与其他传统模式的区别，以及其自身的优缺点。

基于以上对比，Reactor模式的优缺点可以总结为：

• 优点：
  • 高并发与可扩展性：能够用少量线程处理大量连接，资源利用率高。
  • 解耦与模块化：将事件分发与业务处理分离，结构清晰。
  • 响应性：避免了因单个I/O操作阻塞而影响其他请求的处理。
• 缺点：
  • 编程复杂度高：基于回调的异步编程模型容易导致代码支离破碎（回调地狱），不易调试。
  • 对CPU密集型任务不友好：如果在事件处理线程中执行耗时计算，会严重拖慢整个系统的响应速度。通常需要与线程池结合，将计算任务卸载。

总结

总而言之，Reactor模式通过事件驱动和I/O多路复用技术，巧妙地解决了高并发网络编程中的资源瓶颈问题。它特别适合构建需要处理成千上万个并发连接的I/O密集型系统，如Web服务器、即时通讯网关和微服务框架等。

虽然编程模型相对复杂，但其在性能和资源消耗上的巨大优势，使其成为现代高性能网络编程的基石之一。知名的网络框架如Netty、Muduo等，都是Reactor模式的优秀实践。