【网络】高级IO——Reactor版TCP服务器

1.什么是Reactor

1.1.餐厅里的Reactor模式

2.Reactor的由来

2.1.单 Reactor 单进程 / 线程

2.2.单 Reactor 多线程 / 多进程

2.3.多 Reactor 多进程 / 线程

3.实现单 Reactor 单进程版本的TCP服务器

3.1.Connection类

3.2.TcpServer类

3.3.Connection的真正用处

3.4.Dispatcher——事件派发器

3.5.回调函数

3.6.区分读写异常事件

3.7.读事件的处理

3.8.处理数据 ——（反）序列化/编解码

3.9.写事件的处理

3.10.异常事件的处理

3.11.源代码

3.12.总结

1.什么是Reactor

Reactor 是一种应用在服务器端的开发模式（也有说法称 Reactor 是一种 IO 模式），目的是提高服务端程序的并发能力

它要解决什么问题呢？

传统的 thread per connection 用法中，线程在真正处理请求之前首先需要从 socket 中读取网络请求，而在读取完成之前，线程本身被阻塞，不能做任何事，这就导致线程资源被占用，而线程资源本身是很珍贵的，尤其是在处理高并发请求时。

而 Reactor 模式指出，在等待 IO 时，线程可以先退出，这样就不会因为有线程在等待 IO 而占用资源。但是这样原先的执行流程就没法还原了，因此，我们可以利用事件驱动的方式，要求线程在退出之前向 event loop 注册回调函数，这样 IO 完成时 event loop 就可以调用回调函数完成剩余的操作。

所以说，Reactor 模式通过减少服务器的资源消耗，提高了并发的能力。当然，从实现角度上，事件驱动编程会更难写，难 debug 一些。

1.1.餐厅里的Reactor模式

我们用“餐厅”类比的话，就像下图：

对于每个新来的顾客，前台都需要找到一个服务员和厨师来服务这个顾客。

服务员给出菜单，并等待点菜
顾客查看菜单，并点菜
服务员把菜单交给厨师，厨师照着做菜
厨师做好菜后端到餐桌上

这就是传统的多线程服务器。每个顾客都有自己的服务团队（线程），在人少的情况下是可以良好的运作的。现在餐厅的口碑好，顾客人数不断增加，这时服务员就有点处理不过来了。

这时老板发现，每个服务员在服务完客人后，都要去休息一下，因此老板就说，“你们都别休息了，在旁边待命”。这样可能 10 个服务员也来得及服务 20 个顾客了。这也是“线程池”的方式，通过重用线程来减少线程的创建和销毁时间，从而提高性能。

但是客人又进一步增加了，仅仅靠剥削服务员的休息时间也没有办法服务这么多客人。老板仔细观察，发现其实服务员并不是一直在干活的，大部分时间他们只是站在餐桌旁边等客人点菜。

于是老板就对服务员说，客人点菜的时候你们就别傻站着了，先去服务其它客人，有客人点好的时候喊你们再过去。对应于下图：

最后，老板发现根本不需要那么多的服务员，于是裁了一波员，最终甚至可以只有一个服务员。

这就是 Reactor 模式的核心思想：减少等待。当遇到需要等待 IO 时，先释放资源，而在 IO 完成时，再通过事件驱动 (event driven) 的方式，继续接下来的处理。从整体上减少了资源的消耗。

2.Reactor的由来

如果要让服务器服务多个客户端，那么最直接的方式就是为每一条连接创建线程。

其实创建进程也是可以的，原理是一样的，进程和线程的区别在于线程比较轻量级些，线程的创建和线程间切换的成本要小些，为了描述简述，后面都以线程为例。

处理完业务逻辑后，随着连接关闭后线程也同样要销毁了，但是这样不停地创建和销毁线程，不仅会带来性能开销，也会造成浪费资源，而且如果要连接几万条连接，创建几万个线程去应对也是不现实的。

要这么解决这个问题呢？我们可以使用「资源复用」的方式。

也就是不用再为每个连接创建线程，而是创建一个「线程池」，将连接分配给线程，然后一个线程可以处理多个连接的业务。

不过，这样又引来一个新的问题，线程怎样才能高效地处理多个连接的业务？

当一个连接对应一个线程时，线程一般采用「read -> 业务处理 -> send」的处理流程，如果当前连接没有数据可读，那么线程会阻塞在 read 操作上（ socket 默认情况是阻塞 I/O），不过这种阻塞方式并不影响其他线程。

但是引入了线程池，那么一个线程要处理多个连接的业务，线程在处理某个连接的 read 操作时，如果遇到没有数据可读，就会发生阻塞，那么线程就没办法继续处理其他连接的业务。

要解决这一个问题，最简单的方式就是将 socket 改成非阻塞，然后线程不断地轮询调用 read 操作来判断是否有数据，这种方式虽然该能够解决阻塞的问题，但是解决的方式比较粗暴，因为轮询是要消耗 CPU 的，而且随着一个线程处理的连接越多，轮询的效率就会越低。

上面的问题在于，线程并不知道当前连接是否有数据可读，从而需要每次通过 read 去试探。

那有没有办法在只有当连接上有数据的时候，线程才去发起读请求呢？答案是有的，实现这一技术的就是 I/O 多路复用。

I/O 多路复用技术会用一个系统调用函数来监听我们所有关心的连接，也就说可以在一个监控线程里面监控很多的连接。

我们熟悉的 select/poll/epoll 就是内核提供给用户态的多路复用系统调用，线程可以通过一个系统调用函数从内核中获取多个事件。

select/poll/epoll 是如何获取网络事件的呢？

在获取事件时，先把我们要关心的连接传给内核，再由内核检测：

如果没有事件发生，线程只需阻塞在这个系统调用，而无需像前面的线程池方案那样轮训调用 read 操作来判断是否有数据。
如果有事件发生，内核会返回产生了事件的连接，线程就会从阻塞状态返回，然后在用户态中再处理这些连接对应的业务即可。

当下开源软件能做到网络高性能的原因就是 I/O 多路复用吗？

是的，基本是基于 I/O 多路复用，用过 I/O 多路复用接口写网络程序的同学，肯定知道是面向过程的方式写代码的，这样的开发的效率不高。

于是，大佬们基于面向对象的思想，对 I/O 多路复用作了一层封装，让使用者不用考虑底层网络 API 的细节，只需要关注应用代码的编写。大佬们还为这种模式取了个让人第一时间难以理解的名字：Reactor 模式。

Reactor 翻译过来的意思是「反应堆」，可能大家会联想到物理学里的核反应堆，实际上并不是的这个意思。

这里的反应指的是「对事件反应」，也就是来了一个事件，Reactor 就有相对应的反应/响应。

事实上，Reactor 模式也叫 Dispatcher 模式，我觉得这个名字更贴合该模式的含义，即 I/O 多路复用监听事件，收到事件后，根据事件类型分配（Dispatch）给某个进程 / 线程。

Reactor 模式主要由 Reactor 和处理资源池这两个核心部分组成，它俩负责的事情如下：

Reactor 负责监听和分发事件，事件类型包含连接事件、读写事件；
处理资源池负责处理事件，如 read -> 业务逻辑 -> send；

Reactor 模式是灵活多变的，可以应对不同的业务场景，灵活在于：

Reactor 的数量可以只有一个，也可以有多个；
处理资源池可以是单个进程 / 线程，也可以是多个进程 /线程；

将上面的两个因素排列组设一下，理论上就可以有 4 种方案选择：

单 Reactor 单进程 / 线程；
单 Reactor 多进程 / 线程；
多 Reactor 单进程 / 线程；
多 Reactor 多进程 / 线程；

其中，「多 Reactor 单进程 / 线程」实现方案相比「单 Reactor 单进程 / 线程」方案，不仅复杂而且也没有性能优势，因此实际中并没有应用。

剩下的 3 个方案都是比较经典的，且都有应用在实际的项目中：

单 Reactor 单进程 / 线程；
单 Reactor 多线程 / 进程；
多 Reactor 多进程 / 线程；

方案具体使用进程还是线程，要看使用的编程语言以及平台有关：

Java 语言一般使用线程，比如 Netty;
C 语言使用进程和线程都可以，例如 Nginx 使用的是进程，Memcache 使用的是线程。

接下来，分别介绍这三个经典的 Reactor 方案。

2.1.单 Reactor 单进程 / 线程

一般来说，C 语言实现的是「单 Reactor 单进程」的方案，因为 C 语编写完的程序，运行后就是一个独立的进程，不需要在进程中再创建线程。

我们来看看「单 Reactor 单进程」的方案示意图：

可以看到进程里有 Reactor、Acceptor、Handler 这三个对象：

Reactor 对象的作用是监听和分发事件；
Acceptor 对象的作用是获取连接；
Handler 对象的作用是处理业务；

对象里的 select、accept、read、send 是系统调用函数，dispatch 和「业务处理」是需要完成的操作，其中 dispatch 是分发事件操作。

接下来，介绍下「单 Reactor 单进程」这个方案：

Reactor 对象通过 select （IO 多路复用接口）监听事件，收到事件后通过 dispatch 进行分发，具体分发给 Acceptor 对象还是 Handler 对象，还要看收到的事件类型；
如果是连接建立的事件，则交由 Acceptor 对象进行处理，Acceptor 对象会通过 accept 方法获取连接，并创建一个 Handler 对象来处理后续的响应事件；
如果不是连接建立事件，则交由当前连接对应的 Handler 对象来进行响应；
Handler 对象通过 read -> 业务处理 -> send 的流程来完成完整的业务流程。
单 Reactor 单进程的方案因为全部工作都在同一个进程内完成，所以实现起来比较简单，不需要考虑进程间通信，也不用担心多进程竞争。

但是，这种方案存在 2 个缺点：

第一个缺点，因为只有一个进程，无法充分利用多核 CPU 的性能；
第二个缺点，Handler 对象在业务处理时，整个进程是无法处理其他连接的事件的，如果业务处理耗时比较长，那么就造成响应的延迟；
所以，单 Reactor 单进程的方案不适用计算机密集型的场景，只适用于业务处理非常快速的场景。

Redis 是由 C 语言实现的，它采用的正是「单 Reactor 单进程」的方案，因为 Redis 业务处理主要是在内存中完成，操作的速度是很快的，性能瓶颈不在 CPU 上，所以 Redis 对于命令的处理是单进程的方案。

2.2.单 Reactor 多线程 / 多进程

如果要克服「单 Reactor 单线程 / 进程」方案的缺点，那么就需要引入多线程 / 多进程，这样就产生了单 Reactor 多线程/ 多进程的方案。

闻其名不如看其图，先来看看「单 Reactor 多线程」方案的示意图如下：

详细说一下这个方案：

Reactor 对象通过 select （IO 多路复用接口）监听事件，收到事件后通过 dispatch 进行分发，具体分发给 Acceptor 对象还是 Handler 对象，还要看收到的事件类型；
如果是连接建立的事件，则交由 Acceptor 对象进行处理，Acceptor 对象会通过 accept 方法获取连接，并创建一个 Handler 对象来处理后续的响应事件；
如果不是连接建立事件，则交由当前连接对应的 Handler 对象来进行响应；

上面的三个步骤和单 Reactor 单线程方案是一样的，接下来的步骤就开始不一样了：

Handler 对象不再负责业务处理，只负责数据的接收和发送，Handler 对象通过 read 读取到数据后，会将数据发给子线程里的 Processor 对象进行业务处理；
子线程里的 Processor 对象就进行业务处理，处理完后，将结果发给主线程中的 Handler 对象，接着由 Handler 通过 send 方法将响应结果发送给 client；

单 Reator 多线程的方案优势在于能够充分利用多核 CPU 的能，那既然引入多线程，那么自然就带来了多线程竞争资源的问题。

例如，子线程完成业务处理后，要把结果传递给主线程的 Reactor 进行发送，这里涉及共享数据的竞争。

要避免多线程由于竞争共享资源而导致数据错乱的问题，就需要在操作共享资源前加上互斥锁，以保证任意时间里只有一个线程在操作共享资源，待该线程操作完释放互斥锁后，其他线程才有机会操作共享数据。

聊完单 Reactor 多线程的方案，接着来看看单 Reactor 多进程的方案。

事实上，单 Reactor 多进程相比单 Reactor 多线程实现起来很麻烦，主要因为要考虑子进程 <-> 父进程的双向通信，并且父进程还得知道子进程要将数据发送给哪个客户端。

而多线程间可以共享数据，虽然要额外考虑并发问题，但是这远比进程间通信的复杂度低得多，因此实际应用中也看不到单 Reactor 多进程的模式。

另外，「单 Reactor」的模式还有个问题，因为一个 Reactor 对象承担所有事件的监听和响应，而且只在主线程中运行，在面对瞬间高并发的场景时，容易成为性能的瓶颈的地方。

2.3.多 Reactor 多进程 / 线程

要解决「单 Reactor」的问题，就是将「单 Reactor」实现成「多 Reactor」，这样就产生了第多 Reactor 多进程 / 线程的方案。

老规矩，闻其名不如看其图。多 Reactor 多进程 / 线程方案的示意图如下（以线程为例）：

方案详细说明如下：

主线程中的 MainReactor 对象通过 select 监控连接建立事件，收到事件后通过 Acceptor 对象中的 accept 获取连接，将新的连接分配给某个子线程；
子线程中的 SubReactor 对象将 MainReactor 对象分配的连接加入 select 继续进行监听，并创建一个 Handler 用于处理连接的响应事件。
如果有新的事件发生时，SubReactor 对象会调用当前连接对应的 Handler 对象来进行响应。
Handler 对象通过 read -> 业务处理 -> send 的流程来完成完整的业务流程。

多 Reactor 多线程的方案虽然看起来复杂的，但是实际实现时比单 Reactor 多线程的方案要简单的多，原因如下：

主线程和子线程分工明确，主线程只负责接收新连接，子线程负责完成后续的业务处理。
主线程和子线程的交互很简单，主线程只需要把新连接传给子线程，子线程无须返回数据，直接就可以在子线程将处理结果发送给客户端。

大名鼎鼎的两个开源软件 Netty 和 Memcache 都采用了「多 Reactor 多线程」的方案。

采用了「多 Reactor 多进程」方案的开源软件是 Nginx，不过方案与标准的多 Reactor 多进程有些差异。

具体差异表现在主进程中仅仅用来初始化 socket，并没有创建 mainReactor 来 accept 连接，而是由子进程的 Reactor 来 accept 连接，通过锁来控制一次只有一个子进程进行 accept（防止出现惊群现象），子进程 accept 新连接后就放到自己的 Reactor 进行处理，不会再分配给其他子进程。

3.实现单 Reactor 单进程版本的TCP服务器

接下来我们就要来实现一个单 Reactor 单进程版本的TCP服务器

首先我们需要下面这些文件

这个Socket.hpp，nocopy.hpp，Epoller.hpp都是我们封装好了的，我们直接使用就行了，接下来我们只需要编写tcpserver.hpp和main.cc

3.1.Connection类

承接上一节中的 epoll 服务器：现在的问题是，来自用户的数据可能会被 TCP 协议拆分成多个报文，那么服务器怎么才能知道什么时候最后一个小报文被接收了呢？要保证完整地读取客户端发送的数据，服务器需要将这次读取到的数据保存起来，对它们进行一定的处理（报文可能会有报头，以解决粘包问题），最后将它们拼接起来，再向上层应用程序交付。

问题是 Recver 中的缓冲区 buffer 是一个局部变量，每次循环都会重置。而服务端可能会有成百上千个来自客户端建立连接后打开的文件描述符，这无法保证为每个文件描述符都保存本轮循环读取的数据。

解决办法是为套接字文件描述符建立独立的接收和发送缓冲区，因为套接字是基于连接的，所以用一个名为 Connection 的类来保存所有和连接相关的属性，例如文件描述符，收发缓冲区，以及对文件描述符的操作（包括读、写和异常操作），所以要设置三个回调函数以供后续在不同的分支调用，最后还要设置一个回指指针，它将会保存服务器对象的地址，到后面会介绍它的用处。


#pragma once
#include<iostream>
#include<string>
#include<functional>
#include<memory>class Connection;
using func_t =std::function<void(std::shared_ptr<Connection>)>;class Connection
{private:
int _sock;
std::string _inbuffer;//这里来当输入缓冲区，但是这里是有缺点的，它不能处理二进制流
std::string _outbuffer;func_t _recv_cb;//读回调函数
func_t _send_cb;//写回调函数
func_t _except_cd;////添加一个回指指针
std::shared_ptr<TcpServer> _tcp_server_ptr;
};
class TcpServer
{};

Connection结构中除了包含文件描述符和其对应的读回调、写回调和异常回调外，还包含一个输入缓冲区_inbuffer、一个输出缓冲区_outbuffer以及一个回指指针_tcp_setver_ptr

当某个文件描述符的读事件就绪时，调用recv函数读取客户端发来的数据，但并不能保证读到了一个完整报文，因此需要将读取到的数据暂时存放到该文件描述符对应的_inBuffer中，当_inBuffer中可以分离出一个完整的报文后再将其分离出来进行数据处理，_inBuffer本质就是用来解决粘包问题的

当处理完一个报文请求后，需将响应数据发送给客户端，但并不能保证底层TCP的发送缓冲区中有足够的空间写入，因此需将要发送的数据暂时存放到该文件描述符对应的_outBuffer中，当底层TCP的发送缓冲区中有空间，即写事件就绪时，再依次发送_outBuffer中的数据

Connection结构中设置回指指针_svrPtr，便于快速找到TcpServer对象，因为后续需要根据Connection结构找到这个TcpServer对象。如上层业务处理函数NetCal函数向_outBuffer输出缓冲区递交数据后，需通过Connection中的回指指针，"提醒"TcpServer进行处理

Connection结构中需提供一个管理回调的成员函数，便于外部对回调进行设置

3.2.TcpServer类

按照之前的经验，我们很快就能写出下面这个

#include <memory> // 包含shared_ptr和unique_ptr等智能指针的头文件  
#include <unordered_map> // 包含unordered_map的头文件  class TcpServer  
{  
public:  // 构造函数，初始化TCP服务器  // 接收一个端口号作为参数，用于创建监听套接字，并初始化事件轮询器  TcpServer(uint16_t port)  : _port(port), // 存储端口号  _listensock_ptr(new Sock()), // 创建一个Sock对象用于监听，并使用shared_ptr管理  _epoller_ptr(new Epoller()) // 创建一个Epoller对象用于事件轮询，并使用shared_ptr管理  {  _listensock_ptr->Socket(); // 调用Sock的Socket方法创建套接字  _listensock_ptr->Bind(_port); // 绑定端口号  _listensock_ptr->Listen(); // 开始监听  // 注意：这里可能还需要将监听套接字添加到事件轮询器中，但代码中没有显示  }  // 初始化方法，可以在这里添加额外的初始化逻辑  // 但从提供的代码来看，这个方法目前是空的  void Init()  {  // 可以在这里添加初始化代码  }  // 启动方法，用于启动服务器的事件循环  // 但从提供的代码来看，这个方法目前是空的  // 通常，这里会包含启动事件轮询器的逻辑  void Start()  {  // 启动事件轮询器，开始处理网络事件  // 注意：实际代码中需要实现这部分逻辑  }  // 析构函数，用于清理资源  // 注意：由于使用了shared_ptr，这里的资源清理将是自动的  // 但如果还有其他需要手动清理的资源（如文件描述符等），则应该在这里处理  ~TcpServer()  {  // 析构函数会自动调用shared_ptr的析构，从而销毁Sock和Epoller对象  // 如果需要，可以在这里添加额外的清理代码  }  private:  uint16_t _port; // 存储服务器监听的端口号  std::shared_ptr<Epoller> _epoller_ptr; // 事件轮询器的智能指针  std::unordered_map<int, std::shared_ptr<Connection>> _connections; // 存储连接的哈希表，键为套接字描述符，值为Connection对象的智能指针  std::shared_ptr<Sock> _listensock_ptr; // 监听套接字的智能指针  
};

这里回答一个问题

当服务器开始运行时，一定会有大量的Connection结构体对象需要被new出来，那么这些结构体对象需不需要被管理呢？

当然是需要的，所以在服务器类里面，定义了一个哈希表_connections，用sock来作为哈希表的键值，sock对应的结构体connection和sock一起作为键值对，也就是哈希桶中存储的值(存储键值对<sock, connection>)，今天是不会出现哈希冲突的，所以每个键值下面的哈希桶只会挂一个键值对，即一个<sock, connection>.

初始化服务器时，第一个需要被添加到哈希表中的sock，一定是listensock，所以在init方法中，先把listensock添加到哈希表里面，添加的同时还要传该listensock所对应的关心事件的方法，对于listensock来说，只需要关注读方法即可，其他两个方法设为nullptr即可。

接下来是使用ET模式的，所以我们要实施下面这三点

相比于LT模式，关于ET（边缘触发）模式的使用，确实需要关注这几个关键点：

设置EPOLLET标志：在将文件描述符（如socket）添加到epoll实例时，你需要通过epoll_event结构体中的events字段设置EPOLLET标志，以启用边缘触发模式。这告诉epoll，你希望在该文件描述符的状态从非就绪变为就绪时只接收一次事件通知。
将socket文件描述符设置为非阻塞：由于ET模式要求你能够在一个事件通知中尽可能多地处理数据，直到没有更多数据可读，因此将socket设置为非阻塞模式是非常重要的。这允许你的read调用在没有数据可读时立即返回EAGAIN或EWOULDBLOCK，而不是阻塞等待。
循环调用read直到返回EAGAIN或EWOULDBLOCK：在接收到一个ET模式的事件通知后，你需要在一个循环中调用read函数，不断尝试从socket中读取数据，直到read返回EAGAIN或EWOULDBLOCK。这表示当前没有更多数据可读，你可以安全地继续等待下一个事件通知。

为了方便使用，我们将设置文件描述符为非阻塞的代码封装到set_non_blocking函数里面去了

TcpServer类

class TcpServer : public nocopy
{static const int num = 64;public:TcpServer(uint16_t port) : _port(port),_listensock_ptr(new Sock()),_epoller_ptr(new Epoller()),_quit(true){_listensock_ptr->Socket();set_non_blocking(_listensock_ptr->Fd()); // 设置listen文件描述符为非阻塞_listensock_ptr->Bind(_port);_listensock_ptr->Listen();}~TcpServer(){}// 设置文件描述符为非阻塞int set_non_blocking(int fd){int flags = fcntl(fd, F_GETFL, 0);if (flags == -1){perror("fcntl: get flags");return -1;}if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -1){perror("fcntl: set non-blocking");return -1;}return 0;}void Init(){}void Start(){_quit = false;// 将listen套接字添加到epoll中->将listensock和他关心的事件，添加到内核的epoll模型中的红黑树里面// 1.将listensock添加到红黑树_epoller_ptr->EpollUpDate(EPOLL_CTL_ADD, _listensock_ptr->Fd(), (EPOLLIN | EPOLLET)); // 注意这里的EPOLLET设置了ET模式// 设置listen文件描述符为非阻塞,这个在初始化已经完成了struct epoll_event revs[num]; // 专门用来处理事件的while (!_quit){int n=_epoller_ptr->EpollerWait(revs,num);for(int i=0;i<num;i++){}}_quit = true;}private:uint16_t _port;std::shared_ptr<Epoller> _epoller_ptr;std::unordered_map<int, std::shared_ptr<Connection>> _connections;std::shared_ptr<Sock> _listensock_ptr;bool _quit;
};

3.3.Connection的真正用处

写到这里，我们发现我们这个代码和之前的不是差不多吗？但事实上，现在才到关键的地方！！

我们在上面提及：

当某个文件描述符的读事件就绪时，调用recv函数读取客户端发来的数据，但并不能保证读到了一个完整报文，因此需要将读取到的数据暂时存放到该文件描述符对应的_inBuffer中，当_inBuffer中可以分离出一个完整的报文后再将其分离出来进行数据处理，_inBuffer本质就是用来解决粘包问题的

当处理完一个报文请求后，需将响应数据发送给客户端，但并不能保证底层TCP的发送缓冲区中有足够的空间写入，因此需将要发送的数据暂时存放到该文件描述符对应的_outBuffer中，当底层TCP的发送缓冲区中有空间，即写事件就绪时，再依次发送_outBuffer中的数据

写到这里，我们需要理解，我们不仅仅需要将listen套接字加入到我们的epoll的红黑树里面，我们还需要将listen套接字建立一个Connection对象，将listen套接字加入到Connection中，同时还要即要将<listen套接字，Connection>放到我们的unordered_map对象_connections里面去。只要我们把Connection对象管理好了，我们的连接就管理好了。

所以，我们现在需要重新编写一下我们的连接类——Connection类

Connection类

class Connection
{
public:Connection(int sock, std::shared_ptr<TcpServer> tcp_server_ptr) : _sock(sock), _tcp_server_ptr(tcp_server_ptr){}~Connection(){}void setcallback(func_t recv_cb, func_t send_cb, func_t except_cb){_recv_cb = recv_cb;_send_cb = send_cb;_except_cd = except_cb;}private:int _sock;std::string _inbuffer; // 这里来当输入缓冲区，但是这里是有缺点的，它不能处理二进制流std::string _outbuffer;func_t _recv_cb;   // 读回调函数func_t _send_cb;   // 写回调函数func_t _except_cd; //// 添加一个回指指针std::shared_ptr<TcpServer> _tcp_server_ptr;};

现在我们就需要来实现我们上面的内容了

void AddConnection(int sock, uint32_t event, func_t recv_cb, func_t send_cb, func_t except_cb,std::string clientip="0.0.0.0",uint16_t clientport=0){// 1.给listen套接字创建一个Connection对象std::shared_ptr<Connection> new_connection = std::make_shared<Connection>(sock, std::shared_ptr<TcpServer>(this)); // 创建Connection对象new_connection->setcallback(recv_cb, send_cb, except_cb);// 2.添加到_connections里面去_connections.insert(std::make_pair(sock, new_connection));// 将listen套接字添加到epoll中->将listensock和他关心的事件，添加到内核的epoll模型中的红黑树里面// 3.将listensock添加到红黑树_epoller_ptr->EpollUpDate(EPOLL_CTL_ADD, sock, event); // 注意这里的EPOLLET设置了ET模式std::cout << "add a new connection success,sockfd:" << sock << std::endl;}

正如上面所说，这个函数完成了三件事情。

1.给listen套接字创建一个Connection对象
2.添加到_connections里面去
3.将listensock添加到红黑树

接下来我们就可以修改我们的Init函数

init函数

 void Init(){_listensock_ptr->Socket();set_non_blocking(_listensock_ptr->Fd()); // 设置listen文件描述符为非阻塞_listensock_ptr->Bind(_port);_listensock_ptr->Listen();AddConnection(_listensock_ptr->Fd(),(EPOLLIN|EPOLLET),nullptr,nullptr,nullptr);//暂时设置成nullptr}

好了，我们现在就到了我们的事件派发过程。

3.4.Dispatcher——事件派发器

我们在这里将Start函数更名为Loop函数，并且创建一个新函数Dispatcher

bool IsConnectionSafe(int fd){auto iter = _connections.find(fd);if (iter == _connections.end()){return false;}else{return true;}}void Dispatcher() // 事件派发器{int n = _epoller_ptr->EpollerWait(revs, num); // 获取已经就绪的事件for (int i = 0; i < num; i++){uint32_t events = revs[i].events;int sock = revs[i].data.fd;// 如果出现异常，统一转发为读写问题，只需要处理读写就行if (events & EPOLLERR) // 出现错误了{events |= (EPOLLIN | EPOLLOUT);}if (events & EPOLLHUP){events |= (EPOLLIN | EPOLLOUT);}// 只需要处理读写就行if (events & EPOLLIN&&IsConnectionSafe(sock)) // 读事件就绪{if(_connections[sock]->_recv_cb)_connections[sock]->_recv_cb(_connections[sock]);}if (events & EPOLLOUT&&IsConnectionSafe(sock)) // 写事件就绪{if(_connections[sock]->_send_cb)_connections[sock]->_send_cb(_connections[sock]);}}}void Loop(){_quit = false;while (!_quit){Dispatcher();}_quit = true;}

事件派发器是真正服务器要开始运行了，服务器会将就绪的每个连接都进行处理，首先如果连接不在哈希表中，那就说明这个连接中的sock还没有被添加到epoll模型中的红黑树，不能直接进行处理，需要先添加到红黑树中，然后让epoll_wait来拿取就绪的连接再告知程序员，这个时候再进行处理，这样才不会等待，而是直接进行数据拷贝。

Loop中处理就绪的事件的方法非常非常的简单，如果该就绪的fd关心的是读事件，那就直接调用该sock所在连接结构体内部的读方法即可，如果是写事件那就调用写方法即可。有人说那如果fd关心异常事件呢？其实异常事件大部分也都是读事件，不过也有写事件，所以处理异常的逻辑我们直接放到读方法和写方法里面即可，当有异常事件到来时，直接去对应的读方法或写方法里面执行对应的逻辑即可。

以下是一些可能触发 EPOLLERR 事件的情况的示例:

1.连接错误:当使用非阻塞套接字进行连接时，如果连接失败，套接字的 epoll 事件集中将包含 EPOLLERR 事件。可以通过检査 events 字段中是否包含 EPOLLERR 来处理连接错误。
2.接收错误:在非阻塞套接字上进行读取操作时，如果发生错误，例如对方关闭了连接或者接收缓冲区溢出，套接字的 epol 事件集中将包含 EPOLLERR 事件。
3.发送错误:在非阻塞套接字上进行写入操作时，如果发生错误，例如对方关闭了连接或者发送缓冲区溢出，套接字的 epol 事件集中将包含 EPOLLERR 事件。
4.文件操作错误:当使用 epol 监听文件描述符时，如果在读取或写入文件时发生错误，文件描述符的 epol 事件集中将包含 EPOLLERR 事件。

需要注意的是，EPOLLERR 事件通常与 EPOLLIN 或 EPOLLOUT 事件一起使用。当发生 EPOLLERR 事件时，通常也会同时发生 EPOLLIN 或EPOLLOUT 事件..

假设某个异常事件发生了，那么这个异常事件会自动被内核设置到epoll_wait返回的事件集中，这个异常事件一定会和一个sock关联，比如客户端和服务器用sock通信着，突然客户端关闭连接，那么服务器的sock上原本关心着读事件，此时内核会自动将异常事件设置到该sock关心的事件集合里，在处理sock关心的读事件时，读方法会捎带处理掉这个异常事件，处理方式为服务器关闭通信的sock，因为客户端已经把连接断开了，服务器没必要维护和这个客户端的连接了，服务器也断开就好，这样的逻辑在读方法里面就可以实现。

3.5.回调函数

我们看上面的事件派发器，最后都是派发给_send_cb和_recv_cb函数，这两个函数我们还没有设置呢！所以我们需要来设置一下

 void Init(){_listensock_ptr->Socket();set_non_blocking(_listensock_ptr->Fd()); // 设置listen文件描述符为非阻塞_listensock_ptr->Bind(_port);_listensock_ptr->Listen();AddConnection(_listensock_ptr->Fd(), (EPOLLIN | EPOLLET),std::bind(&TcpServer::Accepter, this, std::placeholders::_1), nullptr, nullptr); // 暂时设置成nullptr}void Accepter(std::shared_ptr<Connection> conection){while (1){struct sockaddr_in peer;socklen_t len=sizeof(peer);int sock=accept(conection->Getsock(),(sockaddr*)&peer,&len);if(sock>0){set_non_blocking(sock);//设置非阻塞AddConnection (sock,EPOLLIN,nullptr,nullptr,nullptr);}else{if(errno==EWOULDBLOCK) break;else if(errno==EINTR) break;else break;}}}

在代码实现上，给AddConnection传参时，用到了一个C++11的知识，就是bind绑定的使用，一般情况下，如果你将包装器包装的函数指针类型传参给包装器类型时，是没有任何问题的，因为包装器本质就是一个仿函数，内部调用了被包装的对象的方法，所以传参是没有任何问题的。

但如果你要是在类内传参，那就有问题了，会出现类型不匹配的问题，这个问题真的很恶心，而且这个问题一报错就劈里啪啦的报一大堆错，因为function是模板，C++报错最恶心的就是模板报错，一报错人都要炸了。话说回来，为什么是类型不匹配呢？因为在类内调用类内方法时，其实是通过this指针来调用的，如果你直接将Accepter方法传给AddConnection，两者类型是不匹配的，因为Accepter的第一个参数是this指针，正确的做法是利用包装器的适配器bind来进行传参，bind将Accepter进行绑定，前两个参数为绑定的对象类型和给绑定的对象所传的参数，因为Accepter第一个参数是this指针，所以第一个参数就可以固定传this，后面的一个参数不应该是现在传，而应该是调用Accepter方法的时候再传，只有这样才能在类内将类成员函数指针传给包装器类型。
不过吧还有一种不常用的方法，就是利用lambda表达式来进行传参，lambda可以捕捉上下文的this指针，然后再把lambda类型传给包装器类型，这种方式不常用，用起来也怪别扭的，function和bind是适配模式，两者搭配在一起用还是更顺眼一些，lambda这种方式了解一下就好。

为了演示效果，我们写了一个打印函数来展示！

    void Accepter(std::shared_ptr<Connection> conection){while (1){struct sockaddr_in peer;socklen_t len=sizeof(peer);int sock=accept(conection->Getsock(),(sockaddr*)&peer,&len);if(sock>0){//获取客户端信息uint16_t clientport=ntohs(peer.sin_port);char buffer[128];inet_ntop(AF_INET,&(peer.sin_addr),buffer,sizeof(buffer));std::cout<<"get a new client from:"<<buffer<<conection->Getsock()<<std::endl;set_non_blocking(sock);//设置非阻塞AddConnection (sock,EPOLLIN,nullptr,nullptr,nullptr);}else{if(errno==EWOULDBLOCK) break;else if(errno==EINTR) break;else break;}}}void PrintConnection(){std::cout<<"_connections fd list: "<<std::endl;for(auto&connection:_connections){std::cout<<connection.second->Getsock()<<" ";}std::cout<<std::endl;}

tcpserver.hpp

#pragma once
#include <iostream>
#include <string>
#include <functional>
#include <memory>
#include <unordered_map>#include "Socket.hpp"
#include "Epoller.hpp"
#include "nocopy.hpp"class Connection;
using func_t = std::function<void(std::shared_ptr<Connection>)>;
class TcpServer;class Connection
{
public:Connection(int sock, std::shared_ptr<TcpServer> tcp_server_ptr) : _sock(sock), _tcp_server_ptr(tcp_server_ptr){}~Connection(){}void setcallback(func_t recv_cb, func_t send_cb, func_t except_cb){_recv_cb = recv_cb;_send_cb = send_cb;_except_cb = except_cb;}int Getsock(){return _sock;}private:int _sock;std::string _inbuffer; // 这里来当输入缓冲区，但是这里是有缺点的，它不能处理二进制流std::string _outbuffer;public:func_t _recv_cb;   // 读回调函数func_t _send_cb;   // 写回调函数func_t _except_cb; //// 添加一个回指指针std::shared_ptr<TcpServer> _tcp_server_ptr;
};
class TcpServer : public nocopy
{static const int num = 64;public:TcpServer(uint16_t port) : _port(port),_listensock_ptr(new Sock()),_epoller_ptr(new Epoller()),_quit(true){}~TcpServer(){}// 设置文件描述符为非阻塞int set_non_blocking(int fd){int flags = fcntl(fd, F_GETFL, 0);if (flags == -1){perror("fcntl: get flags");return -1;}if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -1){perror("fcntl: set non-blocking");return -1;}return 0;}void Init(){_listensock_ptr->Socket();set_non_blocking(_listensock_ptr->Fd()); // 设置listen文件描述符为非阻塞_listensock_ptr->Bind(_port);_listensock_ptr->Listen();AddConnection(_listensock_ptr->Fd(), (EPOLLIN | EPOLLET),std::bind(&TcpServer::Accepter, this, std::placeholders::_1), nullptr, nullptr); // 暂时设置成nullptr}void Accepter(std::shared_ptr<Connection> conection){while (1){struct sockaddr_in peer;socklen_t len=sizeof(peer);int sock=accept(conection->Getsock(),(sockaddr*)&peer,&len);if(sock>0){//获取客户端信息uint16_t clientport=ntohs(peer.sin_port);char buffer[128];inet_ntop(AF_INET,&(peer.sin_addr),buffer,sizeof(buffer));std::cout<<"get a new client from:"<<buffer<<conection->Getsock()<<std::endl;set_non_blocking(sock);//设置非阻塞AddConnection (sock,EPOLLIN,nullptr,nullptr,nullptr);}else{if(errno==EWOULDBLOCK) break;else if(errno==EINTR) break;else break;}}}void PrintConnection(){std::cout<<"_connections fd list: "<<std::endl;for(auto&connection:_connections){std::cout<<connection.second->Getsock()<<" ";}std::cout<<std::endl;}void AddConnection(int sock, uint32_t event, func_t recv_cb, func_t send_cb, func_t except_cb){// 1.给listen套接字创建一个Connection对象std::shared_ptr<Connection> new_connection = std::make_shared<Connection>(sock, std::shared_ptr<TcpServer>(this)); // 创建Connection对象new_connection->setcallback(recv_cb, send_cb, except_cb);// 2.添加到_connections里面去_connections.insert(std::make_pair(sock, new_connection));// 将listen套接字添加到epoll中->将listensock和他关心的事件，添加到内核的epoll模型中的红黑树里面// 3.将listensock添加到红黑树_epoller_ptr->EpollUpDate(EPOLL_CTL_ADD, sock, event); // 注意这里的EPOLLET设置了ET模式std::cout<<"add a new connection success,sockfd:"<<sock<<std::endl;}bool IsConnectionSafe(int fd){auto iter = _connections.find(fd);if (iter == _connections.end()){return false;}else{return true;}}void Dispatcher() // 事件派发器{int n = _epoller_ptr->EpollerWait(revs, num); // 获取已经就绪的事件for (int i = 0; i < num; i++){uint32_t events = revs[i].events;int sock = revs[i].data.fd;// 如果出现异常，统一转发为读写问题，只需要处理读写就行if (events & EPOLLERR) // 出现错误了{events |= (EPOLLIN | EPOLLOUT);}if (events & EPOLLHUP){events |= (EPOLLIN | EPOLLOUT);}// 只需要处理读写就行if (events & EPOLLIN && IsConnectionSafe(sock)) // 读事件就绪{if (_connections[sock]->_recv_cb)_connections[sock]->_recv_cb(_connections[sock]);}if (events & EPOLLOUT && IsConnectionSafe(sock)) // 写事件就绪{if (_connections[sock]->_send_cb)_connections[sock]->_send_cb(_connections[sock]);}}}void Loop(){_quit = false;while (!_quit){Dispatcher();PrintConnection();}_quit = true;}private:uint16_t _port;std::shared_ptr<Epoller> _epoller_ptr;std::unordered_map<int, std::shared_ptr<Connection>> _connections;std::shared_ptr<Sock> _listensock_ptr;bool _quit;struct epoll_event revs[num]; // 专门用来处理事件的
};

我们看看效果

还可以啊！！！到这里我们就算是打通了我们连接的过程，我们接着看

3.6.区分读写异常事件

我们看看这个Accepter函数

  void Accepter(std::shared_ptr<Connection> conection){while (1){struct sockaddr_in peer;socklen_t len=sizeof(peer);int sock=accept(conection->Getsock(),(sockaddr*)&peer,&len);if(sock>0){//获取客户端信息uint16_t clientport=ntohs(peer.sin_port);char buffer[128];inet_ntop(AF_INET,&(peer.sin_addr),buffer,sizeof(buffer));std::cout<<"get a new client from:"<<buffer<<conection->Getsock()<<std::endl;set_non_blocking(sock);//设置非阻塞AddConnection (sock,EPOLLIN,nullptr,nullptr,nullptr);}else{if(errno==EWOULDBLOCK) break;else if(errno==EINTR) break;else break;}}}

注意里面的AddConnection函数，这个函数是用文件描述符来创建Connection对象的。但是文件描述符有两种啊！一个是listen套接字，一个是普通套接字。Listen套接字只关心读事件，而其他文件描述符则是关心读，写，异常事件的！！但是我们在这里却统一使用了AddConnection (sock,EPOLLIN,nullptr,nullptr,nullptr);一棍子打死，只关心读事件。这样子是非常不合理的。

所以对应AddConnection (sock,EPOLLIN,nullptr,nullptr,nullptr);这里，我们需要修改

//连接管理器void Accepter(std::shared_ptr<Connection> conection){while (1){struct sockaddr_in peer;socklen_t len=sizeof(peer);int sock=accept(conection->Getsock(),(sockaddr*)&peer,&len);if(sock>0){//获取客户端信息uint16_t clientport=ntohs(peer.sin_port);char buffer[128];inet_ntop(AF_INET,&(peer.sin_addr),buffer,sizeof(buffer));std::cout<<"get a new client from:"<<buffer<<conection->Getsock()<<std::endl;set_non_blocking(sock);//设置非阻塞AddConnection (sock,EPOLLIN,std::bind(&TcpServer::Recver, this, std::placeholders::_1),std::bind(&TcpServer::Sender, this, std::placeholders::_1),std::bind(&TcpServer::Excepter, this, std::placeholders::_1);}else{if(errno==EWOULDBLOCK) break;else if(errno==EINTR) break;else break;}}}//事件管理器void Recver(std::shared_ptr<Connection> conection){std::cout<<"haha ,got you"<<conection->Getsock()<<std::endl;}void Sender(std::shared_ptr<Connection> conection){}void Excepter(std::shared_ptr<Connection> conection){}

我们可以测试一下，到底有没有用

我们发现它在一直打印！说明我们成功了！！

3.7.读事件的处理

好了，我们现在就应该来处理读写异常事件

读事件的处理

Recver这里还是和之前一样的问题，也是前面在写三个多路转接接口服务器时，一直没有处理的问题，你怎么保证你一次就把所有数据全部都读上来了呢？

如果不能保证，那就和Accepter一样，必须打死循环来进行读取，当recv返回值大于0，那我们就把读取到的数据先放入缓冲区，缓冲区在哪里呢？

其实就在connection参数所指向的结构体里面，结构体里会有sock所对应的收发缓冲区。然后就调用外部传入的回调函数_service，对服务器收到的数据进行应用层的业务逻辑处理。

当recv读到0时，说明客户端把连接关了，那这就算异常事件，直接回调sock对应的异常处理方法即可。
当recv的返回值小于0，同时错误码被设置为EAGAIN或EWOULDBLOCK时，则说明recv已经把sock底层的数据全部读走了，则此时直接break跳出循环即可。也有可能是被信号给中断了，则此时应该继续执行循环
另外一种情况就是recv系统调用真的出错了，则此时也调用sock的异常方法进行处理即可。

业务逻辑处理方法应该在本次循环读取到所有的数据之后再进行处理。

class Connection
{
......void Append(std::string info)//读取成功了，就把读取的内容存到这里来！{_inbuffer+=info;}private:int _sock;std::string _inbuffer; // 这里来当输入缓冲区，但是这里是有缺点的，它不能处理二进制流std::string _outbuffer;};

我们接着写，

为了方便，我们这里给Connection类加入两个新成员，下面是修改的部分


class Connection
{
public:uint16_t _clientport;std::string _clientip;
};
class TcpServer : public nocopy
{// 连接管理器void Accepter(std::shared_ptr<Connection> conection){while (1){struct sockaddr_in peer;socklen_t len = sizeof(peer);int sock = accept(conection->Getsock(), (sockaddr *)&peer, &len);if (sock > 0){// 获取客户端信息uint16_t clientport = ntohs(peer.sin_port);char buffer[128];inet_ntop(AF_INET, &(peer.sin_addr), buffer, sizeof(buffer));std::cout << "get a new client from:" << buffer << conection->Getsock() << std::endl;set_non_blocking(sock); // 设置非阻塞AddConnection(sock, EPOLLIN,std::bind(&TcpServer::Recver, this, std::placeholders::_1),std::bind(&TcpServer::Sender, this, std::placeholders::_1),std::bind(&TcpServer::Excepter, this, std::placeholders::_1),buffer,clientport);}else{if (errno == EWOULDBLOCK)break;else if (errno == EINTR)break;elsebreak;}}}// 事件管理器void Recver(std::shared_ptr<Connection> conection){int sock = conection->Getsock();while (1){char buffer[128];memset(buffer, 0, sizeof(buffer));ssize_t n = recv(sock, buffer, sizeof(buffer) - 1, 0); // 非阻塞读取if (n > 0)                                             // 成功了！！{conection->Append(buffer); // 把读取的数据放到Connection对象的输入缓冲区里面}else if (n == 0){}else{}}}void AddConnection(int sock, uint32_t event, func_t recv_cb, func_t send_cb, func_t except_cb,std::string clientip="0.0.0.0",uint16_t clientport=0){// 1.给listen套接字创建一个Connection对象std::shared_ptr<Connection> new_connection = std::make_shared<Connection>(sock, std::shared_ptr<TcpServer>(this)); // 创建Connection对象new_connection->setcallback(recv_cb, send_cb, except_cb);new_connection->_clientip=clientip;new_connection->_clientport=clientport;// 2.添加到_connections里面去_connections.insert(std::make_pair(sock, new_connection));// 将listen套接字添加到epoll中->将listensock和他关心的事件，添加到内核的epoll模型中的红黑树里面// 3.将listensock添加到红黑树_epoller_ptr->EpollUpDate(EPOLL_CTL_ADD, sock, event); // 注意这里的EPOLLET设置了ET模式std::cout << "add a new connection success,sockfd:" << sock << std::endl;}};

具体修改看代码即可

我们接着完善我们的读事件的处理

 // 事件管理器void Recver(std::shared_ptr<Connection> conection){int sock = conection->Getsock();while (1){char buffer[128];memset(buffer, 0, sizeof(buffer));ssize_t n = recv(sock, buffer, sizeof(buffer) - 1, 0); // 非阻塞读取if (n > 0)                                             // 成功了！！{conection->Append(buffer); // 把读取的数据放到Connection对象的输入缓冲区里面}else if (n == 0) // 客户端{std::cout << "sockfd:" << sock << ",client:" << conection->_clientip << ":" << conection->_clientport << "quit" << std::endl;Excepter(conection);}else{if (errno == EWOULDBLOCK)break;else if (errno == EINTR)break;else{std::cout << "sockfd:" << sock << ",client:" << conection->_clientip << ":" << conection->_clientport << "recv err" << std::endl;Excepter(conection);}}}}void Sender(std::shared_ptr<Connection> conection){}void Excepter(std::shared_ptr<Connection> conection){}

好，我们把源代码拿出来测试一下

tcpserver.hpp

#pragma once
#include <iostream>
#include <string>
#include <functional>
#include <memory>
#include <unordered_map>#include "Socket.hpp"
#include "Epoller.hpp"
#include "nocopy.hpp"class Connection;
using func_t = std::function<void(std::shared_ptr<Connection>)>;
class TcpServer;class Connection
{
public:Connection(int sock, std::shared_ptr<TcpServer> tcp_server_ptr) : _sock(sock), _tcp_server_ptr(tcp_server_ptr){}~Connection(){}void setcallback(func_t recv_cb, func_t send_cb, func_t except_cb){_recv_cb = recv_cb;_send_cb = send_cb;_except_cb = except_cb;}int Getsock(){return _sock;}void Append(std::string info){_inbuffer += info;}public:int _sock;std::string _inbuffer; // 这里来当输入缓冲区，但是这里是有缺点的，它不能处理二进制流std::string _outbuffer;public:func_t _recv_cb;                            // 读回调函数func_t _send_cb;                            // 写回调函数func_t _except_cb;                          //std::shared_ptr<TcpServer> _tcp_server_ptr; // 添加一个回指指针uint16_t _clientport;std::string _clientip;
};
class TcpServer : public nocopy
{static const int num = 64;public:TcpServer(uint16_t port) : _port(port),_listensock_ptr(new Sock()),_epoller_ptr(new Epoller()),_quit(true){}~TcpServer(){}// 设置文件描述符为非阻塞int set_non_blocking(int fd){int flags = fcntl(fd, F_GETFL, 0);if (flags == -1){perror("fcntl: get flags");return -1;}if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -1){perror("fcntl: set non-blocking");return -1;}return 0;}void Init(){_listensock_ptr->Socket();set_non_blocking(_listensock_ptr->Fd()); // 设置listen文件描述符为非阻塞_listensock_ptr->Bind(_port);_listensock_ptr->Listen();AddConnection(_listensock_ptr->Fd(), (EPOLLIN | EPOLLET),std::bind(&TcpServer::Accepter, this, std::placeholders::_1), nullptr, nullptr); // 暂时设置成nullptr}// 连接管理器void Accepter(std::shared_ptr<Connection> conection){while (1){struct sockaddr_in peer;socklen_t len = sizeof(peer);int sock = accept(conection->Getsock(), (sockaddr *)&peer, &len);if (sock > 0){// 获取客户端信息uint16_t clientport = ntohs(peer.sin_port);char buffer[128];inet_ntop(AF_INET, &(peer.sin_addr), buffer, sizeof(buffer));std::cout << "get a new client from:" << buffer << conection->Getsock() << std::endl;set_non_blocking(sock); // 设置非阻塞AddConnection(sock, EPOLLIN,std::bind(&TcpServer::Recver, this, std::placeholders::_1),std::bind(&TcpServer::Sender, this, std::placeholders::_1),std::bind(&TcpServer::Excepter, this, std::placeholders::_1),buffer, clientport);}else{if (errno == EWOULDBLOCK)break;else if (errno == EINTR)break;elsebreak;}}}// 事件管理器void Recver(std::shared_ptr<Connection> conection){int sock = conection->Getsock();while (1){char buffer[128];memset(buffer, 0, sizeof(buffer));ssize_t n = recv(sock, buffer, sizeof(buffer) - 1, 0); // 非阻塞读取if (n > 0)                                             // 成功了！！{conection->Append(buffer); // 把读取的数据放到Connection对象的输入缓冲区里面}else if (n == 0) // 客户端{std::cout << "sockfd:" << sock << ",client:" << conection->_clientip << ":" << conection->_clientport << "quit" << std::endl;Excepter(conection);}else{if (errno == EWOULDBLOCK)break;else if (errno == EINTR)break;else{std::cout << "sockfd:" << sock << ",client:" << conection->_clientip << ":" << conection->_clientport << "recv err" << std::endl;Excepter(conection);}}}}void Sender(std::shared_ptr<Connection> conection){}void Excepter(std::shared_ptr<Connection> conection){std::cout<<"Execpted ! fd:"<<conection->Getsock()<<std::endl;}void PrintConnection(){std::cout << "_connections fd list: " ;for (auto &connection : _connections){std::cout << connection.second->Getsock() << " ";std::cout<<connection.second->_inbuffer;}std::cout << std::endl;}void AddConnection(int sock, uint32_t event, func_t recv_cb, func_t send_cb, func_t except_cb,std::string clientip = "0.0.0.0", uint16_t clientport = 0){// 1.给listen套接字创建一个Connection对象std::shared_ptr<Connection> new_connection = std::make_shared<Connection>(sock, std::shared_ptr<TcpServer>(this)); // 创建Connection对象new_connection->setcallback(recv_cb, send_cb, except_cb);new_connection->_clientip = clientip;new_connection->_clientport = clientport;// 2.添加到_connections里面去_connections.insert(std::make_pair(sock, new_connection));// 将listen套接字添加到epoll中->将listensock和他关心的事件，添加到内核的epoll模型中的红黑树里面// 3.将listensock添加到红黑树_epoller_ptr->EpollUpDate(EPOLL_CTL_ADD, sock, event); // 注意这里的EPOLLET设置了ET模式std::cout << "add a new connection success,sockfd:" << sock << std::endl;}bool IsConnectionSafe(int fd){auto iter = _connections.find(fd);if (iter == _connections.end()){return false;}else{return true;}}void Dispatcher() // 事件派发器{int n = _epoller_ptr->EpollerWait(revs, num); // 获取已经就绪的事件for (int i = 0; i < num; i++){uint32_t events = revs[i].events;int sock = revs[i].data.fd;// 如果出现异常，统一转发为读写问题，只需要处理读写就行if (events & EPOLLERR) // 出现错误了{events |= (EPOLLIN | EPOLLOUT);}if (events & EPOLLHUP){events |= (EPOLLIN | EPOLLOUT);}// 只需要处理读写就行if (events & EPOLLIN && IsConnectionSafe(sock)) // 读事件就绪{if (_connections[sock]->_recv_cb)_connections[sock]->_recv_cb(_connections[sock]);}if (events & EPOLLOUT && IsConnectionSafe(sock)) // 写事件就绪{if (_connections[sock]->_send_cb)_connections[sock]->_send_cb(_connections[sock]);}}}void Loop(){_quit = false;while (!_quit){Dispatcher();PrintConnection();}_quit = true;}private:uint16_t _port;std::shared_ptr<Epoller> _epoller_ptr;std::unordered_map<int, std::shared_ptr<Connection>> _connections;std::shared_ptr<Sock> _listensock_ptr;bool _quit;struct epoll_event revs[num]; // 专门用来处理事件的
};

好像没有什么问题啊！！！我们确实做到了读取数据！但是我们还是没有处理这些数据！！

所以我们又要弄一个回调函数来处理这些数据

class TcpServer : public nocopy
{public:TcpServer(uint16_t port,func_t OnMessage) : _port(port),_listensock_ptr(new Sock()),_epoller_ptr(new Epoller()),_quit(true),_OnMessage(OnMessage){}
...// 事件管理器void Recver(std::shared_ptr<Connection> conection){int sock = conection->Getsock();while (1){char buffer[128];memset(buffer, 0, sizeof(buffer));ssize_t n = recv(sock, buffer, sizeof(buffer) - 1, 0); // 非阻塞读取if (n > 0)                                             // 成功了！！{conection->Append(buffer); // 把读取的数据放到Connection对象的输入缓冲区里面}else if (n == 0) // 客户端{std::cout << "sockfd:" << sock << ",client:" << conection->_clientip << ":" << conection->_clientport << "quit" << std::endl;Excepter(conection);}else{if (errno == EWOULDBLOCK)break;else if (errno == EINTR)break;else{std::cout << "sockfd:" << sock << ",client:" << conection->_clientip << ":" << conection->_clientport << "recv err" << std::endl;Excepter(conection);}}_OnMessage(conection);//将读取的数据交给上层处理}}private:....//上层处理数据func_t _OnMessage;//将数据交给上层
};

我们要求上层来完成检测，来处理粘包的问题！这个_OnMessage不就是回调函数吗！！！

我们简单的写一个

mian.cc

#include"tcpserver.hpp"
#include<memory>void DefaultOmMessage(std::shared_ptr<Connection> connection_ptr)
{std::cout<<"上层得到了数据："<<connection_ptr->_inbuffer<<std::endl;}int main()
{std::unique_ptr<TcpServer> svr(new TcpServer(8877,DefaultOmMessage));svr->Init();svr->Loop();
}

我们来测试一下行不行

3.8.处理数据 ——（反）序列化/编解码

可以哦！！！我们已经拿到数据了，那我们应该怎么处理数据呢？我们应该在这里完成序列和反序列化！！！由于我之前写过序列和反序列化，以及编码解码的代码，所以我直接拿过复制粘贴好了。

Serialization.hpp

#pragma
#define CRLF "\t"               // 分隔符
#define CRLF_LEN strlen(CRLF)   // 分隔符长度
#define SPACE " "               // 空格
#define SPACE_LEN strlen(SPACE) // 空格长度
#define OPS "+-*/%"             // 运算符#include <iostream>
#include <string>
#include <cstring>
#include<assert.h>//参数len为in的长度，是一个输出型参数。如果为0代表err
std::string decode(std::string& in,size_t*len)
{assert(len);//如果长度为0是错误的// 1.确认in的序列化字符串完整（分隔符）*len=0;size_t pos = in.find(CRLF);//查找\t第一次出现时的下标//查找不到，errif(pos == std::string::npos){return "";//返回空串}// 2.有分隔符，判断长度是否达标// 此时pos下标正好就是标识大小的字符长度std::string inLenStr = in.substr(0,pos);//从下标0开始一直截取到第一个\t之前//到这里我们要明白，我们这上面截取的是最开头的长度，也就是说，我们截取到的一定是个数字，这个是我们序列化字符的长度size_t inLen = atoi(inLenStr.c_str());//把截取的这个字符串转int，inLen就是序列化字符的长度//传入的字符串的长度 - 第一个\t前面的字符数 - 2个\tsize_t left = in.size() - inLenStr.size()- 2*CRLF_LEN;//原本预计的序列化字符串长度if(left<inLen){//真实的序列化字符串长度和预计的字符串长度进行比较return ""; //剩下的长度（序列化字符串的长度）没有达到标明的长度}// 3.走到此处，字符串完整，开始提取序列化字符串std::string ret = in.substr(pos+CRLF_LEN,inLen);//从pos+CRLF_LEN下标开始读取inLen个长度的字符串——即序列化字符串*len = inLen;// 4.因为in中可能还有其他的报文（下一条）// 所以需要把当前的报文从in中删除，方便下次decode，避免二次读取size_t rmLen = inLenStr.size() + ret.size() + 2*CRLF_LEN;//长度+2个\t+序列字符串的长度in.erase(0,rmLen);//移除从索引0开始长度为rmLen的字符串// 5.返回return ret;
}//编码不需要修改源字符串，所以const。参数len为in的长度
std::string encode(const std::string& in,size_t len)
{std::string ret = std::to_string(len);//将长度转为字符串添加在最前面，作为标识ret+=CRLF;ret+=in;ret+=CRLF;return ret;
}class Request//客户端使用的
{
public:// 将用户的输入转成内部成员// 用户可能输入x+y，x+ y，x +y,x + y等等格式// 提前修改用户输入（主要还是去掉空格），提取出成员Request(){}// 删除输入中的空格void rmSpace(std::string &in){std::string tmp;for (auto e : in){if (e != ' '){tmp += e;}}in = tmp;}// 序列化 （入参应该是空的，会返回一个序列化字符串）void serialize(std::string &out)//这个是客户端在发送消息给服务端时使用的，在这之后要先编码，才能发送出去{// x + yout.clear(); // 序列化的入参是空的out += std::to_string(_x);out += SPACE;out += _ops; // 操作符不能用tostring，会被转成asciiout += SPACE;out += std::to_string(_y);// 不用添加分隔符（这是encode要干的事情）}//序列化之后应该要编码，去加个长度// 反序列化（解开bool deserialize(const std::string &in)//这个是服务端接收到客户端发来的消息后使用的，在这之前要先解码{// x + y 需要取出x，y和操作符size_t space1 = in.find(SPACE);  // 第一个空格if (space1 == std::string::npos) // 没找到{return false;}size_t space2 = in.rfind(SPACE); // 第二个空格if (space2 == std::string::npos) // 没找到{return false;}// 两个空格都存在，开始取数据std::string dataX = in.substr(0, space1);std::string dataY = in.substr(space2 + SPACE_LEN); // 默认取到结尾std::string op = in.substr(space1 + SPACE_LEN, space2 - (space1 + SPACE_LEN));if (op.size() != 1){return false; // 操作符长度有问题}// 没问题了，转内部成员_x = atoi(dataX.c_str());_y = atoi(dataY.c_str());_ops = op[0];return true;}public:int _x;int _y;char _ops;
};class Response // 服务端必须回应
{public:Response(int code = 0, int result = 0): _exitCode(code), _result(result){}// 序列化void serialize(std::string &out)//这个是服务端发送消息给客户端使用的，使用之后要编码{// code retout.clear();out += std::to_string(_exitCode);out += SPACE;out += std::to_string(_result);out += CRLF;}// 反序列化bool deserialize(const std::string &in)//这个是客户端接收服务端消息后使用的，使用之前要先解码{// 只有一个空格size_t space = in.find(SPACE);  // 寻找第一个空格的下标if (space == std::string::npos) // 没找到{return false;}std::string dataCode = in.substr(0, space);std::string dataRes = in.substr(space + SPACE_LEN);_exitCode = atoi(dataCode.c_str());_result = atoi(dataRes.c_str());return true;}public:int _exitCode; // 计算服务的退出码int _result;   // 结果
};
Response Caculater(const Request& req)
{Response resp;//构造函数中已经指定了exitcode为0switch (req._ops){case '+':resp._result = req._x + req._y;break;case '-':resp._result = req._x - req._y;break;case '*':resp._result = req._x * req._y;break;case '%':{if(req._y == 0){resp._exitCode = -1;//取模错误break;}resp._result = req._x % req._y;//取模是可以操作负数的break;}case '/':{if(req._y == 0){resp._exitCode = -2;//除0错误break;}resp._result = req._x / req._y;//取模是可以操作负数的break;}default:resp._exitCode = -3;//操作符非法break;}return resp;
}

接下来我们就可以对读取的数据进行处理了！！！怎么处理呢？我们还是跟自定义协议那篇一样，搞一个计算器好了！

Caculater函数

Response Caculater(const Request& req)
{Response resp;//构造函数中已经指定了exitcode为0switch (req._ops){case '+':resp._result = req._x + req._y;break;case '-':resp._result = req._x - req._y;break;case '*':resp._result = req._x * req._y;break;case '%':{if(req._y == 0){resp._exitCode = -1;//取模错误break;}resp._result = req._x % req._y;//取模是可以操作负数的break;}case '/':{if(req._y == 0){resp._exitCode = -2;//除0错误break;}resp._result = req._x / req._y;//取模是可以操作负数的break;}default:resp._exitCode = -3;//操作符非法break;}return resp;
}

我们把这个放到了我们的 Serialization.hpp

接下来就接着修改我们的main.cc好了！

main.cc

#include "tcpserver.hpp"
#include <memory>
#include "Serialization.hpp"void DefaultOmMessage(std::shared_ptr<Connection> connection_ptr)
{std::cout << "上层得到了数据：" << connection_ptr->_inbuffer << std::endl;std::string inbuf = connection_ptr->_inbuffer;size_t packageLen = inbuf.size();//由于我们是使用telnet来测试的所以，我们就不解码了/*// 3.1.解码和反序列化客户端传来的消息std::string package = decode(inbuf, &packageLen); // 解码if (packageLen == 0){printf("decode err: %s[%d] status: %d", connection_ptr->_clientip, connection_ptr->_clientport, packageLen);// 报文不完整或有误}*/Request req;bool deStatus = req.deserialize(inbuf); // 使用Request的反序列化，packsge内部各个成员已经有了数值if (deStatus)                             // 获取消息反序列化成功{// 3.2.获取结构化的相应Response resp = Caculater(req); // 将计算任务的结果存放到Response里面去// 3.3.序列化和编码响应std::string echoStr;resp.serialize(echoStr); // 序列化//由于我们使用的是telnet来测试的，所以我们不编码了// echoStr = encode(echoStr, echoStr.size()); // 编码// 3.4.写入，发送返回值给输出缓冲区connection_ptr->_outbuffer=echoStr;std::cout<<connection_ptr->_outbuffer<<std::endl;}else // 客户端消息反序列化失败{printf("deserialize err: %s[%d] status: %d", connection_ptr->_clientip, connection_ptr->_clientport, deStatus);return;}}int main()
{std::unique_ptr<TcpServer> svr(new TcpServer(8877, DefaultOmMessage));svr->Init();svr->Loop();
}

我们测试一下

可以啊！！！接下来就是专门处理写事件

3.9.写事件的处理

之前写服务器时，我们从来没处理过写事件，写事件和读事件不太一样，关心读事件是要常设置的，但写事件一般都是就绪的，因为内核发送缓冲区大概率都是有空间的，如果每次都要让epoll帮我们关心读事件，这其实是一种资源的浪费，因为大部分情况下，你send数据，都是会直接将应用层数据拷贝到内核缓冲区的，不会出现等待的情况，而recv就不太一样，recv在读取的时候，有可能数据还在网络里面，所以recv要等待的概率是比较高的，所以对于读事件来说，常常都要将其设置到sock所关心的事件集合中。

但写事件并不是这样的，写事件应该是偶尔设置到关心集合中，比如你这次没把数据一次性发完，但你又没设置该sock关心写事件，当下次写事件就绪了，也就是内核发送缓冲区有空间了，epoll_wait也不会通知你，那你还怎么发送剩余数据啊，所以这个时候你就应该设置写事件关心了，让epoll_wait帮你监视sock上的写事件，以便于下次epoll_wait通知你时，你还能够继续发送上次没发完的数据。

这个时候可能有人会问，ET模式不是只会通知一次吗？如果我这次设置了写关心，但下次发送数据的时候，还是没发送完毕（因为内核发送缓冲区可能没有剩余空间了），那后面ET模式是不是就不会通知我了呀，那我还怎么继续发送剩余的数据呢？ET模式在底层就绪的事件状态发生变化时，还会再通知上层一次的，对于读事件来说，当数据从无到有，从有到多状态发生变化时，ET就还会通知上层一次，对于写事件来说，当内核发送缓冲区剩余空间从无到有，从有到多状态发生变化时，ET也还会通知上层一次，所以不用担心数据发送不完的问题产生，因为ET是会通知我们的。

在循环外，我们只需要通过判断outbuffer是否为空的情况，来决定是否要设置写事件关心，当数据发送完了那我们就取消对于写事件的关心，不占用epoll的资源，如果数据没发送完，那就设置对于写事件的关心，因为我们要保证下次写事件就绪时，epoll_wait能够通知我们对写事件进行处理。

main.cc

#include "tcpserver.hpp"
#include <memory>
#include "Serialization.hpp"void DefaultOmMessage(std::shared_ptr<Connection> connection_ptr)
{std::cout << "上层得到了数据：" << connection_ptr->_inbuffer << std::endl;std::string inbuf = connection_ptr->_inbuffer;size_t packageLen = inbuf.size();//由于我们是使用telnet来测试的所以，我们就不解码了/*// 3.1.解码和反序列化客户端传来的消息std::string package = decode(inbuf, &packageLen); // 解码if (packageLen == 0){printf("decode err: %s[%d] status: %d", connection_ptr->_clientip, connection_ptr->_clientport, packageLen);// 报文不完整或有误}*/Request req;bool deStatus = req.deserialize(inbuf); // 使用Request的反序列化，packsge内部各个成员已经有了数值if (deStatus)                             // 获取消息反序列化成功{// 3.2.获取结构化的相应Response resp = Caculater(req); // 将计算任务的结果存放到Response里面去// 3.3.序列化和编码响应std::string echoStr;resp.serialize(echoStr); // 序列化//由于我们使用的是telnet来测试的，所以我们不编码了// echoStr = encode(echoStr, echoStr.size()); // 编码// 3.4.写入，发送返回值给输出缓冲区connection_ptr->_outbuffer=echoStr;std::cout<<connection_ptr->_outbuffer<<std::endl;//发送connection_ptr->_tcp_server_ptr->Sender(connection_ptr);//调用里面的方法来发送}else // 客户端消息反序列化失败{printf("deserialize err: %s[%d] status: %d", connection_ptr->_clientip, connection_ptr->_clientport, deStatus);return;}}

我们注意最后一句，我们直接把发生认为交给了Tcpserver类里的Sender函数，不过这个函数还没有写，我们来写一下！

Tcpserver类里的Sender函数

void Sender(std::shared_ptr<Connection> connection) // 使用shared_ptr管理Connection对象的生命周期  
{  // 无限循环，直到输出缓冲区为空或发生需要退出的错误  while (true)  {  // 引用当前连接的输出缓冲区  auto &outbuffer = connection->_outbuffer;  // 尝试发送数据，send函数返回发送的字节数，或者-1表示错误  ssize_t n = send(connection->Getsock(), outbuffer.data(), outbuffer.size(), 0);  // 如果n大于0，表示部分或全部数据发送成功  if (n > 0)  {  // 从输出缓冲区中移除已发送的数据  outbuffer.erase(0, n);  // 如果输出缓冲区为空，则退出循环  if (outbuffer.empty())  break;  }  // 如果n等于0，表示连接已关闭（对端执行了关闭操作），退出函数  else if (n == 0)  {  return;  }  // 处理发送失败的情况  else  {  // 根据errno的值判断错误类型  if (errno == EWOULDBLOCK)  {  // EWOULDBLOCK表示非阻塞模式下资源暂时不可用，可稍后重试，但这里选择直接退出循环  break;  }  else if (errno == EINTR)  {  // EINTR表示操作被信号中断，可以安全地重新尝试  continue;  }  else  {  // 打印错误信息，包含套接字描述符和客户端IP地址及端口  std::cout << "sockfd:" << connection->Getsock() << ",client:" << connection->_clientip << ":" << connection->_clientport << "send error" << std::endl;  // 调用异常处理回调函数  Excepter(conection);  // 退出循环  break;  }  }  }  
}

但是这还不够啊，万一我没写完数据呢？我们还需要进行写处理，这里我们就借助epoll机制来帮我们。

// Sender函数负责通过给定的Connection对象发送数据。  
// 它使用epoll机制来管理套接字的读写事件，根据发送情况调整对写事件的关注。  
void Sender(std::shared_ptr<Connection> connection)  
{  // 引用Connection对象的输出缓冲区  auto &outbuffer = connection->_outbuffer;  // 循环发送数据，直到输出缓冲区为空或发生错误  while (true)  {  // 尝试发送数据  ssize_t n = send(connection->Getsock(), outbuffer.data(), outbuffer.size(), 0); // 注意：使用.data()获取缓冲区首地址  // 如果发送成功（n > 0）  if (n > 0)  {  // 从输出缓冲区中移除已发送的数据  outbuffer.erase(0, n);  // 如果输出缓冲区为空，则退出循环  if (outbuffer.empty())  break;  }  // 如果n == 0，表示连接已正常关闭（对端调用了close），退出函数  else if (n == 0)  {  return;  }  // 处理发送失败的情况  else  {  // 检查errno以确定错误类型  if (errno == EWOULDBLOCK)  {  // EWOULDBLOCK表示资源暂时不可用，通常发生在非阻塞模式下，退出循环  break;  }  else if (errno == EINTR)  {  // EINTR表示操作被信号中断，继续循环尝试发送  continue;  }  else  {  // 打印错误信息，并调用Connection对象的异常处理回调函数  std::cout << "sockfd:" << connection->Getsock() << ",client:" << connection->_clientip << ":" << connection->_clientport << "send error" << std::endl;  Excepter(conection);  break;  }  }  }  // 根据输出缓冲区是否为空，调整对写事件的关注  if (!outbuffer.empty())  {  // 如果输出缓冲区不为空，开启对写事件的关注  EnableEvent(connection->Getsock(), true, true); // 同时关心读和写事件  }  else  {  // 如果输出缓冲区为空，关闭对写事件的关注，但保持对读事件的关注  EnableEvent(connection->Getsock(), true, false); // 只关心读事件  }  
}  // EnableEvent函数用于根据给定的参数，更新epoll事件监听器对套接字的事件关注。  
void EnableEvent(int sock, bool readable, bool sendable)  
{  // 初始化events变量，用于存储要设置的事件标志  uint32_t events = 0;  // 如果需要关注读事件，则设置EPOLLIN标志  events |= (readable ? EPOLLIN : 0);  // 如果需要关注写事件，则设置EPOLLOUT标志  events |= (sendable ? EPOLLOUT : 0);  // 总是设置EPOLLET标志，表示使用边缘触发模式（Edge Triggered）  events |= EPOLLET;  // 调用epoll更新函数，修改对指定套接字的事件关注  // 注意：_epoller_ptr应该是一个指向epoll事件监听器对象的指针，EPOLL_CTL_MOD表示修改现有事件  _epoller_ptr->EpollUpDate(EPOLL_CTL_MOD, sock, events);  
}

我们测试一下，这能不能完成写的任务啊！

很好，显然是成功了！！！

3.10.异常事件的处理

大家仔细看看上面的代码就会知道，读写事件出问题了之后，都是立马将错误和异常都传递给了Exceptrt函数！

下面是异常事件的处理方法，我们统一对所有异常事件，都先将其从epoll模型中移除，然后关闭文件描述符，最后将conn从哈希表_connecions中移除。

void Excepter(std::shared_ptr<Connection> conection){std::cout << "Execpted ! fd:" << conection->Getsock() << std::endl;//1.将文件描述符从epoll模型里面移除来_epoller_ptr->EpollUpDate(EPOLL_CTL_DEL,conection->Getsock(),0);//2.关闭异常的文件描述符close(conection->Getsock());//3.从unordered_map中删除_connections.erase(conection->Getsock());}

我们来测试一下：

我们连接我们的服务器，然后退出，服务器就会打印下面这些内容

很显然失败了！

为什么呢？就是connection指针的生命周期有问题

值得注意的是，connnection指针指向的连接结构体空间，必须由我们自己释放，有人说，为什么啊？你哈希表不是都已经erase了么？为什么还要程序员自己再delete连接结构体空间呢？

这里要给大家说明一点的是，所有的容器在erase的时候，都只释放容器自己所new出来的空间，像哈希表这样的容器，它会new一个存储键值对的节点空间，节点里面存储着conn指针和sockfd，当调用哈希表的erase时，哈希表只会释放它自己new出来的节点空间，至于这个节点空间里面存储了一个Connection类型的指针，并且这个指针变量指向一个结构体空间，这些事情哈希表才不会管你呢，容器只会释放他自己开辟的空间，哈希表是vector挂单链表的方式来实现的。
所以我们要自己手动释放conn指向的空间，如果你不想自己手动释放conn指向的堆空间资源，则可以存储智能指针对象，这样在哈希表erase时，就会释放智能指针对象的空间，从而自动调用Connection类的析构函数。

这样搞起来其实还是很麻烦的，所以我们就自己手动释放就好了，如果不手动释放那就会造成内存泄露。

由于实现起来比较麻烦，需要修改代码的很多地方，我就不改了，但是最核心的部分就像下面这样子

至此，我们的简易版本的单Reactor单进程版本的TCP服务器就算完成了！

3.11.源代码

tcpserver.hpp

#pragma once
#include <iostream>
#include <string>
#include <functional>
#include <memory>
#include <unordered_map>#include "Socket.hpp"
#include "Epoller.hpp"
#include "nocopy.hpp"
#include <sys/socket.h>
#include <sys/types.h>class Connection;
using func_t = std::function<void(std::shared_ptr<Connection>)>;
class TcpServer;class Connection
{
public:Connection(int sock, std::shared_ptr<TcpServer> tcp_server_ptr) : _sock(sock), _tcp_server_ptr(tcp_server_ptr){}~Connection(){}void setcallback(func_t recv_cb, func_t send_cb, func_t except_cb){_recv_cb = recv_cb;_send_cb = send_cb;_except_cb = except_cb;}int Getsock(){return _sock;}void Append(std::string info){_inbuffer += info;}public:int _sock;std::string _inbuffer; // 这里来当输入缓冲区，但是这里是有缺点的，它不能处理二进制流std::string _outbuffer;public:func_t _recv_cb;                            // 读回调函数func_t _send_cb;                            // 写回调函数func_t _except_cb;                          //std::shared_ptr<TcpServer> _tcp_server_ptr; // 添加一个回指指针uint16_t _clientport;std::string _clientip;
};
class TcpServer : public nocopy
{static const int num = 64;public:TcpServer(uint16_t port, func_t OnMessage) : _port(port),_listensock_ptr(new Sock()),_epoller_ptr(new Epoller()),_quit(true),_OnMessage(OnMessage){}~TcpServer(){}// 设置文件描述符为非阻塞int set_non_blocking(int fd){int flags = fcntl(fd, F_GETFL, 0);if (flags == -1){perror("fcntl: get flags");return -1;}if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -1){perror("fcntl: set non-blocking");return -1;}return 0;}void Init(){_listensock_ptr->Socket();set_non_blocking(_listensock_ptr->Fd()); // 设置listen文件描述符为非阻塞_listensock_ptr->Bind(_port);_listensock_ptr->Listen();AddConnection(_listensock_ptr->Fd(), (EPOLLIN | EPOLLET),std::bind(&TcpServer::Accepter, this, std::placeholders::_1), nullptr, nullptr); // 暂时设置成nullptr}// 连接管理器void Accepter(std::shared_ptr<Connection> conection){while (1){struct sockaddr_in peer;socklen_t len = sizeof(peer);int sock = accept(conection->Getsock(), (sockaddr *)&peer, &len);if (sock > 0){// 获取客户端信息uint16_t clientport = ntohs(peer.sin_port);char buffer[128];inet_ntop(AF_INET, &(peer.sin_addr), buffer, sizeof(buffer));std::cout << "get a new client from:" << buffer << conection->Getsock() << std::endl;set_non_blocking(sock); // 设置非阻塞AddConnection(sock, EPOLLIN,std::bind(&TcpServer::Recver, this, std::placeholders::_1),std::bind(&TcpServer::Sender, this, std::placeholders::_1),std::bind(&TcpServer::Excepter, this, std::placeholders::_1),buffer, clientport);}else{if (errno == EWOULDBLOCK)break;else if (errno == EINTR)break;elsebreak;}}}// 事件管理器void Recver(std::shared_ptr<Connection> conection){int sock = conection->Getsock();while (1){char buffer[128];memset(buffer, 0, sizeof(buffer));ssize_t n = recv(sock, buffer, sizeof(buffer) - 1, 0); // 非阻塞读取if (n > 0)                                             // 成功了！！{conection->Append(buffer); // 把读取的数据放到Connection对象的输入缓冲区里面}else if (n == 0) // 客户端{std::cout << "sockfd:" << sock << ",client:" << conection->_clientip << ":" << conection->_clientport << "quit" << std::endl;Excepter(conection);}else{if (errno == EWOULDBLOCK)break;else if (errno == EINTR)break;else{std::cout << "sockfd:" << sock << ",client:" << conection->_clientip << ":" << conection->_clientport << "recv err" << std::endl;Excepter(conection);}}_OnMessage(conection); // 将读取的数据交给上层处理}}void Sender(std::shared_ptr<Connection> conection){std::string outbuffer = conection->_outbuffer;while (1){ssize_t n = send(conection->Getsock(), outbuffer.data(), outbuffer.size(), 0);if (n > 0) // 发成功了{outbuffer.erase(0, n);if (outbuffer.empty())break;}else if (n == 0){return;}else{if (errno == EWOULDBLOCK)break;else if (errno == EINTR)continue;else{std::cout << "sockfd:" << conection->Getsock() << ",client:" << conection->_clientip << ":" << conection->_clientport << "send error" << std::endl;Excepter(conection);break;}}}if (!outbuffer.empty()){// 开启对写事件的关心EnableEvent(conection->Getsock(), true, true); // 关心读写}else{// 关闭对写事件的关心EnableEvent(conection->Getsock(), true, false); // 关心读,不关心写}}void EnableEvent(int sock, bool readable, bool sendable){uint32_t events = 0;events |= (readable ? EPOLLIN : 0) | (sendable ? EPOLLOUT : 0) | EPOLLET;_epoller_ptr->EpollUpDate(EPOLL_CTL_MOD, sock, events);}void Excepter(std::shared_ptr<Connection> conection){if (!conection){// Connection 对象可能已被销毁，无需进一步操作std::cout << "Connection already destroyed, no further action taken." << std::endl;return;}std::cout << "Execpted ! fd:" << conection->Getsock() << std::endl;// 1.将文件描述符从epoll模型里面移除来_epoller_ptr->EpollUpDate(EPOLL_CTL_DEL, conection->Getsock(), 0);// 2.关闭异常的文件描述符close(conection->Getsock());// 3.从unordered_map中删除_connections.erase(conection->Getsock());}void PrintConnection(){std::cout << "_connections fd list: ";for (auto &connection : _connections){std::cout << connection.second->Getsock() << " ";std::cout << connection.second->_inbuffer;}std::cout << std::endl;}void AddConnection(int sock, uint32_t event, func_t recv_cb, func_t send_cb, func_t except_cb,std::string clientip = "0.0.0.0", uint16_t clientport = 0){// 1.给listen套接字创建一个Connection对象std::shared_ptr<Connection> new_connection = std::make_shared<Connection>(sock, std::shared_ptr<TcpServer>(this)); // 创建Connection对象new_connection->setcallback(recv_cb, send_cb, except_cb);new_connection->_clientip = clientip;new_connection->_clientport = clientport;// 2.添加到_connections里面去_connections.insert(std::make_pair(sock, new_connection));// 将listen套接字添加到epoll中->将listensock和他关心的事件，添加到内核的epoll模型中的红黑树里面// 3.将listensock添加到红黑树_epoller_ptr->EpollUpDate(EPOLL_CTL_ADD, sock, event); // 注意这里的EPOLLET设置了ET模式std::cout << "add a new connection success,sockfd:" << sock << std::endl;}bool IsConnectionSafe(int fd){auto iter = _connections.find(fd);if (iter == _connections.end()){return false;}else{return true;}}void Dispatcher() // 事件派发器{int n = _epoller_ptr->EpollerWait(revs, num); // 获取已经就绪的事件for (int i = 0; i < num; i++){uint32_t events = revs[i].events;int sock = revs[i].data.fd;// 如果出现异常，统一转发为读写问题，只需要处理读写就行if (events & EPOLLERR) // 出现错误了{events |= (EPOLLIN | EPOLLOUT);}if (events & EPOLLHUP){events |= (EPOLLIN | EPOLLOUT);}// 只需要处理读写就行if (events & EPOLLIN && IsConnectionSafe(sock)) // 读事件就绪{if (_connections[sock]->_recv_cb)_connections[sock]->_recv_cb(_connections[sock]);}if (events & EPOLLOUT && IsConnectionSafe(sock)) // 写事件就绪{if (_connections[sock]->_send_cb)_connections[sock]->_send_cb(_connections[sock]);}}}void Loop(){_quit = false;while (!_quit){Dispatcher();PrintConnection();}_quit = true;}private:uint16_t _port;std::shared_ptr<Epoller> _epoller_ptr;std::unordered_map<int, std::shared_ptr<Connection>> _connections;std::shared_ptr<Sock> _listensock_ptr;bool _quit;struct epoll_event revs[num]; // 专门用来处理事件的// 上层处理数据func_t _OnMessage; // 将数据交给上层
};

Epoller.hpp

#pragma once#include <iostream>
#include <sys/epoll.h>
#include <unistd.h>
#include <cerrno>
#include "nocopy.hpp"class Epoller : public nocopy
{static const int size = 128;public:Epoller(){_epfd = epoll_create(size);if (_epfd == -1){perror("epoll_creat error");}else{printf("epoll_creat successful:%d\n", _epfd);}}~Epoller(){if (_epfd > 0){close(_epfd);}}int EpollerWait(struct epoll_event revents[],int num){int n=epoll_wait(_epfd,revents,num,3000);return n;}int EpollUpDate(int oper,int sock,uint16_t event){int n;if(oper==EPOLL_CTL_DEL)//将该事件从epoll红黑树里面删除{n=epoll_ctl(_epfd,oper,sock,nullptr);if(n!=0){printf("delete epoll_ctl error");}}else{//添加和修改,即EPOLL_CTL_MOD和EPOLL_CTL_ADDstruct epoll_event ev;ev.events=event;ev.data.fd=sock;//方便我们知道是哪个fd就绪了n=epoll_ctl(_epfd,oper,sock,&ev);if(n!=0){perror("add epoll_ctl error");}}return n;}private:int _epfd;
};

nocopy.hpp

#pragma once  class nocopy  
{  
public:  // 允许使用默认构造函数（由编译器自动生成）  nocopy() = default;   // 禁用拷贝构造函数，防止通过拷贝来创建类的实例  nocopy(const nocopy&) = delete;   // 禁用赋值运算符，防止类的实例之间通过赋值操作进行内容复制  nocopy& operator=(const nocopy&) = delete;   
};

Socket.hpp

#pragma once#include <iostream>
#include <string>
#include <unistd.h>
#include <cstring>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <unistd.h>
#include <fcntl.h>// 定义一些错误代码
enum
{SocketErr = 2, // 套接字创建错误BindErr,       // 绑定错误ListenErr,     // 监听错误
};// 监听队列的长度
const int backlog = 10;class Sock // 服务器专门使用
{
public:Sock() : sockfd_(-1) // 初始化时，将sockfd_设为-1，表示未初始化的套接字{}~Sock(){// 析构函数中可以关闭套接字，但这里选择不在析构函数中关闭，因为有时需要手动管理资源}// 创建套接字void Socket(){sockfd_ = socket(AF_INET, SOCK_STREAM, 0);if (sockfd_ < 0){printf("socket error, %s: %d", strerror(errno), errno); // 错误exit(SocketErr);                                        // 发生错误时退出程序}int opt = 1;setsockopt(sockfd_, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)); // 服务器主动关闭后快速重启}// 将套接字绑定到指定的端口上void Bind(uint16_t port){// 让服务器绑定IP地址与端口号struct sockaddr_in local;memset(&local, 0, sizeof(local));   // 清零local.sin_family = AF_INET;         // 网络local.sin_port = htons(port);       // 我设置为默认绑定任意可用IP地址local.sin_addr.s_addr = INADDR_ANY; // 监听所有可用的网络接口if (bind(sockfd_, (struct sockaddr *)&local, sizeof(local)) < 0) // 让自己绑定别人{printf("bind error, %s: %d", strerror(errno), errno);exit(BindErr);}}// 监听端口上的连接请求void Listen(){if (listen(sockfd_, backlog) < 0){printf("listen error, %s: %d", strerror(errno), errno);exit(ListenErr);}}// 接受一个连接请求int Accept(std::string *clientip, uint16_t *clientport){struct sockaddr_in peer;socklen_t len = sizeof(peer);int newfd = accept(sockfd_, (struct sockaddr *)&peer, &len);if (newfd < 0){printf("accept error, %s: %d", strerror(errno), errno);return -1;}char ipstr[64];inet_ntop(AF_INET, &peer.sin_addr, ipstr, sizeof(ipstr));*clientip = ipstr;*clientport = ntohs(peer.sin_port);return newfd; // 返回新的套接字文件描述符}// 连接到指定的IP和端口——客户端才会用的bool Connect(const std::string &ip, const uint16_t &port){struct sockaddr_in peer; // 服务器的信息memset(&peer, 0, sizeof(peer));peer.sin_family = AF_INET;peer.sin_port = htons(port);inet_pton(AF_INET, ip.c_str(), &(peer.sin_addr));int n = connect(sockfd_, (struct sockaddr *)&peer, sizeof(peer));if (n == -1){std::cerr << "connect to " << ip << ":" << port << " error" << std::endl;return false;}return true;}// 关闭套接字void Close(){close(sockfd_);}// 获取套接字的文件描述符int Fd(){return sockfd_;}private:int sockfd_; // 套接字文件描述符
};

Serialization.hpp

#pragma
#define CRLF "\t"               // 分隔符
#define CRLF_LEN strlen(CRLF)   // 分隔符长度
#define SPACE " "               // 空格
#define SPACE_LEN strlen(SPACE) // 空格长度
#define OPS "+-*/%"             // 运算符#include <iostream>
#include <string>
#include <cstring>
#include<assert.h>//参数len为in的长度，是一个输出型参数。如果为0代表err
std::string decode(std::string& in,size_t*len)
{assert(len);//如果长度为0是错误的// 1.确认in的序列化字符串完整（分隔符）*len=0;size_t pos = in.find(CRLF);//查找\t第一次出现时的下标//查找不到，errif(pos == std::string::npos){return "";//返回空串}// 2.有分隔符，判断长度是否达标// 此时pos下标正好就是标识大小的字符长度std::string inLenStr = in.substr(0,pos);//从下标0开始一直截取到第一个\t之前//到这里我们要明白，我们这上面截取的是最开头的长度，也就是说，我们截取到的一定是个数字，这个是我们序列化字符的长度size_t inLen = atoi(inLenStr.c_str());//把截取的这个字符串转int，inLen就是序列化字符的长度//传入的字符串的长度 - 第一个\t前面的字符数 - 2个\tsize_t left = in.size() - inLenStr.size()- 2*CRLF_LEN;//原本预计的序列化字符串长度if(left<inLen){//真实的序列化字符串长度和预计的字符串长度进行比较return ""; //剩下的长度（序列化字符串的长度）没有达到标明的长度}// 3.走到此处，字符串完整，开始提取序列化字符串std::string ret = in.substr(pos+CRLF_LEN,inLen);//从pos+CRLF_LEN下标开始读取inLen个长度的字符串——即序列化字符串*len = inLen;// 4.因为in中可能还有其他的报文（下一条）// 所以需要把当前的报文从in中删除，方便下次decode，避免二次读取size_t rmLen = inLenStr.size() + ret.size() + 2*CRLF_LEN;//长度+2个\t+序列字符串的长度in.erase(0,rmLen);//移除从索引0开始长度为rmLen的字符串// 5.返回return ret;
}//编码不需要修改源字符串，所以const。参数len为in的长度
std::string encode(const std::string& in,size_t len)
{std::string ret = std::to_string(len);//将长度转为字符串添加在最前面，作为标识ret+=CRLF;ret+=in;ret+=CRLF;return ret;
}class Request//客户端使用的
{
public:// 将用户的输入转成内部成员// 用户可能输入x+y，x+ y，x +y,x + y等等格式// 提前修改用户输入（主要还是去掉空格），提取出成员Request(){}// 删除输入中的空格void rmSpace(std::string &in){std::string tmp;for (auto e : in){if (e != ' '){tmp += e;}}in = tmp;}// 序列化 （入参应该是空的，会返回一个序列化字符串）void serialize(std::string &out)//这个是客户端在发送消息给服务端时使用的，在这之后要先编码，才能发送出去{// x + yout.clear(); // 序列化的入参是空的out += std::to_string(_x);out += SPACE;out += _ops; // 操作符不能用tostring，会被转成asciiout += SPACE;out += std::to_string(_y);// 不用添加分隔符（这是encode要干的事情）}//序列化之后应该要编码，去加个长度// 反序列化（解开bool deserialize(const std::string &in)//这个是服务端接收到客户端发来的消息后使用的，在这之前要先解码{// x + y 需要取出x，y和操作符size_t space1 = in.find(SPACE);  // 第一个空格if (space1 == std::string::npos) // 没找到{return false;}size_t space2 = in.rfind(SPACE); // 第二个空格if (space2 == std::string::npos) // 没找到{return false;}// 两个空格都存在，开始取数据std::string dataX = in.substr(0, space1);std::string dataY = in.substr(space2 + SPACE_LEN); // 默认取到结尾std::string op = in.substr(space1 + SPACE_LEN, space2 - (space1 + SPACE_LEN));if (op.size() != 1){return false; // 操作符长度有问题}// 没问题了，转内部成员_x = atoi(dataX.c_str());_y = atoi(dataY.c_str());_ops = op[0];return true;}public:int _x;int _y;char _ops;
};class Response // 服务端必须回应
{public:Response(int code = 0, int result = 0): _exitCode(code), _result(result){}// 序列化void serialize(std::string &out)//这个是服务端发送消息给客户端使用的，使用之后要编码{// code retout.clear();out += std::to_string(_exitCode);out += SPACE;out += std::to_string(_result);out += CRLF;}// 反序列化bool deserialize(const std::string &in)//这个是客户端接收服务端消息后使用的，使用之前要先解码{// 只有一个空格size_t space = in.find(SPACE);  // 寻找第一个空格的下标if (space == std::string::npos) // 没找到{return false;}std::string dataCode = in.substr(0, space);std::string dataRes = in.substr(space + SPACE_LEN);_exitCode = atoi(dataCode.c_str());_result = atoi(dataRes.c_str());return true;}public:int _exitCode; // 计算服务的退出码int _result;   // 结果
};
Response Caculater(const Request& req)
{Response resp;//构造函数中已经指定了exitcode为0switch (req._ops){case '+':resp._result = req._x + req._y;break;case '-':resp._result = req._x - req._y;break;case '*':resp._result = req._x * req._y;break;case '%':{if(req._y == 0){resp._exitCode = -1;//取模错误break;}resp._result = req._x % req._y;//取模是可以操作负数的break;}case '/':{if(req._y == 0){resp._exitCode = -2;//除0错误break;}resp._result = req._x / req._y;//取模是可以操作负数的break;}default:resp._exitCode = -3;//操作符非法break;}return resp;
}

main.cc

#include "tcpserver.hpp"
#include <memory>
#include "Serialization.hpp"void DefaultOmMessage(std::shared_ptr<Connection> connection_ptr)
{std::cout << "上层得到了数据：" << connection_ptr->_inbuffer << std::endl;std::string inbuf = connection_ptr->_inbuffer;size_t packageLen = inbuf.size();//由于我们是使用telnet来测试的所以，我们就不解码了/*// 3.1.解码和反序列化客户端传来的消息std::string package = decode(inbuf, &packageLen); // 解码if (packageLen == 0){printf("decode err: %s[%d] status: %d", connection_ptr->_clientip, connection_ptr->_clientport, packageLen);// 报文不完整或有误}*/Request req;bool deStatus = req.deserialize(inbuf); // 使用Request的反序列化，packsge内部各个成员已经有了数值if (deStatus)                             // 获取消息反序列化成功{// 3.2.获取结构化的相应Response resp = Caculater(req); // 将计算任务的结果存放到Response里面去// 3.3.序列化和编码响应std::string echoStr;resp.serialize(echoStr); // 序列化//由于我们使用的是telnet来测试的，所以我们不编码了// echoStr = encode(echoStr, echoStr.size()); // 编码// 3.4.写入，发送返回值给输出缓冲区connection_ptr->_outbuffer=echoStr;std::cout<<connection_ptr->_outbuffer<<std::endl;connection_ptr->_tcp_server_ptr->Sender(connection_ptr);//调用里面的方法来发送}else // 客户端消息反序列化失败{printf("deserialize err: %s[%d] status: %d", connection_ptr->_clientip, connection_ptr->_clientport, deStatus);return;}}int main()
{std::unique_ptr<TcpServer> svr(new TcpServer(8877, DefaultOmMessage));svr->Init();svr->Loop();
}

3.12.总结

Reactor主要围绕事件派发和自动反应展开的，就比如连接请求到来，epoll_wait提醒程序员就绪的事件到来，应该尽快处理，则与就绪事件相关联的sock会对应着一个connection结构体，这个结构体我觉得就是反应堆模式的精华所在，无论是什么样就绪的事件，每个sock都会有对应的回调方法，所以处理就绪的事件很容易，直接回调connection内的对应方法即可，是读事件就调用读方法，是写事件就调用写方法，是异常事件，则在读方法或写方法中处理IO的同时，顺便处理掉异常事件。

所以我感觉Reactor就像一个化学反应堆，你向这个反应堆里面扔一些连接请求，或者网络数据，则这个反应堆会自动匹配相对应的处理机制来处理到来的事件，很方便，同时由于ET模式和EPOLL，这就让Reactor在处理高并发连接时，展现出了不俗的实力。