---

✨✨✨项目地址在这里 ✨✨✨

✨✨✨https://gitee.com/penggli_2_0/TcpServer✨✨✨

---

1 高并发服务器

实现高并发服务器的基础是实现基于事件触发的Reactor模型，通过Reactor模型对事件进行统一管理。对此我们需要设计：

• 事件管理模块：只负责事件的管理，及时更新对应描述符的监控事件集，以及获取就绪事件集
• 多路转接模块：这个模块是对Epoll多路转接模型的封装，封装必要接口，方便使用。通过Channel获取监控事件集，放入epoll模型中进行监控。最终通过事件循环Poll 方法实时获取就绪活跃Channel。
• 反应堆模块：这个模块是对基于事件触发的Reactor模型，对所有描述符进行事件监控。内部封装Poller和定时器时间轮。根据就绪事件集执行任务池方法。同时监控定时器任务，定期执行定时任务！
• 连接模块：这是该项目中最重要的一个模块！该模块就是对连接进行全方位的管理，对上面的模块进行整合，通信连接的所有操作都是通过这个模块进行！对该连接描述符的回调函数都在这里进行设置，同时也负责如何处理连接数据。
• 监听模块：专门对监听套接字进行管理的类，本质就是简易的Connection，其中负责连接套接字的数据处理！

2 事件管理 Channel模块

1. 成员变量：
   • 当前需要监控的事件集
   • 当前连接触发的事件集
   • 绑定的EventLoop对象，这是管理当前连接的对象。Channel只是对事件进行管理
   • 可读事件触发的回调函数
   • 可写事件触发的回调函数
   • 错误事件触发的回调函数
   • 断开事件触发的回调函数
   • 任意事件触发的回调函数
2. 成员函数：
   • 设置Revents函数
   • 检查当前是否可读
   • 检查当前是否可写
   • 启动/关闭可写监控
   • 启动/关闭可读监控
   • 关闭所有事件的监控
   • 移除监控
   • 更新事件监控：将新的事件集更新到EventLoop中，进行监控
   • 设置回调函数
   • HandleEvent用来进行事件处理！

// Channel类管理事件集
class Poller;
class EventLoop;class Channel
{
private:int _fd;// Poller *_poller;EventLoop *_loop;uint32_t _events;  // 需要监控的事件集uint32_t _revents; // 当前连接就绪的事件集using EventCallBack = std::function<void()>;// 五个回调函数EventCallBack _read_cb;  // 可读事件回调函数EventCallBack _write_cb; // 可写事件回调函数EventCallBack _close_cb; // 连接断开事件回调函数EventCallBack _error_cb; // 错误事件回调函数EventCallBack _event_cb; // 任意事件回调函数
public:Channel(EventLoop *loop, int fd) : _fd(fd), _loop(loop), _events(0), _revents(0) {}// 设置回调函数void SetReadCallBack(const EventCallBack &cb) { _read_cb = cb; }void SetWriteCallBack(const EventCallBack &cb) { _write_cb = cb; }void SetCloseCallBack(const EventCallBack &cb) { _close_cb = cb; }void SetErrorCallBack(const EventCallBack &cb) { _error_cb = cb; }void SetEventCallBack(const EventCallBack &cb) { _event_cb = cb; }// 设置Revents函数void SetRevents(uint32_t events){// LOG(DEBUG, "事件更新revents:%d\n", events);_revents = events;}// 返回需要监控的事件集uint32_t Events() { return _events; }int Fd() { return _fd; }// 检查当前是否可读bool Readable() { return (_events & EPOLLIN); }// 检查当前是否可写bool Writeable() { return (_events & EPOLLOUT); }// 启动/关闭可写监控void EnableRead(){LOG(DEBUG, "fd:%d 加入EPOLLIN监控\n", _fd);_events |= EPOLLIN;Update();}void DisableRead(){_events &= ~EPOLLIN;Update();}// 启动/关闭可读监控void EnableWrite(){_events |= EPOLLOUT;Update();}void DisableWrite(){_events &= ~EPOLLOUT;Update();}// 关闭所有事件的监控void DisableAll(){_events = 0;Update();}// 移除监控 --- 涉及poller 要在poller之后进行实现void Remove();// 更新Channel的事件监控void Update();// HandleEvent用来进行事件处理！void HandleEvent(){// 根据revents判断需要执行哪些回调函数//      可读事件                  半关闭连接                   带外数据 紧急数据if ((_revents & EPOLLIN) || (_revents & EPOLLRDHUP) || (_revents & EPOLLPRI)){if (_read_cb)_read_cb();// 不管任何事件都要调用的回调函数！if (_event_cb)_event_cb(); // 放到事件处理完之后调用 刷新活跃度}// --- 有可能释放连接的操作事件 ， 一次只处理一个！---// 可写事件if (_revents & EPOLLOUT){if (_write_cb)_write_cb();// 不管任何事件都要调用的回调函数！if (_event_cb)_event_cb(); // 放到事件处理完之后调用 刷新活跃度}// 错误事件else if (_revents & EPOLLERR){// 不管任何事件都要调用的回调函数！if (_event_cb)_event_cb();// _event_cb必须放在前面 ，因为出错就会释放连接if (_error_cb)_error_cb();}// 连接关闭else if (_revents & EPOLLHUP){if (_event_cb)_event_cb();// _event_cb必须放在前面 ，因为出错就会释放连接if (_close_cb)_close_cb();}}
};
//...
//...
void Channel::Remove() { return _loop->RemoveEvent(this); }
// 更新Channel的事件监控
void Channel::Update() { return _loop->UpdateEvent(this); }

3 多路转接 Poller模块

Poller模块是对描述符IO进行监控的模块。

其中对多路转接的接口进行封装：

1. 添加/修改描述符的事件监控，存在就修改，不存在就添加！
2. 移除描述符的事件监控

所以：

1. 必须要有一个epoll模型
2. 一个struct epoll_event数组，监控时获取所有的活跃事件
3. 使用哈希表管理描述符与事件管理的Channel对象。

运行逻辑为：

• 对描述符进行监控，通过Channel才能知道描述符需要监控什么事件
• 当描述符就绪了，通过描述符在hash表中找到对应的Channel，只有找到了Channel才知道要调用什么回调函数！

这里最核心的部分：事件循环Poll 方法。
Poll 方法是事件循环的核心，它调用epoll_wait阻塞等待事件发生。当有事件发生时，epoll_wait 返回，Poll 方法遍历返回的事件列表 _evs，根据事件对应的文件描述符在 _event_channels 中找到对应的Channel对象，设置事件类型，并将其加入到活跃事件列表 active 中，以供上层进行后续处理。

// 多路转接方法Poller
#define MAX_POLLERSIZE 1024class Poller
{
private:int _epfd; // Epoll模型struct epoll_event _evs[MAX_POLLERSIZE];std::unordered_map<int, Channel *> _event_channels; // fd与Channel的映射表
private:void Update(Channel *channel, int op){// 根据channel初始化struct epoll_event ev;ev.data.fd = channel->Fd();ev.events = channel->Events();int ret = ::epoll_ctl(_epfd, op, channel->Fd(), &ev);if (ret < 0){LOG(ERROR, "epoll_ctl failed!\n");::abort(); // 直接退出程序}return;}bool HasChannel(Channel *channel){auto it = _event_channels.find(channel->Fd());if (it == _event_channels.end()){return false;}return true;}public:// 构造函数Poller(){_epfd = ::epoll_create(MAX_POLLERSIZE);if (_epfd < 0){LOG(FATAL, "epoll_create failed!\n");::abort();}}// 更新Eventvoid UpdateEvent(Channel *channel){// 先判断channel是否在channels中if (HasChannel(channel) == false){// 先建立托管//_event_channels[channel->Fd()] = channel;LOG(INFO, "Poller 加入新的fd事件托管 fd:%d\n", channel->Fd());_event_channels.insert(std::make_pair(channel->Fd(), channel));// 进行添加return Update(channel, EPOLL_CTL_ADD);}else{// 进行更新LOG(INFO, "Poller 加入事件托管 fd:%d\n", channel->Fd());return Update(channel, EPOLL_CTL_MOD);}}// 移除Eventvoid RemoveEvent(Channel *channel){// 先判断channel是否在channels中if (HasChannel(channel) == false){// 不在_event_channel里直接返回！return;}// 进行移除_event_channels.erase(channel->Fd());Update(channel, EPOLL_CTL_DEL);}// 开始监控事件void Poll(std::vector<Channel *> *active){// 进行监控// 阻塞式等待int nfds = ::epoll_wait(_epfd, _evs, MAX_POLLERSIZE, -1);if (nfds < 0){if (errno == EINTR)return;LOG(ERROR, "epoll_wait error:%s", strerror(errno));::abort();}// 对evs中的事件进行处理// LOG(DEBUG, "Poll 获取到新事件 n:%d\n", nfds);for (int i = 0; i < nfds; i++){auto it = _event_channels.find(_evs[i].data.fd);// 判断是否存在 不存在直接返回assert(it != _event_channels.end());// LOG(DEBUG, "channel : %d _evs[i].events:%d\n", _evs[i].data.fd, _evs[i].events);//   进行调用it->second->SetRevents(_evs[i].events);// 进行处理// it->second->HandleEvent();// 处理结束放入活跃队列active->push_back(it->second);}}
};

4 反应堆 EventLoop模块

EventLoop模块是管理事件监控管理的模块，就是Reactor反应堆模型。该模块与线程一一对应关联！

监控一个连接，这个连接一旦就绪，就要进行处理！如果这个连接描述符在多个线程中都触发了事件，就会存在线程安全问题！因此我们需要将一个连接的事件监控，以及连接事件处理和其他操作都放在同一个线程中进行处理！

后续如果接入了线程池，那么如何保证一个连接的所有操作都在EventLoop所在线程中？在EventLoop()中，添加一个任务队列。对连接的所有操作，都进行一次封装，对连接的操作要当做任务放入任务队列
事件监控 -> 事件处理（放入队列） -> 执行任务
这样可以保证对于连接的所有操作都是在一个线程中执行的，不涉及线程安全问题，是对于任务队列的操作有线程安全问题！只需要给task的操作加一把锁即可！

EventLoop处理流程

• 在线程中对描述符进行事件监控
• 有描述符就绪则对描述符进行事件处理，要保证处理回调函数的操作都在同一个线程中
• 所以的就绪事件处理完了，这时候再去将任务队列中的的任务进行执行

成员变量

1. 线程ID :当事件就绪时，需要处理时，处理过程中，如果连接要进行某种操作，这些操作必须在EventLoop对应线程中执行，保证对连接的各项操作都是线程安全的，可以根据ID判断所执行的操作是不是在所属线程中，如果在当前线程中可以直接执行；如果执行的操作不再线程中，才需要加入到任务池中，等待事件处理完然后执行任务
2. Poller模型：对连接描述符进行监控
3. 任务队列：执行任务
4. 事件通知eventfd：本质是一个计算器，每写入一次就就绪了一次。有可能等待描述符IO事件就绪，执行流阻塞，这时候任务队列将得不到执行,所以需要eventfd来唤醒事件监控的阻塞
5. _eventfd的管理Channel：这里使用智能指针！
6. 线程互斥锁：保证线程安全
7. TimeQueue时间轮模块：添加定时任务，负责对超时非活跃连接进行销毁

成员函数：

1. 构造函数: 初始化成员变量 ，设置Channel的回调函数ReadEvent，启动可读事件监控
2. IsInLoop函数：判断当前线程是否是EventLoop对应线程
3. 执行任务RunInLoop函数：判断将要执行的任务是否处于当前线程，如果是直接执行回调；否则压入任务池中！这个接口是提供给外部的！
4. 将操作压入任务池 QueueInLoop函数：将操作压入任务池，注意要进行上锁！
5. 唤醒函数WakeUpEventfd：有可能因为没有事件就绪导致的epoll阻塞！进行唤醒
6. 添加/修改监控： 更新对应Channel的事件监控。
7. 移除监控：移除对应Channel的事件监控。
8. 启动函数：进行事件监控，通过Poller 获取活跃连接；根据活跃连接进行事件处理， 遍历活跃连接，进行事件回调。然后执行所有任务。
9. 执行所有任务RunAllTask：将任务池的任务置换出来，然后进行执行，注意加锁！
10. 针对_eventfd的处理 :
    • 创建 int CreateEventfd () 标志位 EFD_CLOEXEC | EFD_NONBLOCK
    • 读取 void ReadEventfd 获取eventfd读取次
    • 唤醒 WakeUpEventfd 向Event中写入一个数据

timewheel 与 EventLoop 模块整合操作：

1. 通过timefd设置定时器，内核会定时向文件描述符写入，触发超时事件
2. 时间轮timewheel：实现每次执行RunTimeTask,都可以执行一波到期的任务！

要实现一个完整的秒级定时器，就需要将这两个功能整合到一起：

• timefd设置为每秒触发一次定时事件，当事件触发，则运行一次timewheel的runtimertask,执行一下所有的过期任务
• timefd的事件的监控与触发，可以融合EventLoop来进行实现！

修改时间轮timewheel模块：

1. 增加定时器 _timerfd：设置读回调函数 并放入EvnetLoop中启动监控
2. 对_timerfd进行管理的 _timer_channel：使用智能指针进行管理！
3. 增添EventLoop：绑定对应的EventLoop！
4. 创建Createtimefd接口 ：设置为超时时间为 1 S
5. ReadTimerfd接口：读取timerfd的数据
6. Ontime接口：触发超时事件，进行读取timerfd , 进行超时任务的处理RunTimetask
7. 添加/刷新/取消定时任务 ：这三个接口设计了_timers的操作，会有线程安全问题！必须在EventLoop所在线程执行，都要通过EventLoop进行 RunInLoop操作！

/ EventLoop类 --- Reactor反应堆模型，管理监控连接事件
using Functor = std::function<void()>;
class EventLoop
{
private:std::thread::id _event_id;               // 线程IDint _eventfd;                            // eventfd 用于通知事件std::unique_ptr<Channel> _event_channel; // 管理Event事件的Channel对象Poller _poller;                          // epoll模型std::vector<Functor> _tasks; // 任务池std::mutex _mtx;             // 互斥锁保护线程// 时间轮TimeWheel _timer_wheel;private:int CreateEventfd(){//                      禁止进程复制   启动非阻塞读取int efd = ::eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);if (efd < 0){LOG(ERROR, "Eventfd Create failed!\n");return -1;}return efd;}void ReadEvent(){// 从eventfd中读取数据// 注意每次读取的都是一个8字节数据uint64_t ret;int n = ::read(_eventfd, &ret, sizeof(uint64_t));if (n <= 0){//    被信号打断           表示没有数据if (errno == EINTR || errno == EAGAIN)return;LOG(ERROR, "eventfd recv failed!\n");::abort();}// 读取数据执行任务return;}void WeakUpEventfd(){uint64_t val = 1;int n = ::write(_eventfd, &val, sizeof(uint64_t));if (n <= 0){if (errno == EINTR)return;LOG(ERROR, "eventfd send failed!\n");::abort();}// 读取数据执行任务return;}public:EventLoop() : _event_id(std::this_thread::get_id()),_eventfd(CreateEventfd()),_event_channel(new Channel(this, _eventfd)),_timer_wheel(this){//_poller = Poller();// 设置Eventfd的读回调函数_event_channel->SetReadCallBack(std::bind(&EventLoop::ReadEvent, this));// 设置读事件监控_event_channel->EnableRead();LOG(DEBUG, "EventLoop 构造完成\n");}bool IsInLoop(){return (_event_id == std::this_thread::get_id());}void AssertInLoop(){return assert(_event_id == std::this_thread::get_id());}void RunInLoop(const Functor &cb){if (IsInLoop())return cb();// 否则压入任务池return QueueLoop(cb);}void QueueLoop(const Functor &cb){{std::unique_lock<std::mutex> lock(_mtx);_tasks.push_back(cb);}// 唤醒有可能因为没有事件就绪导致的epoll阻塞！WeakUpEventfd();}// 添加/修改监控void UpdateEvent(Channel *channel){return _poller.UpdateEvent(channel);}// 移除监控void RemoveEvent(Channel *channel){return _poller.RemoveEvent(channel);}void RunAllTask(){std::vector<Functor> tasks;{// 进行上锁std::unique_lock<std::mutex> lock(_mtx);_tasks.swap(tasks);}// 进行执行函数for (auto &f : tasks){f();}return;}// 开始监控函数void Start(){while (1){std::vector<Channel *> actives;_poller.Poll(&actives);// 事件处理（放入队列） 遍历活跃连接，进行事件回调for (auto &channel : actives){channel->HandleEvent();}// 执行任务 执行任务RunAllTaskRunAllTask();}}// 增加时间轮系列接口void TimerAdd(uint64_t id, int delay, Task_t cb) { _timer_wheel.TimerAdd(id, delay, cb); }void TimerRefresh(uint64_t id) { _timer_wheel.TimerRefresh(id); }void TimerCancel(uint64_t id) { _timer_wheel.TimerCancel(id); }bool HasTimer(uint64_t id) { return _timer_wheel.HasTimer(id); }
};

5 核心模块 - 连接Connection模块

这是该项目中最重要的一个模块！
该模块就是对连接进行全方位的管理，对通信连接的所有操作都是通过这个模块进行！

需要管理：

1. 套接字的管理，可以进行套接字操作
2. 连接事件的管理，可读，可写，错误，挂断，任意事件
3. 缓冲区的管理，从Socket读取/发送数据 需要经过缓冲区，便于Socket数据的接收与发送
4. 协议上下文的管理，记录请求数据的处理过程

• 因为连接接收到数据之后要如何处理，需要用户决定，因此必须需要业务处理回调函数！
• 一个连接建立成功之后，该如何处理，由用户决定！因此必须有连接建立成功的回调函数！
• 同样关闭前，需要如何处理，也由用户决定，因此必须由关闭连接回调函数。
• 任意事件的产生，需不需要某种处理，由用户决定，因此必须由任意事件的回调函数！

提供功能：

1. 发送数据 — 给用户提供的发送数据接口，只是将数据拷贝到发送缓冲区，然后启动写事件监控
2. 关闭连接 — 给用户提供的关闭连接接口，应该在实际释放连接之前，查看输入输出缓冲区是否有数据待处理‘
3. 超时管理 — 通过给用户接口，用来 启动/取消 超时销毁功能
4. 协议切换 — 一个连接接收数据后如何进行业务处理，取决于上下文，以及数据的业务处理函数

Connection模块是对连接的管理模块，对于连接的所有操作都是通过这个模块完成的！

当这样的一个场景中：对连接进行操作的时候，但是连接已经被释放了，导致内存访问错误，程序崩溃！
对于这个场景的解决方案：使智能指针share_ptr对Connection进行管理,只有计数为0时才会真正释放！

Connection要继承enable_shared_from_this<Connection>，这样可以方便的快速获取当前对象的shared_ptr指针。
成员变量：

1. 连接ID :uint64_t _conn_id ：便于连接的管理和查找

2. 套接字描述符：int sockfd ：连接关联的文件描述符

3. 套接字对象：Socket _socket：套接字操作管理模块

4. 事件管理： Channel _channel：连接事件管理模块

5. 输入输出缓冲区:

   • Buffer _in_buffer; 存放Socket中读取的数据
   • Buffer _out_buffer; 存放要发送给对端的数据

6. 通用数据类型上下文: Any _context;

7. 连接状态：ConnStatu _statu; DISCONNECTED 未连接 ; CONNECTING 连接建立成功 待处理状态 ;CONNECTED 连接建立完成 可以通信; DISCONNETCING 待关闭。

8. 5种回调函数 ：

   • 连接回调函数类型为 std::function<void(const PtrConn&)>;
   • 处理时回调函数 std::function<void(const PtrConn& , Buffer*)>;
   • 关闭阶段的回调 std::function<void(const PtrConn&)>;
   • 还需要组件内的连接关闭回调 因为服务器组件内会把所有的连接管理起来 一旦某个连接关闭 就应该从管理的地方移除自己的信息！
   • 任意事件触发的回调 std::function<void(const PtrConn&)>;

9. 超时销毁标志位 bool _enable_inactive_release： 判断是否启动非活跃销毁的标志，默认是false

10. 定时器ID :可以直接简化为使用_conn_id。

11. Reactor模型 EventLoop _loop*： 连接所关联的EventLoop。

成员函数：

1. 线程内发送数据接口： void SendInLoop(char* data , size_t len)
2. 线程内关闭连接接口：void ShutdownInLoop()，这个关闭操作不是真正的关闭函数 而是判断是否还有时间要进行处理
3. 线程内开启超时销毁接口：void EnableInactiveReleaseInLoop(int sec)
4. 线程内取消超时销毁接口：void CancelInactiveReleaseInLoop()
5. 线程内设置回调函数接口：void UpgradeInLoop(Any Context , const ConnectionCallback …)
6. 实际的释放接口：void ReleaseInLoop()
7. 初始化设置接口：EstablishedInLoop() 连接获取之后所处的状态下要进行各种设置 EstablishedInLoop() 给Channel设置事件回调 启动读监控
8. 4个Channel回调函数 ：
   • HandleRead ：接收Socket数据放到接收缓冲区中 使用非阻塞读取，读取出错调用ShutdownInLoop 检查缓冲区再进行关闭，将读取到的数据写入输入缓冲区，然后根据输入缓冲区数据大小调用_message_callback回调进行业务处理
   • HandleSend： 描述符可写事件就绪后要调用的函数，将发送缓冲区的数据发送，_out_buffer中保存的就是要发送的数据，根据返回值进行处理，发送错误就要关闭连接了 关闭之前如果还有数据需要进行处理 ，千万不能忘记将读偏移向后移动！如果现在的状态是待关闭状态，则有数据发送完之后释放连接 ，没有数据就直接关闭，输出缓冲区没有数据了就不用再监控写事件了！

• HandleClose：关闭连接
• HandleError：出现错误，直接关闭连接
• HandleEvent：刷新定时销毁任务活跃度

用户接口：

1. 构造函数、析构函数
2. 获取Id
3. 获取描述符fd
4. 是否处于连接状态 bool ConnStatu()
5. 设置上下文 void SetContext(const Any & context)
6. 获取上下文 Any* GetContext()
7. 设置4个回调函数 SetConnectCB(const ConnectionCallback& cb)…
8. 发送数据 void Send(char* data , size_t len) 将数据放到发送缓冲区，启动写事件监控
9. 关闭连接 void Shutdown() 提供给用户的关闭，不是真正的关闭连接 ， 需要判断有没有数据待处理
10. 取消超时销毁 void CancelInactiveRelease()
11. 协议切换 Upgrade(Any Context , const ConnectionCallback …) 切换协议 — 重置上下文以及阶段性处理函数
12. 进行channel回调设置 Established() 启动读监控

class Connection;// 核心模块 --- Connection类
typedef enum
{DISCONNECTED = 0,CONNECTING,CONNECTED,DISCONNETCING
} ConnStatu;using PtrConn = std::shared_ptr<Connection>;class Connection : public std::enable_shared_from_this<Connection>
{private:uint64_t _conn_id;           // connection连接IDSocket _socket;              // 管理的套接字int _sockfd;                 // 套接字fdEventLoop *_loop;            // connection连接关联的EventLoop对象Any _context;                // 上下文数据Channel _channel;            // 管理连接事件Buffer _in_buffer;           // 输入缓冲区 存放Socket中读取的数据Buffer _out_buffer;          // 输出缓冲区 存放要发送给对端的数据bool _enable_active_release; // 是否开启超时销毁 默认是falseConnStatu _statu;            // Connection连接状态// 5 个 回调函数 --- 注意使用智能指针 防止在执行任务之前Connection销毁using ConnectedCallBack = std::function<void(const PtrConn &)>;         // 连接时进行的回调函数using MessageCallBack = std::function<void(const PtrConn &, Buffer *)>; // 处理数据时的回调函数using ClosedCallBack = std::function<void(const PtrConn &)>;            // 关闭连接时的回调函数using AnyEventCallBack = std::function<void(const PtrConn &)>;          // 处理任意事件时的回调函数ConnectedCallBack _conn_cb;  // 连接回调函数类型MessageCallBack _message_cb; // 处理时回调函数ClosedCallBack _closed_cb;   // 关闭阶段的回调AnyEventCallBack _event_cb;  // 任意事件触发的回调// 还需要组件内的连接关闭回调 因为服务器组件内会把所有的连接管理起来 一旦某个连接关闭 就应该从管理的地方移除自己的信息！ClosedCallBack _event_closed_cb;private:// 读事件触发的函数void HandleRead(){// 接收Socket数据放到接收缓冲区中// LOG(DEBUG, "HandleRead\n");char buf[65536] = {0};int ret = _socket.NonBlockRecv(buf, 65536);// 返回值 为 - 1说明读取错误if (ret < 0){return ShutdownInLoop();}// 返回值为0说明没读取到数据// 将数据写入到缓冲区_in_buffer.WriteAndPush(buf, ret);// 然后调用_message_callback回调if (_in_buffer.ReadAbleSize() > 0){return _message_cb(shared_from_this(), &_in_buffer);}}// 写事件触发的函数void HandleWrite(){// 将输出缓冲区的数据向Socket描述符中进行非阻塞写入int ret = _socket.NonBlockSend(_out_buffer.ReadPos(), _out_buffer.ReadAbleSize());// 发送错误就关闭连接if (ret < 0){// 如果输入缓冲区还有数据，就要进行处理之后再关闭连接if (_in_buffer.ReadAbleSize() > 0)_message_cb(shared_from_this(), &_in_buffer);return ReleaseInLoop();}// 缓冲区读偏移向后移动_out_buffer.MoveReadOffset(ret);if (_out_buffer.ReadAbleSize() == 0){// 关闭写事件监控_channel.DisableWrite();// 如果是待关闭状态就进行关闭if (_statu == DISCONNETCING){return Release();}}return;}// 连接关闭触发的函数void HandleClose(){// 如果还有数据，就进行一次处理if (_in_buffer.ReadAbleSize() > 0){_message_cb(shared_from_this(), &_in_buffer);}// 释放连接return Release();}// 错误事件触发的函数void HandleError(){return HandleClose(); // 直接调用关闭函数}void HandleEvent(){// 刷新活跃度if (_enable_active_release == true){_loop->TimerRefresh(_conn_id);}if (_event_cb)_event_cb(shared_from_this());}// 发送函数 不是直接进行发送 而是将数据拷贝到输出缓冲区 然后启动写事件监控void SendInLoop(const char *data, size_t len){if (_statu == DISCONNECTED)return;_out_buffer.WriteAndPush(data, len);if (_channel.Writeable() == false){_channel.EnableWrite();}}// 这个关闭操作不是真正的关闭函数 而是腰判断是否还有时间要进行处理void ShutdownInLoop(){_statu = DISCONNETCING; // 状态设置为待关闭状态// 如果输入缓冲区有数据 ，要进行一次处理if (_in_buffer.ReadAbleSize() > 0){if (_message_cb)_message_cb(shared_from_this(), &_in_buffer);}// 如果输出缓存区有数据，就要启动写事件监控if (_out_buffer.ReadAbleSize() > 0){if (_channel.Writeable() == false)_channel.EnableWrite();}// 如果输出缓冲区数据没有 直接进行关闭if (_out_buffer.ReadAbleSize() == 0){Release();}}void EnableInactiveReleaseInLoop(int sec){// LOG(DEBUG, "EnableInactiveReleaseInLoop %d s\n", sec);//  将超时销毁标志位设置为true_enable_active_release = true;// 如果已经有超时任务，那么就进行一次刷新if (_loop->HasTimer(_conn_id)){return _loop->TimerRefresh(_conn_id);}// 没有就进行添加_loop->TimerAdd(_conn_id, sec, std::bind(&Connection::Release, this));}void CancelInactiveReleaseInLoop(){// 将超时销毁标志位设置为false_enable_active_release = false;// 如果有超时任务才进行取消if (_loop->HasTimer(_conn_id)){_loop->TimerCancel(_conn_id);}}void UpgradeInLoop(const Any &context,const ConnectedCallBack &conn_cb,const MessageCallBack &mess_cb,const ClosedCallBack &closed_cb,const AnyEventCallBack &event_cb){_context = context;_conn_cb = conn_cb;_message_cb = mess_cb;_closed_cb = closed_cb;_event_cb = event_cb;}// 真正的关闭函数void ReleaseInLoop(){LOG(INFO, "Realse Connection:%p\n", shared_from_this());// 1. 修改连接状态_statu = DISCONNECTED;// 2. 移除所有事件监控_channel.Remove();// 3. 关闭描述符_socket.Close();// 4. 取消定时任务if (_loop->HasTimer(_conn_id))CancelInactiveReleaseInLoop();// 5. 执行用户设置的关闭回调if (_closed_cb)_closed_cb(shared_from_this());// 6. 执行组件内的关闭回调if (_event_closed_cb)_event_closed_cb(shared_from_this());}// 连接获取之后所处的状态下要进行各种设置 启动读事件监控 调用回调函数void EstablishedInLoop(){// LOG(DEBUG, "EstablishedInLoop()\n");//  必须是连接中状态才执行assert(_statu == CONNECTING);_statu = CONNECTED;    // 1.执行完 修改为已连接状态_channel.EnableRead(); // 2.启动读事件监控if (_conn_cb)_conn_cb(shared_from_this()); // 3. 调用连接回调函数}public:Connection(EventLoop *loop, uint64_t conn_id, int sockfd) : _conn_id(_conn_id),_socket(sockfd),_sockfd(sockfd),_loop(loop),_channel(_loop, _sockfd),_enable_active_release(false),_statu(CONNECTING){// 设置channel回调函数_channel.SetCloseCallBack(std::bind(&Connection::HandleClose, this));_channel.SetErrorCallBack(std::bind(&Connection::HandleError, this));_channel.SetWriteCallBack(std::bind(&Connection::HandleWrite, this));_channel.SetEventCallBack(std::bind(&Connection::HandleEvent, this));_channel.SetReadCallBack(std::bind(&Connection::HandleRead, this));}~Connection(){LOG(INFO, "Release Connection :%p", this);}// 基础接口uint64_t Id() { return _conn_id; }                          // 返回Connection的idint Fd() { return _sockfd; }                                // 返回套接字描述符void SetContext(const Any &context) { _context = context; } // 设置上下文Any *GetContext() { return &_context; }                     // 获取上下文bool Connected() { return (_statu == CONNECTED); }          // 判断是否处于连接状态！//  设置回调函数void SetConnectCB(const ConnectedCallBack &cb) { _conn_cb = cb; }void SetMessageCB(const MessageCallBack &cb) { _message_cb = cb; }void SetClosedCB(const ClosedCallBack &cb) { _closed_cb = cb; }void SetAnyEventCB(const AnyEventCallBack &cb) { _event_cb = cb; }void SetSvrClosedCB(const ClosedCallBack &cb) { _event_closed_cb = cb; }// 发送数据void Send(const char *data, size_t len) { _loop->RunInLoop(std::bind(&Connection::SendInLoop, this, data, len)); }// 关闭连接 --- 提供给用户的关闭，_不是真正的关闭连接 ， 需要判断有没有数据待处理void Shutdown() { _loop->RunInLoop(std::bind(&Connection::ShutdownInLoop, this)); }// 启动超时销毁void EnableInactiveRelease(int sec) { _loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop, this, sec)); }// 取消超时销毁void CancelInactiveRelease() { _loop->RunInLoop(std::bind(&Connection::CancelInactiveReleaseInLoop, this)); }// 进行channel回调设置void Established() { _loop->RunInLoop(std::bind(&Connection::EstablishedInLoop, this)); }void Release(){// LOG(DEBUG, "Release()\n");_loop->QueueLoop(std::bind(&Connection::ReleaseInLoop, this));}// 切换协议void Upgrade(const Any &context,const ConnectedCallBack &conn_cb,const MessageCallBack &mess_cb,const ClosedCallBack &closed_cb,const AnyEventCallBack &event_cb){// 切换协议 --- 重置上下文以及阶段性回调处理函数 这个函数必须在EventLoop中立刻执行// 预防新事件触发后 ，处理时还是原先的协议！ --- 导致数据处理异常_loop->AssertInLoop();_loop->RunInLoop(std::bind(&Connection::UpgradeInLoop, this, context, conn_cb, mess_cb, closed_cb, event_cb));}
};

6 监听套接字 Acceptor模块

专门对监听套接字进行管理的类

1. 创建一个监听套接字
2. 启动读事件监控，获取新连接
3. 事件触发后，获取新连接
4. 为新连接创建Connection进行管理（这是服务器模块进行的）

该模块只进行监听连接的管理，因此获取到新连接的描述符之后，对于新连接描述符如何处理其实并不关心！

成员变量
1. 套接字对象：Socket 用于创建监听套接字
2. EventLoop* _loop ：对监听套接字进行事件管理
3. Channel _channel ：用于对监听套接字进行事件管理
4. 新连接获取之后的回调函数 AcceptCallBack AcceptCallBack _accept_callback;

成员函数
1. 构造函数
2. 监听套接字读事件回调函数 — 调用 _accept_callback，进行新连接处理
3. 创建套接字 返回描述符。

class Acceptor
{
private:Socket _socket;   // 套接字对象EventLoop *_loop; // 对监听套接字进行事件监控Channel _channel; // 用于对今天套接的事件管理using AcceptCallBack = std::function<void(int)>;AcceptCallBack _accept_callback;int CreateSocket(int port){bool ret = _socket.CreateServer(port);assert(ret == true);return _socket.Sockfd();}// 读事件回调函数void HandleRead(){// 获取新连接int newfd = _socket.Accept();if (newfd < 0){LOG(ERROR, "Accept failed\n");return;}if (_accept_callback)_accept_callback(newfd);}public:Acceptor(EventLoop *loop, int port) : _socket(CreateSocket(port)), _loop(loop), _channel(_loop, _socket.Sockfd()){_channel.SetReadCallBack(std::bind(&Acceptor::HandleRead, this));// 开启读事件监控！//_channel.EnableRead();}~Acceptor(){_socket.Close();}void SetAcceptCallBack(const AcceptCallBack &cb){_accept_callback = cb;}void Listen(){// 开启读事件监控！_channel.EnableRead();}
};

现在，服务器模块基本实现！