【Linux网络编程】IO多种转接之Reactor

Reactor

在这里插入图片描述

点赞👍👍收藏🌟🌟关注💖💖
你的支持是对我最大的鼓励，我们一起努力吧!😃😃

基于上一篇epoll的学习，现在我们也知道epoll的工作模式有两种，一种默认LT工作模式，另一种是ET模式。关于epoll的LT工作模式我们已经写过了。接下来我们写一份基于ET模式下的Reator，处理所有的IO。

Reactor = 如何正确的处理IO+协议定制+业务逻辑处理

下面我们写一个简洁版的Reactor，它是一个半同步半异步IO，具体它什么原理，怎么做的，有什么特征。我们在代码层面上解开它的面纱。代码写完总结就理解了。其实Reactor是在Liunx网络中，最常用，最频繁的一种网络IO设计模式！

我们是这样打算的，对错误码，日志函数，套接字，epoll做封装然后在写服务器的时候用到的时候调用即可。错误码，日志函数，套接字以前我们封装过今天直接用就行了。

错误码封装

#pragma once

enum
{
    USAGG_ERR = 1,
    SOCKET_ERR,
    BIND_ERR,
    LISTEN_ERR,
    EPOLL_CREATE_ERR
};

日志函数封装

#pragma once

#include<iostream>
#include<string>
#include<stdio.h>
#include <cstdarg>
#include<ctime>
#include<sys/types.h>
#include<unistd.h>
#include<fstream>

#define DEBUG  0
#define NORMAL  1
#define WARNING 2
#define ERROR   3
#define FATAL   4

#define LOG_NORMAL "log.txt"
#define LOG_ERR "log.error"

const char* level_to_string(int level)
{
    switch(level)
    {
        case DEBUG: return "DEBUG";
        case NORMAL: return "NORMAL";
        case WARNING: return "WARNING";
        case ERROR: return "ERROR";
        case FATAL: return "FATAL";
    }
}

//时间戳变成时间
char* timeChange()
{
    time_t now=time(nullptr);
    struct tm* local_time;
    local_time=localtime(&now);

    static char time_str[1024];

    snprintf(time_str,sizeof time_str,"%d-%d-%d %d-%d-%d",local_time->tm_year + 1900,\
                    local_time->tm_mon + 1, local_time->tm_mday,local_time->tm_hour, \
                    local_time->tm_min, local_time->tm_sec);

    return time_str;
}



void logMessage(int level,const char* format,...)
{
    //[日志等级] [时间戳/时间] [pid] [message]
    //[WARNING] [2024-3-21 10-46-03] [123] [创建sock失败]
#define NUM 1024
    //获取时间
    char* nowtime=timeChange();
    char logprefix[NUM];
    snprintf(logprefix,sizeof logprefix,"[%s][%s][pid: %d]",level_to_string(level),nowtime,getpid());

    //
    char logconten[NUM];
    va_list arg;
    va_start(arg,format);
    vsnprintf(logconten,sizeof logconten,format,arg);

    
    std::cout<<logprefix<<logconten<<std::endl;  
};

今天这里我们把套接字封装改一下，以前把它相关接口都写成静态的了，不需要对象直接调用了。今天呢，都改成非静态的，未来这个类提供的方法都要以对象调用访问。

#pragma once

#include <iostream>
#include <string>
#include <cstring>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include "Err.hpp"
#include "Log.hpp"

using namespace std;

class Sock
{
    const static int backlog = 32;
    const static int defaultsock = -1;

public:
    Sock(int sock = defaultsock) : _listensock(sock)
    {}

    ~Sock()
    {
        if (_listensock != defaultsock)
            close(_listensock);
    }

public:
    int sock()
    {
        // 1. 创建socket文件套接字对象
        _listensock= socket(AF_INET, SOCK_STREAM, 0);
        if (_listensock < 0)
        {
            logMessage(FATAL, "create socket error");
            exit(SOCKET_ERR);
        }
        logMessage(NORMAL, "create socket success: %d", _listensock);


        int opt = 1;
        setsockopt(_listensock, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt));
    }

	//方便外面获取_listensock
    int Fd()
    {
        return _listensock;
    }

    void Bind(int port)
    {
        // 2. bind绑定自己的网络信息
        struct sockaddr_in local;
        memset(&local, 0, sizeof(local));
        local.sin_family = AF_INET;
        local.sin_port = htons(port);
        local.sin_addr.s_addr = INADDR_ANY;
        if (bind(_listensock, (struct sockaddr *)&local, sizeof(local)) < 0)
        {
            logMessage(FATAL, "bind socket error");
            exit(BIND_ERR);
        }
        logMessage(NORMAL, "bind socket success");
    }

    void Listen()
    {
        // 3. 设置socket 为监听状态
        if (listen(_listensock, backlog) < 0)
        {
            logMessage(FATAL, "listen socket error");
            exit(LISTEN_ERR);
        }
        logMessage(NORMAL, "listen socket success");
    }

    int Accept(string *clientip, uint16_t *clientport,int* err)
    {
        struct sockaddr_in peer;
        socklen_t len = sizeof(peer);
        int sock = accept(_listensock, (struct sockaddr *)&peer, &len);
        *err=errno;
        if (sock < 0){}
            //logMessage(ERROR, "accept error, next");
        else
        {
            //logMessage(NORMAL, "accept a new link success, get new sock: %d", sock); // ?
            *clientip = inet_ntoa(peer.sin_addr);
            *clientport = ntohs(peer.sin_port);
        }
        return sock;
    }

private:
    int _listensock;
};

对于一个epoll来说，首先要先给epoll创建出来，然后epoll的接口都要用到epoll创建成功的返回值。所以直接在这个epoll类中定义个成员变量直接就把创建epoll成功的返回值拿到。然后服务器不需要直接拿着这个返回值，而直接调用这个类对象使用里面的提供的接口就行了。epoll这里先写一个大体框架，后面需要什么了再加

#pragma once

#include <iostream>
#include <sys/epoll.h>
#include <unistd.h>
#include <sys/types.h>
#include "Err.hpp"
#include "Log.hpp"

using namespace std;

const int defaultepfd = -1;
const int size = 128;

class Epoller
{

public:
    Epoller(int epfd = defaultepfd) : _epfd(epfd)
    {
    }

    ~Epoller()
    {
        if (_epfd != defaultepfd)
            close(_epfd);
    }

public:
   
private:
    int _epfd;//创建epoll返回值
};

目前调用逻辑，后面在加内容

#include "TcpServer.hpp"
#include "Err.hpp"
#include <memory>

static void usage(std::string proc)
{
    std::cerr << "Usage:\n\t" << proc << " port" << "\n\n";
}

int main(int argc, char *argv[])
{
    if (argc != 2)
    {
        usage(argv[0]);
        exit(USAGG_ERR);
    }

    uint16_t port=atoi(argv[1]);
    std::unique_ptr<TcpServer> uls(new TcpServer(port));
    uls->initServer();
    uls->Dispatch();

    return 0;
}

服务器大体框架

#pragma once

#include <iostream>
#include "Sock.hpp"
#include "Err.hpp"
#include "Log.hpp"
#include "Epoller.hpp"

const int defaultport = 8080;

class TcpServer
{

public:
    TcpServer(int port = defaultport) : _port(port)
    {
    }

    ~TcpServer()
    {
    }

    void initServer()
    {  
    }

    void Dispatch()//事件派发
    {
    }
public:


private:
    uint16_t _port;
    Sock _sock;
    Epoller _epoll;
};

接下来就是编写服务器了

初始化服务器

void initServer()
{  
    // 1.创建套接字
    _sock.sock();
    _sock.Bind(_port);
    _sock.Listen();

    // 2.创建epoll模型
}

创建套接字之后，我们要在创建一个epoll模型，因此我们在Epoller写一个创建epoll模型的接口，然后服务器直接调用就行了

```cpp
void Create()
{
    _epfd = epoll_create(size);
    if (_epfd < 0)
    {
        logMessage(FATAL, "epoll_create error, code: %d, errstring: %s", errno, strerror(errno));
        exit(EPOLL_CREATE_ERR);
    }
    logMessage(NORMAL, "epoll create success, epfd: %d", _epfd);
}

void initServer()
{  
    // 1.创建套接字
    _sock.sock();
    _sock.Bind(_port);
    _sock.Listen();

    // 2.创建epoll模型
    _epoll.Create();
    
	// 3.将目前唯一的一个sock，添加到epoll中
}

创建好套接字，epoll模型接下来我们先把_listensock套接字添加到epoll里，看看这样写有没有什么问题

不过先在Epoller类中补充一个用户告诉内核你要帮我关心对应fd的什么事件。

// user -> kernel
bool AddEvents(int sock, uint32_t event)
{
    struct epoll_event ev;
    ev.events = event;
    ev.data.fd = sock;
    int n = epoll_ctl(_epfd, EPOLL_CTL_ADD, sock, &ev);
    if (n < 0)
    {
        logMessage(ERROR, "sock join epoll fail");
        return false;
    }
    return true;
}

写好后我们直接调用就好了

void initServer()
{  
    // 1.创建套接字
    _sock.sock();
    _sock.Bind(_port);
    _sock.Listen();

    // 2.创建epoll模型
    _epoll.Create();

    // 3.将目前唯一的一个sock，添加到epoll中
    _epoll.AddEvents(_sock.Fd(),EPOLLIN | EPOLLET);
}

但是现在有小一个问题，前面说过的，如果未来你的套接字工作模式是ET模式，那么该套接字必须处于非阻塞，_listensock套接字创建处理默认就是阻塞的，因此我们需要将文件描述符设为非阻塞！

我们在高级IO的非阻塞IO哪里就已经写过一个将一个fd变成非阻塞了，现在拿过来用就行了，不过我们也还是把它封装起来。然后调用。

#include <iostream>
#include <unistd.h>
#include <fcntl.h>

using namespace std;

class Util
{
public:
    static bool SetNonBlock(int fd)
    {
        int fl = fcntl(fd, F_GETFL);
        if (fl < 0)
            return false;
        fcntl(fd, F_SETFL, fl | O_NONBLOCK);
        return true;
    }
};

到目前为止好像貌似没问题了，但是真的吗？

void initServer()
{  
    // 1.创建套接字
    _sock.sock();
    _sock.Bind(_port);
    _sock.Listen();

    // 2.创建epoll模型
    _epoll.Create();

    // 3. 将目前唯一的一个sock，添加到epoller中， 之前需要先将对应的fd设置成为非阻塞
    Util::SetNonBlock(_sock.Fd());
    _epoll.AddEvents(_sock.Fd(),EPOLLIN | EPOLLET);
}

我们回过头看看之前写过的epoll服务器的代码，这里处理普通sock就绪事件的问题，你怎么保证你把本轮收到的数据都读完？即使把本轮数据都读完了，就一定能够读到一个完整的请求吗？即使未来在这里循环读取，可以反正我们非阻塞也都写过。那就能保证读到一个完整请求吗？
不一定！

那没读到完整的请求我们是不是只能把本轮读到的数据暂时保存到buffer里，可是暂时保存到buffer里，你保存了，那别人怎么办？ 你这个sock和其他sock说你们别着急先别覆盖我的数据，buffer里面保存的是我的数据，你们先不要写。你想多了！

你读完之后整个这个代码区间就全部释放掉了，因为这是栈上面的空间。所以即便你没读完，或者你把这轮读完了。那么下轮在读，buffer早就释放了。你可能本轮读到后半部分但前半部分已经没有了，可能不到下次，下一个循环进来就给你清空了。

在这里插入图片描述
所以怎么样进行正确读写，光有一个套接字和定义一个栈上的缓存区远远不够！

所以我们需要给每一个文件描述符，都要把输入，输出用户层缓存区都要给它带上！

现在我们清楚知道历史代码的问题，我们要正确写整个服务，现在已经不够了。我们需要将每一个套接字进行封装，每一个套接字都要包含自己对应的输入，输出缓存区空间，只有每一个套接字都有自己对应的缓存空间，读取的时候把数据读取赞存在自己对应的缓存区中，没读完下次在读，这时候在添加到我自己的缓存区里，不就行了吗，这个时候谁和谁都不揉在一起。所以我们在封装一个类！ 也就是说我认为未来每一个套接字都看类对象。

class Connection
{
public:
    Connection(int sock = defaultport) : sock_(sock)
    {
    }

    ~Connection()
    {
        if (sock_ == defaultsock)
            close(sock_);
    }

public:
    int sock_;//这个类对应的套接字是谁
    string inbuffer_;//输入缓存区,这里我们暂时没有考虑处理图片,视频等二进制信息,不然stirng就不太合适了                                                                                                                     
    string outbuffer_;//输出缓存区 ,你并不能保证你的写事件就绪
    
}；

针对于上面未来每一个Connection对象，因为每一个套接字未来面对的都有自己的读方法、写方法、异常方法，所以在Connection类中，对一个套接字来讲要提供三个回调方法，分别对应读方法、写方法，异常方法。

const int defaultport = 8080;
const int defaultsock = -1;

using func_t = function<void(Connection *)>;

class Connection
{
public:
    Connection(int sock = defaultport) : sock_(sock)
    {
    }

    ~Connection()
    {
        if (sock_ == defaultsock)
            close(sock_);
    }

public:
    int sock_;  //这个类对应的套接字是谁
    string inbuffer_;  //输入缓存区,这里我们暂时没有考虑处理图片,视频等二进制信息,不然stirng就不太合适了                                                                                                                     
    string outbuffer_;  //输出缓存区 ,你并不能保证你的写事件就绪

    func_t recver_;  //从sock_读
    func_t sender_;  //向sock_写
    func_t excepter_;  //处理sock_,IO的时候上面的异常事件
};

所以在初始服务器写的，创建套接字，创建epoll模型，然后将_listensock套接字添加到epoll中，那未来是不是更多的套接字会被accept上来，我们把每一个套接字都看成Connection对象，所以未来整个服务器是不是存在非常多的Conncetion对象。同时_listensock 也是一个sock啊，也要看作成为一个 Connection对象，虽然_listensock不用所谓的输入输出缓存区，包括回调方法最多也是用一个 recver _，但是在我看来_listensock也是一个Connection。所以刚才服初始服务器哪里写_listensock就有不合适了。并且未来每一个sock都是Connection对象，所以服务器要不要把这些多的Connection对象管理起来呢？ 当然要了！那服务器怎么管理？

先描述，在组织！ ，我们管理起来了吗，管理起来了，Connection就是。怎么组织呢？既然每一个Connection对象都有一个对应的套接字，那我们就是哈希表管理。

class TcpServer
{

public:
	//...

private:
    uint16_t _port;

    Sock _sock;
    Epoller _epoll;
    unordered_map<int,Connection*> _connections;//所有链接集合
};

而且这里的代码仅仅是简单的将套接字添加到epoll中，未来我们不仅想将套接字添加到epoll中，并且还想将每一个Connectoin对象添加unordered_map里。在这里插入图片描述

为什么unordered_map我们使用的是套接字做key，因为将来在epoll中一旦有fd就绪了，我们知道是什么事件就绪，也知道是那个fd就绪。知道之后我们就可以在unordered_map根据文件描述符快速的找到对应套接字的Connection对象，然后输入，输出缓存区也就有了。

在这里插入图片描述

所以在写一个添加链接的函数，作为一个服务器收到一个套接字，它就需要添加一个链接。

void AddConnection(int sock, uint32_t events)
{
    // 1. 首先要为该sock创建Connection，并初始化，并添加到connections_


    // 2. 其次将sock与它要关心的事件"写透式"注册到epoll中，让epoll帮我们关心
    bool r = _epoll.AddEvents(sock, events);
    assert(r);
    (void)r;
    
}

剩下的等会再说，所以当你有了新的套接字的时候，不应该是把套接字直接添加到epoll里，而应该是进行AddConnection。_listensock也是如此

void initServer()
{
    // 1.创建套接字
    _sock.sock();
    _sock.Bind(_port);
    _sock.Listen();

    // 2.创建epoll模型
    _epoll.Create();

    // 3. 将目前唯一的一个sock，添加到epoller中， 之前需要先将对应的fd设置成为非阻塞
    // Util::SetNonBlock(_sock.Fd());
    // _epoll.AddEvents(_sock.Fd(), EPOLLIN | EPOLLET);
    AddConnection(_sock.Fd(),EPOLLIN|EPOLLET);
}

void AddConnection(int sock, uint32_t events)
{
   		// 1.设置非阻塞,ET模式fd要非阻塞
        if (events & EPOLLET)
            Util::SetNonBlock(sock);

        // 2. 该sock创建Connection，并初始化，并添加到connections_
        Connection *conn = new Connection(sock);

        // 3.

        // 4. 其次将sock与它要关心的事件"写透式"注册到epoll中，让epoll帮我们关心
        bool r = _epoll.AddEvents(sock, events);
        assert(r);
        (void)r;

        // 5. 将kv添加到connections_
        _connections.insert(make_pair(sock, conn));

        logMessage(DEBUG, "add new sock : %d in epoll and unordered_map", sock);
}

这样写还没完，未来一旦有fd就绪了，然后就可以在unordered_map中根据fd找到对应的Connection，找到之后我们是不是要执行对应的读，写，异常方法，所以任何一个Connection内方法能够被调用，因此Connection类在提供一个给每一个Connection对象注册读，写，异常方法，

这样写的意思是，未来这个AddConnection接口不仅可以被用来注册_listensock套接字，也可以用来被注册一般文件描述符，一般文件描述符也可能关心读，也可以关心写，也可能关心异常。或者只关心读，只关心写等等情况，所以当我们每个套接字注册一个Connection对象，你怎么知道这个套接字未来要执行什么方法呢，所以我们需要Connection提供一个注册方法。
在这里插入图片描述

class Connection
{
public:
    Connection(int sock = defaultport) : sock_(sock)
    {
    }

    ~Connection()
    {
        if (sock_ == defaultsock)
            close(sock_);
    }

    void Rigster(func_t r, func_t s, func_t e)
    {
        recver_ = r;
        sender_ = s;
        excepter_ = e;
    }

public:
    int sock_;         // 这个类对应的套接字是谁
    string inbuffer_;  // 输入缓存区,这里我们暂时没有考虑处理图片,视频等二进制信息,不然stirng就不太合适了
    string outbuffer_; // 输出缓存区 ,你并不能保证你的写事件就绪

    func_t recver_;   // 从sock_读
    func_t sender_;   // 向sock_写
    func_t excepter_; // 处理sock_,IO的时候上面的异常事件
};

所以当使用AddConnection的时候，注册文件描述符到epoll的时候，既要有要关心的fd，还要有关心fd上的什么事件，这两个字段是你要告诉epoll，当fd就绪时你想怎么处理这个fd，所以还要再给三个参数，告诉这个fd就绪时处理什么方法。

在这里插入图片描述

void AddConnection(int sock, uint32_t events,func_t recver, func_t sender, func_t excepter)
{
    // 1.设置非阻塞,ET模式fd要非阻塞
    if (events & EPOLLET)
        Util::SetNonBlock(sock);

    // 2. 该sock创建Connection，并初始化，并添加到connections_
    Connection *conn = new Connection(sock);

    // 3. 给对应的sock设置对应的回调方法
    conn->Rigster(recver, sender, excepter);

    // 4. 其次将sock与它要关心的事件"写透式"注册到epoll中，让epoll帮我们关心
    bool r = _epoll.AddEvents(sock, events);
    assert(r);
    (void)r;

    // 5. 将kv添加到connections_
    _connections.insert(make_pair(sock, conn));

    logMessage(DEBUG, "add new sock : %d in epoll and unordered_map", sock);
}

所以当我们重新审视初始化服务器的代码，创建套接字，创建epoll模型，把_listensock套接字添加到epoll里，对于_listensock套接字我们也要设置对应的方法，它就绪时想怎么读，怎么写，怎么处理异常。不过对于_listensock套接字它只关心读，所以这里我们先给它一个未来要写的从_listensock获取链接的方法，不关心写和处理异常，直接设置为nullptr就行了。

不过这里还有一个细节，我们因为读、写、异常是我们使用的是包装器function，如何让它的对象调用一个类内的非静态成员函数会比较麻烦，因为类内的非静态成员函数内隐藏了一个this指针。这样的话function的参数还要添加一个，然后调用的话还需要再传一个这个类对象过去。但是我们在C++学过bing绑定，可以固定参数，这样以后调用这个成员函数直接就帮我自动传这个参数了，对这里不懂得可以看这里【C++】C++11中

void Accepter(Connection* conn)
{
}

void initServer()
{
    // 1.创建套接字
    _sock.sock();
    _sock.Bind(_port);
    _sock.Listen();

    // 2.创建epoll模型
    _epoll.Create();

    // 3. 将目前唯一的一个sock，添加到epoller中， 之前需要先将对应的fd设置成为非阻塞
    // Util::SetNonBlock(_sock.Fd());
    // _epoll.AddEvents(_sock.Fd(), EPOLLIN | EPOLLET);
    AddConnection(_sock.Fd(), EPOLLIN | EPOLLET,
                  bind(&TcpServer::Accepter, this, placeholders::_1), nullptr, nullptr);
}

未来一旦设置到epoll里fd上有事件就绪了，然后就根据就绪的fd在unordered_map找到对应的Conncetion，然后调用Conncetion曾经注册的对应的方法来进行事件派发，所以接下来我们要写事件派发的接口，让epoll_wait帮我们获取就绪事件，这里有很多写法，不过我们还是不写在一块，多写一个Loop函数然后进行一次事件派发，Dispatch就是循环派发

// 事件派发器
void Dispatch()
{
    int timeout = -1000;
    while (true)
    {
        Loop(timeout);
    }
}


void Loop(int &timeout)
{
   _epoll.Wait()// 获取已经就绪的事件
}

所以在Epoller再来一个接口，我们直接调用就行了，这里我们需要一个拷贝数组，数组大小，和等待的方式

// kernel -> user
int Wait(struct epoll_event *ets, int num, int timeout)
{
    int n = epoll_wait(_epfd, ets, num, timeout);
    switch (n)
    {
    case 0:
        logMessage(NORMAL, "timeout ...");
        break;
    case -1:
        logMessage(WARNING, "epoll_wait failed, code: %d, errstring: %s", errno, strerror(errno));
        break;
    default:
        logMessage(NORMAL, "have event ready");
        break;
    }
    return n;
}

epoll_wait这里注定了我们一次可能会捞取许多就绪事件，所以我们要在服务器中在定义一个保存所有就绪事件的数组，和数组大小然后调用_epoll.Wait()的时候传过去

const int defaultnum = 128;

class TcpServer
{

public:
    TcpServer(int port = defaultport,int num = defaultnum) : _port(port),_revents(nullptr),_num(num)
    {
    }

	

private:
    int _port;

    Sock _sock;
    Epoller _epoll;
    unordered_map<int, Connection *> _con;
    struct epoll_event *_revents;
    int _num;
};

初始服务器的时候，我们就把数组申请好

void initServer()
{
    // 1.创建套接字
    _sock.sock();
    _sock.Bind(_port);
    _sock.Listen();

    // 2.创建epoll模型
    _epoll.Create();

    // 3. 将目前唯一的一个sock，添加到epoller中， 之前需要先将对应的fd设置成为非阻塞
    // Util::SetNonBlock(_sock.Fd());
    // _epoll.AddEvents(_sock.Fd(), EPOLLIN | EPOLLET);
    AddConnection(_sock.Fd(), EPOLLIN | EPOLLET,
                  bind(&TcpServer::Accepter, this, placeholders::_1), nullptr, nullptr);

    // 4.拷贝数组
    _revents = new struct epoll_event[_num];
}

然后接下来就可以进行事件派发了

void Loop(int &timeout)
{
    int n = _epoll.Wait(_revents, _num, timeout); // 获取已经就绪的事件

    for (int i = 0; i < n; ++i) // 下面一定是就绪的事件
    {
        uint32_t event = _revents[i].events; // 就绪的事件
        int sock = _revents[i].data.fd;      // 就绪事件的fd

        // 事件派发
        if ((event & EPOLLIN) && sock == _sock.Fd()) // listensock读就绪
            _connections[sock]->recver_(_connections[sock]);
        if ((event & EPOLLIN) && sock != _sock.Fd()) // 普通sock读就绪
            _connections[sock]->recver_(_connections[sock]);
}

我们注意到不管是_listensock套接字还是普通sock套接字读就绪，它们的处理方法都是一样的，因此我们把它们写在一块。只要是读事件事件，我们都根据文件描述去执行对应的读方法，未来设怎么读都由自己曾经设置对应recver_来定

void Loop(int &timeout)
{
    int n = _epoll.Wait(_revents, _num, timeout); // 获取已经就绪的事件

    for (int i = 0; i < n; ++i) // 下面一定是就绪的事件
    {
        uint32_t event = _revents[i].events; // 就绪的事件
        int sock = _revents[i].data.fd;      // 就绪事件的fd

        // 事件派发

        // if ((event & EPOLLIN) && sock == _sock.Fd()) // listensock读就绪
        //     _connections[sock]->recver_(_connections[sock]);
        // if ((event & EPOLLIN) && sock != _sock.Fd()) // 普通fd读就绪
        //     _connections[sock]->recver_(_connections[sock]);

        if ((event & EPOLLIN))//读事件就绪
            _connections[sock]->recver_(_connections[sock]);
        if ((event & EPOLLOUT))//写事件就绪
            _connections[sock]->sender_(_connections[sock]);
}

这样写还有一些问题，首先我们还要判断一下这个套接字是否在unordered_map真的存在，其次在判断对应套接字的方法是否曾经被设置，不然也可能会存在调用的问题

bool ExitHashSock(int sock)
{
    auto pos = _con.find(sock);
    return pos != _con.end();
}

void Loop(int &timeout)
{
    int n = _epoll.Wait(_revents, _num, timeout); // 获取已经就绪的事件

    for (int i = 0; i < n; ++i) // 下面一定是就绪的事件
    {
        uint32_t event = _revents[i].events; // 就绪的事件
        int sock = _revents[i].data.fd;      // 就绪事件的fd

        // 事件派发


       // _listensock和普通sock都有自己对应的回调方法,因此对_listensock和普通sock处是一样的,不用区分
       if ((event & EPOLLIN) && IsConnectionExists(sock) && (_connections[sock]->recver_))
           _connections[sock]->recver_(_connections[sock]);
       if ((event & EPOLLOUT) && IsConnectionExists(sock) && (_connections[sock]->sender_))
           _connections[sock]->sender_(_connections[sock]);
}

这里还有一丢丢小问题，对我们来说我们从来没说过异常的事件，但并不排除在通信的时候存在其他epoll对应的事件

EPOLLERR : 表示对应的文件描述符发生错误;
EPOLLHUP : 表示对应的文件描述符被挂断;

我们可以这样写，如果当前就绪的事件出现异常了，我们设置进该就绪事件中的读和写去处理。这样的好处是将所有的异常问题，全部转化，成为读写问题 ，本来读写本来就要做异常处理，只要把读写异常处理了，这个异常也就被自动处理了！

void Loop(int &timeout)
{
    int n = _epoll.Wait(_revents, _num, timeout); // 获取已经就绪的事件

    for (int i = 0; i < n; ++i) // 下面一定是就绪的事件
    {
        uint32_t event = _revents[i].events; // 就绪的事件
        int sock = _revents[i].data.fd;      // 就绪事件的fd

		//将所有的异常问题，全部转化，成为读写问题
        if (event & EPOLLERR)
            event | EPOLLIN | EPOLLOUT;
        if (event & EPOLLHUP)
            event | EPOLLIN | EPOLLOUT;

       // _listensock和普通sock都有自己对应的回调方法,因此对_listensock和普通sock处是一样的,不用区分
       if ((event & EPOLLIN) && IsConnectionExists(sock) && (_connections[sock]->recver_))
           _connections[sock]->recver_(_connections[sock]);
       if ((event & EPOLLOUT) && IsConnectionExists(sock) && (_connections[sock]->sender_))
           _connections[sock]->sender_(_connections[sock]);
}

因为我们目前就一个_listensock套接字，所以到目前为止有事件就绪就是_listensock套接字就绪，也就是这里只会执行_listensock套接字设置的读事件也就是我们曾经给_listensock套接字绑定的Accepter函数来获取新链接。

void Accepter(Connection *conn)
{
       string clientip;
       uint16_t clientport;
       int err = 0;//获取错误码,用来判断是非阻塞了/读取被打断了/还是真错误了
       int sock = _sock.Accept(&clientip, &clientport, &err);
       if (sock > 0)
       {
           // 新的sock套接字添加到AddConnetions
           AddConnetions(sock, EPOLLIN | EPOLLET,);
       }
}

对于未来获取到的普通文件描述符，它还要有对应的事件就绪时我们需要怎么处理这个套接字，所以我们需要设置一个对普通文件描述符就绪时它的处理读，写，异常的方法

void recver(Connection *conn)
{
}

void sender(Connection *conn)
{
}

void excepter(Connection *conn)
{
}

void Accepter(Connection *conn)
{
       string clientip;
       uint16_t clientport;
       int err = 0;//获取错误码,用来判断是非阻塞了/读取被打断了/还是真错误了
       int sock = _sock.Accept(&clientip, &clientport, &err);
       if (sock > 0)
       {
           // 新的sock套接字添加到AddConnetions
           AddConnetions(sock, EPOLLIN | EPOLLET,
                              bind(&TcpServer::recver, this, placeholders::_1),
                              bind(&TcpServer::sender, this, placeholders::_1),
                              bind(&TcpServer::excepter, this, placeholders::_1));
       }
}

现在有一个问题时，当时_listesock套接字注册的时候也是ET模式的，并且也设置成非阻塞的了，所以通知的时候只会通知_listesock套接字读就绪一次，Accepter调用也就一次。

在这里插入图片描述

也就是说你能保证底层到来的链接就一个吗？不能！你并不能保证此时正在处理_listesock套接字获取新连接时有几个到来了，有可能同时到来了非常多的连接，所以倒逼程序员必须把所有到达的连接全部都读上来，所有上面Accepet的写法是不对的，必须要循环读取全部都读上来！而且我们早把_listesock设置为非阻塞了，底层没有连接了也不会被阻塞，虽然返回值是-1，但是错误码被设置！可能是底层没错误了错误码被设为EAGAIN 或EWOULDBLOCK，或者读取时被信号中断了错误码被设为EINTR，或者真错误了！

void Accepter(Connection *conn)
{
    while (true)
    {
        string clientip;
        uint16_t clientport;
        int err = 0; // 获取错误码,用来判断是非阻塞了/读取被打断了/还是真错误了
        int sock = _sock.Accept(&clientip, &clientport, &err);
        if (sock > 0)
        {
            // 新的sock套接字添加到AddConnetions
            AddConnection(sock, EPOLLIN | EPOLLET,
                          bind(&TcpServer::recver, this, placeholders::_1),
                          bind(&TcpServer::sender, this, placeholders::_1),
                          bind(&TcpServer::excepter, this, placeholders::_1));

            logMessage(DEBUG, "get a new link, info: [%s:%d]", clientip.c_str(), clientport);
        }
        else
        {
            if (err == EAGAIN || err == EWOULDBLOCK)
                break;
            else if (err == EINTR)
                continue;
            else
                break;
        }
    }
}

上面处理好了，接下来就是处理普通套接字就绪事件了，假设现在读就绪了，我们先处理读事件。怎么读？注定了也要循环读取！

其次你即使把本来数据全部读完就能保证是一个完整报文吗？不能！不是就没有办法向上层交，也就意味着没有办法对它进行处理，此时我们就先把读到的数据放到对应fd的Connection对象的输入缓存区里

void recver(Connection *conn)
{
	   while (true)
	   {
	       char buffer[1024];
	       ssize_t s = read(conn->sock_, buffer, sizeof(buffer) - 1);
	       if (s > 0)//读取成功
	       {
	           buffer[s] = 0;
	           conn->inbuffer_ += buffer;
	       }
	       
	    }
}

如果s==0，就说明对方把连接关掉了，那我们是不是应该把对应的文件描述符从epoll中删除，然后关闭文件描述符，然后从unordered_map中移除等等工作。如果写出问题呢？我们的代码是不是会充满大量的异常调用，所以只要我们设置了对异常的处理，然后有问题的执行一次异常方法就行了。如果s == -1，我们内部再判断是非阻塞返回、是被信号中断、还是着出错误了。

void recver(Connection *conn)
{
	while (true)
	{
	   char buffer[1024];
	   ssize_t s = read(conn->sock_, buffer, sizeof(buffer) - 1);
	   if (s > 0)//读取成功
	   {
	       buffer[s] = 0;
	       conn->inbuffer_ += buffer;
	   }           
	   else if (s == 0)
	      {
	          if (conn->excepter_)
	          {
	              conn->excepter_(conn);
	              return;
	          }
	      }
	      else
	      {
	          if (errno == EAGAIN || errno == EWOULDBLOCK)//只是没数据了
	              break;
	          else if (errno == EIDRM)//读取时被信号中断，继续读
	              continue;
	          else//真错误了
	          {
	              if (conn->excepter_)
	              {
	                  conn->excepter_(conn);
	                  return;
	              }
	          }
	      }
	      	       
	}
}

接下来我们处理一下读取到的报文，如果是完整报文就向上交付，不是就继续读取直到是一个完整报文。

那问题来了你怎么知道是一个完整的报文？
还记得以前在写网络版本计数器是怎么处理的吗？没错定制一个协议！！！
我们以前学过，今天拿过来用就行了 协议封装

#pragma once

#include <iostream>
#include <string>
#include <cstring>
#include <sys/types.h>
#include <sys/socket.h>
#include <jsoncpp/json/json.h>

#define SEP " "
#define SEP_LEN strlen(SEP)
#define LINE_SEP "\r\n"
#define LINE_SEP_LEN strlen(LINE_SEP)

using namespace std;

// "x op y"  -> "content_len"\r\n"x op y"\r\n
string Enlenth(const string &text)//加报头
{
    string send_string = to_string(text.size());
    send_string += LINE_SEP;
    send_string += text;
    send_string += LINE_SEP;

    return send_string;
}

//"content_len"\r\n"x op y"\r\n  -> "x op y"
bool Delenth(const string &packge, string *text)//删除报头
{
    auto pos = packge.find(LINE_SEP);
    if (pos == string::npos)
        return false;
    string text_len_string = packge.substr(0, pos);
    int text_len = stoi(text_len_string);
    *text = packge.substr(pos + LINE_SEP_LEN, text_len);
    return true;
}

class Request//请求
{
public:
    Request() : _x(0), _y(0), _op(0)
    {
    }

    Request(int x, int y, char op) : _x(x), _y(y), _op(op)
    {
    }

    bool serialize(string *out)//序列化
    {
#ifdef MYSELF
        // 结构化 -> "x op y"
        *out = "";
        string x_string = to_string(_x);
        string y_string = to_string(_y);

        *out += x_string;
        *out += SEP;
        *out += _op;
        *out += SEP;
        *out += y_string;
#else
        Json::Value root;
        root["first"] = _x;
        root["second"] = _y;
        root["oper"] = _op;

        Json::FastWriter write;
        *out = write.write(root);
#endif
        return true;
    }

    bool deserialize(const string &in)//反序列
    {
#ifdef MYSELF
        // "x op y" -> 结构化
        auto left = in.find(SEP);
        auto right = in.rfind(SEP);
        if (left == string::npos || right == string::npos)
            return false;
        if (left == right)
            return false;
        if (right - (left + SEP_LEN) != 1)
            return false;

        string x_string = in.substr(0, left); // [0, 2) [start, end) , start, end - start
        string y_string = in.substr(right + SEP_LEN);

        if (x_string.empty())
            return false;
        if (y_string.empty())
            return false;
        _x = stoi(x_string);
        _y = stoi(y_string);
        _op = in[left + SEP_LEN];
#else
        Json::Value root;
        Json::Reader reader;
        reader.parse(in, root);

        _x = root["first"].asInt();
        _y = root["second"].asInt();
        _op = root["oper"].asInt();

#endif

        return true;
    }

public:
    int _x;
    int _y;
    char _op;
};

class Response//响应
{
public:
    Response() : _exitcode(0), _result(0)
    {
    }

    Response(int exitcode, int result) : _exitcode(exitcode), _result(result)
    {
    }

    bool serialize(string *out)//序列化
    {
#ifdef MYSELF
        // 结构化 -> "_exitcode  _result"
        *out = "";
        *out = to_string(_exitcode);
        *out += SEP;
        *out += to_string(_result);
#else
        Json::Value root;
        root["first"] = _exitcode;
        root["second"] = _result;

        Json::FastWriter write;
        *out = write.write(root);
#endif
        return true;
    }

    bool deserialize(const string &in)//反序列化
    {
#ifdef MYSELF
        //"_exitcode  _result" ->结构化
        auto pos = in.find(SEP);
        if (pos == string::npos)
            return false;

        string ec_string = in.substr(0, pos);
        string res_string = in.substr(pos + SEP_LEN);

        if (ec_string.empty())
            return false;
        if (res_string.empty())
            return false;

        _exitcode = stoi(ec_string);
        _result = stoi(res_string);
#else
        Json::Value root;
        Json::Reader reader;
        reader.parse(in, root);
        _exitcode = root["first"].asInt();
        _result = root["second"].asInt();
#endif
        return true;
    }

public:
    int _exitcode; // 0：计算成功，!0表示计算失败，具体是多少，定好标准
    int _result;   // 计算结果
};

//这里是读取一个完整报文！读到后放到text
bool PartOnepackge(string &inbuffer, string *text)
{
    //"content_len"/r/n"x op y"/r/n

    auto pos = inbuffer.find(LINE_SEP);
    if (pos == string::npos) // 没读到一个完整报文
        return false;
    //"content_len"/r/n"x op y"/r/n"content_len" >= 一个完整报文长度
    string text_len_string = inbuffer.substr(0, pos);
    int text_len = stoi(text_len_string);
    int total_len = text_len_string.size() + 2 * LINE_SEP_LEN + text_len;

    cout << "处理前#inbuffer: \n"
         << inbuffer << std::endl;

    if (inbuffer.size() < total_len) // 也没有读到一个完整报文
    {
        cout << "你输入的消息，没有严格遵守我们的协议，正在等待后续的内容, continue" << endl;
        return false;
    }

    // 至少有一个完整的报文
    *text = inbuffer.substr(0, total_len); // 读到一个完整报文
    inbuffer.erase(0, total_len);          // inbuffer内部减去这次读到的一个完整的报文

    cout << "处理后#inbuffer:\n " << inbuffer << endl;
    return true;

}

现在协议定义好了，我们也可以开始对读到的数据进行处理了！这里我们只想让服务器就处理IO事件，因此设置一个回调函数专门处理里面是否读取到一个完整报文，读到一个完整报文就序列化反序列业务处理等等。

using func_t = function<void(Connection *)>;

class TcpServer
{

public:
    TcpServer(func_t f,int port = defaultport, int num = defaultnum) : _service(f),_port(port), _revents(nullptr), _num(num)
    {
    }
	//。。。

private:
    uint16_t _port;

    Sock _sock;
    Epoller _epoll;
    unordered_map<int, Connection *> _connections; // 所有链接集合
    struct epoll_event *_revents;
    int _num;

    func_t _service;//业务处理
};

然后读取一次就交给上层一次，让上层进行处理，而recver专心读取数据就行了。

void recver(Connection *conn)
{
    while (true)
    {
        char buffer[1024];
        ssize_t s = read(conn->sock_, buffer, sizeof(buffer) - 1);
        if (s > 0)
        {
            buffer[s] = 0;
            conn->inbuffer_ += buffer;
            _service(conn);//处理业务
        }
        else if (s == 0)
        {
            if (conn->excepter_)
            {
                conn->excepter_(conn);
                return;
            }
        }
        else
        {
            if (errno == EAGAIN || errno == EWOULDBLOCK)
                break;
            else if (errno == EIDRM)
                continue;
            else
            {
                if (conn->excepter_)
                {
                    conn->excepter_(conn);
                    return;
                }
            }
        }
    }
}

如果读取到一个完整报文后就对请求就行处理，没有就返回继续读取，直到读取到完整的报文

void calculate(Connection *conn)
{
    string onePackage;
    while (PartOnepackge(conn->inbuffer_, &onePackage))
    {
        string req_str;
        // 1.2 我们保证，我们req_text里面一定是一个完整的请求："content_len"\r\n"x op y"\r\n
        if (!Delenth(onePackage, &req_str))
            return;

        cout << "去掉报头的正文：\n"
             << req_str << endl;

        // 2. 对请求Request，反序列化
        // 2.1 得到一个结构化的请求对象
        Request req;
        if (!req.deserialize(req_str))
            return;


        // 3. 计算机处理，req.x, req.op, req.y --- 业务逻辑
        // 3.1 得到一个结构化的响应
        Response resp;
        cal(req, resp); // req的处理结果，全部放入到了resp， 回调是不是不回来了？不是！

}

业务处理我们也写过直接用

void cal(const Request &req, Response &resp)
{
     req已经有结构化完成的数据啦，你可以直接使用
    resp._exitcode = OK;
    resp._result = OK;

    switch (req._op)
    {
    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 = DIV_ERR;
        else
            resp._result = req._x / req._y;
    }
    break;
    case '%':
    {
        if (req._y == 0)
            resp._exitcode = MOD_ERR;
        else
            resp._result = req._x % req._y;
    }
    break;
    default:
        resp._exitcode = OPER_ERR;
        break;
    }
}

计算处理好了我们对响应序列化，得到一个响应的报文，接下来就是给对方发回去

void calculate(Connection *conn)
{
    string onePackage;
    while (PartOnepackge(conn->inbuffer_, &onePackage))
    {
        string req_str;
        // 1.2 我们保证，我们req_text里面一定是一个完整的请求："content_len"\r\n"x op y"\r\n
        if (!Delenth(onePackage, &req_str))
            return;

        cout << "去掉报头的正文：\n"
             << req_str << endl;

        // 2. 对请求Request，反序列化
        // 2.1 得到一个结构化的请求对象
        Request req;
        if (!req.deserialize(req_str))
            return;


        // 3. 计算机处理，req.x, req.op, req.y --- 业务逻辑
        // 3.1 得到一个结构化的响应
        Response resp;
        cal(req, resp); // req的处理结果，全部放入到了resp， 回调是不是不回来了？不是！


        // 4.对响应Response，进行序列化
        // 4.1 得到了一个"字符串"
        string resp_str;
        if (!resp.serialize(&resp_str))
            return;

        // 5 构建成为一个完整的报文
       string send_string= Enlenth(resp_str);
       cout << "--------------result: " << send_string << endl;
		
	   //发回去
    }

}

现在问题是，在epoll中能不能直接发送呢？？
不能！用户无法保证发送条件是就绪的！谁最清楚？？epoll！！
也就是说不光读事件要添加到epoll中，写事件是否就绪也要添加到epoll里。
什么叫做写发送事件就绪呢？
发送缓存区有空间！

服务器刚开始启动，或者很多情况下，发送事件是一直就绪的！可以直接发送！
只不过，如果我们没有发完怎么办？那就需要下一次发送，这里也就要求每一个sock都要有自己的发送缓存区！

所以根据上面的分析，大部分情况发送缓存区都是有空间的，所以其实可以直接发，有人说直接发那我就循环发，是的可以循环发，如果没发完非阻塞就直接返回错了，下次发就行了，可是下次从哪里发呢，所以就需要sock有自己的发送缓存区。

因为大部分情况下发送缓存区都是就绪的，所以直接发，没发完没关系下次在发，可是下次是什么时候，这次没发完不就是用户从缓存区数据量太大把发送缓冲区打满了，那下次发是不是就要等底层发送缓存区事件就绪我再发。所以注定你要将sock的EPOLLOUT事件也要有时候注册进epoll ，让epoll通知我，如果底层发事件就绪了我再发。现在问题是是什么时候注册？在新sock注册到epoll里的时候就开始关心写了可不可以？

在这里插入图片描述

未来事件派发一旦写事件就绪了自动调用sender方法

在这里插入图片描述

看起来很完美，但是一般读事件对于epoll我们要常设，写事件的关心对于epoll我们要按需设置！

也就是说当你需要发的时候，确实发送没发完，你在想办法在epoll中设置一下，让epoll关心对于fd的写事件！ 刚开始设置sock如果你真的需要关心写事件你才需要设置写事件，不要把写事件常设到epoll中，因为你把它常设了，epoll就会一直在返回，因为每一个文件描述符的发送缓存区大概率都是就绪的，就一直的返回，你的处理逻辑就一直在消耗cpu资源，这样没必要，所以写事件我们按序设置。

所以我们的结论就是，因为是ET模式可以直接发送，发完就ok，没发完就让epoll关心一下对应的写事件的关心，所以我们需要常设EPOLLIN的关心，对EPOLLOUT按需设置！

然后这里呢，可能一次发不完，为了下次继续发送，所以我们把response序列化的之后的结果放到该fd的输出缓存区，以供下次发送。然后调用对应fd的读方法把数据发出去。

void calculate(Connection *conn)
{
    string onePackage;
    while (PartOnepackge(conn->inbuffer_, &onePackage))
    {
        string req_str;
        // 1.2 我们保证，我们req_text里面一定是一个完整的请求："content_len"\r\n"x op y"\r\n
        if (!Delenth(onePackage, &req_str))
            return;

        cout << "去掉报头的正文：\n"
             << req_str << endl;

        // 2. 对请求Request，反序列化
        // 2.1 得到一个结构化的请求对象
        Request req;
        if (!req.deserialize(req_str))
            return;


        // 3. 计算机处理，req.x, req.op, req.y --- 业务逻辑
        // 3.1 得到一个结构化的响应
        Response resp;
        cal(req, resp); // req的处理结果，全部放入到了resp， 回调是不是不回来了？不是！


        // 4.对响应Response，进行序列化
        // 4.1 得到了一个"字符串"
        string resp_str;
        if (!resp.serialize(&resp_str))
            return;

        // 5 构建成为一个完整的报文
        conn->outbuffer_ += Enlenth(resp_str);

        std::cout << "--------------result: " << conn->outbuffer_ << std::endl;
        
		// 直接发
	    if (conn->sender_)
	        conn->sender_(conn);
    }

}

调用sender并不能保证数据一次就发完了，因此如果没有发送完毕，需要对对应的sock开启对写事件的关系，如果发完了，我们要关闭对写事件的关心！

void sender(Connection *conn)
{
    while (true)
    {
        ssize_t s = send(conn->sock_, conn->outbuffer_.c_str(), conn->outbuffer_.size(), 0);
        if (s > 0)
        {
            conn->outbuffer_.erase(0, s);
            if (conn->outbuffer_.empty())//发完就退出
                break;
        }
        else
        {
            if (errno == EAGAIN || errno == EWOULDBLOCK)//没发完非阻塞退出
                break;
            else if (errno == EINTR)
                continue;
            else
            {
                if (conn->excepter_)
                {
                    conn->excepter_(conn);
                    return;
                }
            }
        }
    }
    
    // 如果没有发送完毕，需要对对应的sock开启对写事件的关系， 如果发完了，我们要关闭对写事件的关心！
}

这里我们在设计一个接口，可以打开对于一个文件描述符，是否对它写事件关心，是否对它读事件关心的情况

bool EnableReadWrite(int sock, bool reader, bool writer)
{
    uint32_t event = (reader ? EPOLLIN : 0) | (writer ? EPOLLOUT : 0) | EPOLLET;
    //_epoll.Control();
}

Epoller类在设计一个函数，在epoll对对应的fd关心事件进行修改

bool Control(int sock, uint32_t event, int action)
{
    int n = 0;
    if (action == EPOLL_CTL_MOD)
    {
        struct epoll_event ev;
        ev.events = event;
        ev.data.fd = sock;
        n = epoll_ctl(_epfd, action, sock, &ev);
    }
    else if (action == EPOLL_CTL_DEL)
    {
        n = epoll_ctl(_epfd, action, sock, nullptr);//删除就不关心任何事件了
    }
    else
        n = -1;
    return n == 0;
}

bool EnableReadWrite(int sock, bool reader, bool writer)
{
    uint32_t event = (reader ? EPOLLIN : 0) | (writer ? EPOLLOUT : 0) | EPOLLET;
    _epoll.Control(sock, event, EPOLL_CTL_MOD);
}

void sender(Connection *conn)
{
    while (true)
    {
        ssize_t s = send(conn->sock_, conn->outbuffer_.c_str(), conn->outbuffer_.size(), 0);
        if (s > 0)
        {
            conn->outbuffer_.erase(0, s);
            if (conn->outbuffer_.empty())
                break;
        }
        else
        {
            if (errno == EAGAIN || errno == EWOULDBLOCK)
                break;
            else if (errno == EINTR)
                continue;
            else
            {
                if (conn->excepter_)
                {
                    conn->excepter_(conn);
                    return;
                }
            }
        }
    }
    // 如果没有发送完毕，需要对对应的sock开启对写事件的关系， 如果发完了，我们要关闭对写事件的关心！
    if (!conn->outbuffer_.empty()) // 设置关心该fd读
        EnableReadWrite(conn->sock_, true, true);
    else
        EnableReadWrite(conn->sock_, true, false);
}

只要我开启了对写事件的关心，epoll就会帮我关心对应fd的写事件，如果就绪了，就会调用对应的fd的sender，以上就是发送的逻辑。

在这里插入图片描述

如果对方连接出异常了怎么办呢？不管是读出现异常，还是写出现异常！并且在前面我们是所有异常情况全部转化成读写问题，然后所有读写异常我们都是调用excepter进行处理，换句话说所有异常处理都在这里。

我们处理异常逻辑很简单，我们也关闭对应文件描述符。

void excepter(Connection *conn)
{
    logMessage(DEBUG, "Excepter begin");
    //1，先在epoll中删除
    _epoll.Control(conn->sock_, 0, EPOLL_CTL_DEL);
    //2. 然后在unordered_map中删除
     _connections.erase(conn->sock_);
     //3.关闭fd
    close(conn->sock_);

    logMessage(DEBUG, "关闭%d 文件描述符的所有的资源", conn->sock_);

    delete conn;
}

到目前为止所有代码细节我们都写完了，对IO的处理+协议定制+业务逻辑都有，下面验证一下，我们看到能够正常通信，并且还有对异常的处理

在这里插入图片描述
Epoller封装完整代码

#pragma once

#include <iostream>
#include <sys/epoll.h>
#include <unistd.h>
#include <sys/types.h>
#include "Err.hpp"
#include "Log.hpp"

using namespace std;

const int defaultepfd = -1;
const int size = 128;

class Epoller
{

public:
    Epoller(int epfd = defaultepfd) : _epfd(epfd)
    {
    }

    ~Epoller()
    {
        if (_epfd != defaultepfd)
            close(_epfd);
    }

public:
    void Create()
    {
        _epfd = epoll_create(size);
        if (_epfd < 0)
        {
            logMessage(FATAL, "epoll_create error, code: %d, errstring: %s", errno, strerror(errno));
            exit(EPOLL_CREATE_ERR);
        }
        logMessage(NORMAL, "epoll create success, epfd: %d", _epfd);
    }

    // user -> kernel
    bool AddEvents(int sock, uint32_t event)
    {
        struct epoll_event ev;
        ev.events = event;
        ev.data.fd = sock;
        int n = epoll_ctl(_epfd, EPOLL_CTL_ADD, sock, &ev);
        if (n < 0)
        {
            logMessage(ERROR, "sock join epoll fail");
            return false;
        }
        // logMessage(NORMAL, "sock join epoll success");
        return true;
    }

    // kernel -> user
    int Wait(struct epoll_event *ets, int num, int timeout)
    {
        int n = epoll_wait(_epfd, ets, num, timeout);
        switch (n)
        {
        case 0:
            logMessage(NORMAL, "timeout ...");
            break;
        case -1:
            logMessage(WARNING, "epoll_wait failed, code: %d, errstring: %s", errno, strerror(errno));
            break;
        default:
            logMessage(NORMAL, "have event ready");
            break;
        }
        return n;
    }

    bool Control(int sock, uint32_t event, int action)
    {
        int n = 0;
        if (action == EPOLL_CTL_MOD)
        {
            struct epoll_event ev;
            ev.events = event;
            ev.data.fd = sock;
            n = epoll_ctl(_epfd, action, sock, &ev);
        }
        else if (action == EPOLL_CTL_DEL)
        {
            n = epoll_ctl(_epfd, action, sock, nullptr);
        }
        else
            n = -1;
        return n == 0;
    }

private:
    int _epfd;
};

服务器完整代码

#pragma once

#include <iostream>
#include <unordered_map>
#include <functional>
#include <cassert>
#include "Sock.hpp"
#include "Err.hpp"
#include "Log.hpp"
#include "Epoller.hpp"
#include "Util.hpp"

const int defaultport = 8080;
const int defaultsock = -1;
const int defaultnum = 128;

class Connection;

using func_t = function<void(Connection *)>;

class Connection
{
public:
    Connection(int sock = defaultport) : sock_(sock)
    {
    }

    ~Connection()
    {
        if (sock_ == defaultsock)
            close(sock_);
    }

    void Rigster(func_t r, func_t s, func_t e)
    {
        recver_ = r;
        sender_ = s;
        excepter_ = e;
    }

public:
    int sock_;         // 这个类对应的套接字是谁
    string inbuffer_;  // 输入缓存区,这里我们暂时没有考虑处理图片,视频等二进制信息,不然stirng就不太合适了
    string outbuffer_; // 输出缓存区 ,你并不能保证你的写事件就绪

    func_t recver_;   // 从sock_读
    func_t sender_;   // 向sock_写
    func_t excepter_; // 处理sock_,IO的时候上面的异常事件
};

class TcpServer
{

public:
    TcpServer(func_t f, int port = defaultport, int num = defaultnum) : _service(f), _port(port), _revents(nullptr), _num(num)
    {
    }

    ~TcpServer()
    {
        if (_revents)
            delete[] _revents;
    }

    void initServer()
    {
        // 1.创建套接字
        _sock.sock();
        _sock.Bind(_port);
        _sock.Listen();

        // 2.创建epoll模型
        _epoll.Create();

        // 3. 将目前唯一的一个sock，添加到epoller中， 之前需要先将对应的fd设置成为非阻塞
        // Util::SetNonBlock(_sock.Fd());
        // _epoll.AddEvents(_sock.Fd(), EPOLLIN | EPOLLET);
        AddConnection(_sock.Fd(), EPOLLIN | EPOLLET,
                      bind(&TcpServer::Accepter, this, placeholders::_1), nullptr, nullptr);

        // 4.拷贝数组
        _revents = new struct epoll_event[_num];
    }

    // 事件派发
    void Dispatch()
    {
        int timeout = -1000;
        while (true)
        {
            Loop(timeout);
        }
    }

private:
    bool IsConnectionExists(int sock)
    {
        auto pos = _connections.find(sock);
        return pos != _connections.end();
    }

    void Loop(int &timeout)
    {
        int n = _epoll.Wait(_revents, _num, timeout); // 获取已经就绪的事件

        for (int i = 0; i < n; ++i) // 下面一定是就绪的事件
        {
            uint32_t event = _revents[i].events; // 就绪的事件
            int sock = _revents[i].data.fd;      // 就绪事件的fd

            // 将所有的异常问题，全部转化，成为读写问题
            if (event & EPOLLERR)
                event | EPOLLIN | EPOLLOUT;
            if (event & EPOLLHUP)
                event | EPOLLIN | EPOLLOUT;

            // _listensock和普通sock都有自己对应的回调方法,因此对_listensock和普通sock处是一样的,不用区分
            if ((event & EPOLLIN) && IsConnectionExists(sock) && (_connections[sock]->recver_))
                _connections[sock]->recver_(_connections[sock]);
            if ((event & EPOLLOUT) && IsConnectionExists(sock) && (_connections[sock]->sender_))
                _connections[sock]->sender_(_connections[sock]);
        }
    }

    void recver(Connection *conn)
    {
        while (true)
        {
            char buffer[1024];
            ssize_t s = read(conn->sock_, buffer, sizeof(buffer) - 1);
            if (s > 0)
            {
                buffer[s] = 0;
                conn->inbuffer_ += buffer;
                _service(conn);
            }
            else if (s == 0)
            {
                if (conn->excepter_)
                {
                    conn->excepter_(conn);
                    return;
                }
            }
            else
            {
                if (errno == EAGAIN || errno == EWOULDBLOCK)
                    break;
                else if (errno == EIDRM)
                    continue;
                else
                {
                    if (conn->excepter_)
                    {
                        conn->excepter_(conn);
                        return;
                    }
                }
            }
        }
    }

    bool EnableReadWrite(int sock, bool reader, bool writer)
    {
        uint32_t event = (reader ? EPOLLIN : 0) | (writer ? EPOLLOUT : 0) | EPOLLET;
        _epoll.Control(sock, event, EPOLL_CTL_MOD);
    }

    void sender(Connection *conn)
    {
        while (true)
        {
            ssize_t s = send(conn->sock_, conn->outbuffer_.c_str(), conn->outbuffer_.size(), 0);
            if (s > 0)
            {
                conn->outbuffer_.erase(0, s);
                if (conn->outbuffer_.empty())
                    break;
            }
            else
            {
                if (errno == EAGAIN || errno == EWOULDBLOCK)
                    break;
                else if (errno == EINTR)
                    continue;
                else
                {
                    if (conn->excepter_)
                    {
                        conn->excepter_(conn);
                        return;
                    }
                }
            }
        }
        // 如果没有发送完毕，需要对对应的sock开启对写事件的关系， 如果发完了，我们要关闭对写事件的关心！
        if (!conn->outbuffer_.empty()) // 设置关心该fd读
            EnableReadWrite(conn->sock_, true, true);
        else
            EnableReadWrite(conn->sock_, true, false);
    }

    void excepter(Connection *conn)
    {
        logMessage(DEBUG, "Excepter begin");
        _epoll.Control(conn->sock_, 0, EPOLL_CTL_DEL);
        _connections.erase(conn->sock_);
        close(conn->sock_);

        logMessage(DEBUG, "关闭%d 文件描述符的所有的资源", conn->sock_);

        delete conn;
    }
    void Accepter(Connection *conn)
    {
        while (true)
        {
            string clientip;
            uint16_t clientport;
            int err = 0; // 获取错误码,用来判断是非阻塞了/读取被打断了/还是真错误了
            int sock = _sock.Accept(&clientip, &clientport, &err);
            if (sock > 0)
            {
                // 新的sock套接字添加到AddConnetions
                AddConnection(sock, EPOLLIN | EPOLLET,
                              bind(&TcpServer::recver, this, placeholders::_1),
                              bind(&TcpServer::sender, this, placeholders::_1),
                              bind(&TcpServer::excepter, this, placeholders::_1));

                logMessage(DEBUG, "get a new link, info: [%s:%d]", clientip.c_str(), clientport);
            }
            else
            {
                if (err == EAGAIN || err == EWOULDBLOCK)
                    break;
                else if (err == EINTR)
                    continue;
                else
                    break;
            }
        }
    }

    void AddConnection(int sock, uint32_t events, func_t recver, func_t sender, func_t excepter)
    {
        // 1.设置非阻塞,ET模式fd要非阻塞
        if (events & EPOLLET)
            Util::SetNonBlock(sock);

        // 2. 该sock创建Connection，并初始化，并添加到connections_
        Connection *conn = new Connection(sock);

        // 3. 给对应的sock设置对应的回调方法
        conn->Rigster(recver, sender, excepter);

        // 4. 其次将sock与它要关心的事件"写透式"注册到epoll中，让epoll帮我们关心
        bool r = _epoll.AddEvents(sock, events);
        assert(r);
        (void)r;

        // 5. 将kv添加到connections_
        _connections.insert(make_pair(sock, conn));

        logMessage(DEBUG, "add new sock : %d in epoll and unordered_map", sock);
    }

public:
private:
    uint16_t _port;

    Sock _sock;
    Epoller _epoll;
    unordered_map<int, Connection *> _connections; // 所有链接集合
    struct epoll_event *_revents;
    int _num;

    func_t _service;
};

调用逻辑完整代码

#include "TcpServer.hpp"
#include "Err.hpp"
#include <memory>
#include "Protocol.hpp"

enum
{
    OK,
    DIV_ERR,
    MOD_ERR,
    OPER_ERR
};

void cal(const Request &req, Response &resp)
{
     req已经有结构化完成的数据啦，你可以直接使用
    resp._exitcode = OK;
    resp._result = OK;

    switch (req._op)
    {
    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 = DIV_ERR;
        else
            resp._result = req._x / req._y;
    }
    break;
    case '%':
    {
        if (req._y == 0)
            resp._exitcode = MOD_ERR;
        else
            resp._result = req._x % req._y;
    }
    break;
    default:
        resp._exitcode = OPER_ERR;
        break;
    }
}

void calculate(Connection *conn)
{
    string onePackage;
    while (PartOnepackge(conn->inbuffer_, &onePackage))
    {
        string req_str;
        // 1.2 我们保证，我们req_text里面一定是一个完整的请求："content_len"\r\n"x op y"\r\n
        if (!Delenth(onePackage, &req_str))
            return;

        cout << "去掉报头的正文：\n"
             << req_str << endl;

        // 2. 对请求Request，反序列化
        // 2.1 得到一个结构化的请求对象
        Request req;
        if (!req.deserialize(req_str))
            return;


        // 3. 计算机处理，req.x, req.op, req.y --- 业务逻辑
        // 3.1 得到一个结构化的响应
        Response resp;
        cal(req, resp); // req的处理结果，全部放入到了resp， 回调是不是不回来了？不是！


        // 4.对响应Response，进行序列化
        // 4.1 得到了一个"字符串"
        string resp_str;
        if (!resp.serialize(&resp_str))
            return;

        // 5 构建成为一个完整的报文
        conn->outbuffer_ += Enlenth(resp_str);

        std::cout << "--------------result: " << conn->outbuffer_ << std::endl;
    }
    // 直接发
    if (conn->sender_)
        conn->sender_(conn);
}

static void usage(std::string proc)
{
    std::cerr << "Usage:\n\t" << proc << " port" << "\n\n";
}

int main(int argc, char *argv[])
{
    if (argc != 2)
    {
        usage(argv[0]);
        exit(USAGG_ERR);
    }

    uint16_t port=atoi(argv[1]);
    std::unique_ptr<TcpServer> uls(new TcpServer(calculate,port));
    uls->initServer();
    uls->Dispatch();

    return 0;
}

总结一下：这个TcpServer服务器就是传说中的Reactor，对应上服务器上面有一个一个的Connection，未来有哪一个fd就绪了，它就把某一个Connection激活告诉我们事件，Reactor就会进行事件派发。然后去执行对应Connection的读，写，异常方法。这种模式就是Reactor反应堆模式

在这里插入图片描述
Reactor，很明显当我们进行事件就绪了它会回调曾经注册的绑定的方法，今天我们的服务器既保证了就绪事件通知，还负责了IO，其实还负责了业务处理。如果今天我们把一个报文构建成一个任务，扔到后端任务队列，让后端线程池帮我们处理后面的任务。此时对应的Reactor只负责读事件就绪+负责IO，不负责业务处理。

只负责读事件就绪+负责IO 这就是就是半同步
只负责业务处理这就半异步

还有一种服务器少用的Proactor前摄器模式，可以自己了解了解。

如果想把这个服务器改成多多进程多线程，其实可以直接创建多进程，每个进程里都搞一个TcpServer，或者说创建多线程，每一个线程里都搞一个Tcpserver，把_listensock套接字添加到其中一个线程的Reactor里，一旦有连接就绪的时候，不是要执行Accepter吗，执行Accepter的时候就不仅仅是 AddConnection了，而是尝试把这个连接添加到哪一个线程的epoll中，然后就在这个线程里把这个文件描述符处理完。

最后说一点，我们还可以在Connectoin中设置一个lasttime记录最近访问时间，每一次读或写的时候我们都更新一个对应时间戳，所以只要读或写就绪了就可以更新一下对应Connection最近时间，换句话说此时我们就可以在派发事件后当所有事件处理完了，就可以在unordered_map遍历所有的连接，计算每一个连接已经有多长时间没有动了，因为每一个连接都有自己的最近访问时间，每一次访问都会更新，不更新就是最开始的，所以我们可以获取当前时间在减去Connectoin里保存的历史最近访问时间，计算出时间差，然后就可以所以连接进行连接管理。时间超过5分钟都没有访问过的，服务器就直接把你关掉。

在这里插入图片描述