Linux - 第25节 - Linux高级IO（三）

目录

1.Reactor模式

1.1.Reactor模式的定义

1.2.Reactor模式的角色构成

1.3.Reactor模式的工作流程

2.epoll ET服务器（Reactor模式）

2.1.epoll ET服务器源代码

2.2.epoll ET服务器源代码讲解

2.2.1.设计思路

2.2.2.Connection结构

2.2.3.TcpServer类

2.2.4.回调函数

2.2.5.套接字相关

2.2.6.Reactor模式和Proactor模式

1.Reactor模式

1.1.Reactor模式的定义

Reactor反应器模式，也叫做分发者模式或通知者模式，是一种将就绪事件派发给对应服务处理程序的事件设计模式。

1.2.Reactor模式的角色构成

Reactor主要由以下五个角色构成：

角色	解释
Handle（句柄）	用于标识不同的事件，本质就是一个文件描述符。
Sychronous Event Demultiplexer（同步事件分离器）	本质就是一个系统调用，用于等待事件的发生。对于Linux来说，同步事件分离器指的就是I/O多路复用，比如select、poll、epoll等。
Event Handler（事件处理器）	由多个回调方法构成，这些回调方法构成了与应用相关的对于某个事件的处理反馈。
Concrete Event Handler（具体事件处理器）	事件处理器中各个回调方法的具体实现。
Initiation Dispatcher（初始分发器）	初始分发器实际上就是Reactor角色，初始分发器会通过同步事件分离器来等待事件的发生，当对应事件就绪时就调用事件处理器，最后调用对应的回调方法来处理这个事件。

1.3.Reactor模式的工作流程

Reactor模式的工作流程如下：

1.当应用向初始分发器注册具体事件处理器时，应用会标识出该事件处理器希望初始分发器在某个事件发生时向其通知，该事件与Handle关联。
2.初始分发器会要求每个事件处理器向其传递内部的Handle，该Handle向操作系统标识了事件处理器。
3.当所有的事件处理器注册完毕后，应用会启动初始分发器的事件循环，这时初始分发器会将每个事件处理器的Handle合并起来，并使用同步事件分离器等待这些事件的发生。
4.当某个事件处理器的Handle变为Ready状态时，同步事件分离器会通知初始分发器。
5.初始分发器会将Ready状态的Handle作为key，来寻找其对应的事件处理器。
6.初始分发器会调用其对应事件处理器当中对应的回调方法来响应该事件。

2.epoll ET服务器（Reactor模式）

2.1.epoll ET服务器源代码

创建TcpServer.hpp文件，写入下图一所示的代码，创建Epoller.hpp文件，写入下图二所示的代码，创建Log.hpp文件，写入下图三所示的代码，创建Protocol.hpp文件，写入下图四所示的代码，创建Service.hpp文件，写入下图五所示的代码，创建Sock.hpp文件，写入下图六所示的代码，创建Util.hpp文件，写入下图七所示的代码，创建main.cc文件，写入下图八所示的代码，创建Makefile文件，写入下图九所示的代码，使用make命令生成server可执行程序，使用./server 8080命令运行server可执行程序，创建两个新选项卡作为客户端，分别使用telnet 127.0.0.1 8080命令连接服务端，输入ctrl+]进入telnet行，然后回车并输入消息内容发送给服务端，服务端收到消息后进行计算并将计算结果返回给客户端，客户端发送完消息后输入ctrl+]进入telnet行，然后输入quit退出，如下图十所示。

TcpServer.hpp文件：

#pragma once

#include <iostream>
#include <string>
#include <vector>
#include <cerrno>
#include <unordered_map>
#include <functional>
#include "Sock.hpp"
#include "Epoller.hpp"
#include "Log.hpp"
#include "Util.hpp"
#include "Protocol.hpp"

// 基于Reactor模式，编写一个充分读取和写入的，EPOLL(ET)的Server

class Connection;
class TcpServer;

using func_t = std::function<int(Connection *)>;
using callback_t = std::function<int(Connection *, std::string &)>;

// event
class Connection
{
public:
    // 文件描述符
    int sock_;
    TcpServer *R_;
    // 自己的接受和发送缓冲区
    std::string inbuffer_;
    std::string outbuffer_;
    // 回调函数
    func_t recver_; 
    func_t sender_;
    func_t excepter_;

public:
    Connection(int sock, TcpServer *r) : sock_(sock), R_(r)
    {
    }
    void SetRecver(func_t recver) { recver_ = recver; }
    void SetSender(func_t sender) { sender_ = sender; }
    void SetExcepter(func_t excepter) { excepter_ = excepter; }
    ~Connection() {}
};

//Reactor
//单进程：半异步半同步 -- Reactor -- Linux服务最常用 -- 几乎没有之一
// Reactor（tcp）服务器, 即负责事件派发, 又负责IO [又负责业务逻辑的处理]

// Proactor: 前摄模式 -- 其他平台可能出现的模式
// 只负责负责事件派发，就绪的事件推送给后端的进程、线程池， 不关心 负责IO [又负责业务逻辑的处理]

class TcpServer
{
public:
    TcpServer(callback_t cb, int port = 8080) : cb_(cb)
    {
        revs_ = new struct epoll_event[revs_num];
        // 网络功能
        listensock_ = Sock::Socket();
        Util::SetNonBlock(listensock_);
        Sock::Bind(listensock_, port);
        Sock::Listen(listensock_);

        // 多路转接
        epfd_ = Epoller::CreateEpoller();

        // 添加listensock匹配的connection
        AddConnection(listensock_, EPOLLIN | EPOLLET,
                      std::bind(&TcpServer::Accepter, this, std::placeholders::_1), nullptr, nullptr);
    }
    void AddConnection(int sockfd, uint32_t event, func_t recver, func_t sender, func_t excepter)
    {
        if (event & EPOLLET)
            Util::SetNonBlock(sockfd);
        // 添加sockfd到epoll
        Epoller::AddEvent(epfd_, sockfd, event);
        // 将sockfd匹配的Connection也添加到当前的unordered_map中
        Connection *conn = new Connection(sockfd, this);
        conn->SetRecver(recver);
        conn->SetSender(sender);
        conn->SetExcepter(excepter);

        connections_.insert(std::make_pair(sockfd, conn));
        logMessage(DEBUG, "添加新链接到connections成功: %d", sockfd);
    }
    int Accepter(Connection *conn)
    {
        // demo - listensock 也是工作在ET，来一个连接，对应就有事件就绪，那么如何来一批呢？
        while (true)
        {
            std::string clientip;
            uint16_t clientport = 0;
            int sockfd = Sock::Accept(conn->sock_, &clientip, &clientport);
            if (sockfd < 0)
            {
                if (errno == EINTR)
                    continue;
                else if (errno == EAGAIN || errno == EWOULDBLOCK)
                    break;
                else
                {
                    logMessage(WARNING, "accept error");
                    return -1;
                }
            }
            logMessage(DEBUG, "get a new link: %d", sockfd);
            // 注意：默认我们只设置了让epoll帮我们关心读事件，没有关心写事件
            // 为什么没有关注写事件：因为最开始的时候，写空间一定是就绪的！
            // 运行中可能会存在条件不满足 -- 写空间被写满了
            AddConnection(sockfd, EPOLLIN | EPOLLET,
                          std::bind(&TcpServer::TcpRecver, this, std::placeholders::_1),
                          std::bind(&TcpServer::TcpSender, this, std::placeholders::_1),
                          std::bind(&TcpServer::TcpExcepter, this, std::placeholders::_1));
        }
        return 0;
    }

    int TcpRecver(Connection *conn)
    {
        // XXXXXXX\3XXXXXX\3
        while (true)
        {
            char buffer[1024];
            ssize_t s = recv(conn->sock_, buffer, sizeof(buffer) - 1, 0);
            if (s > 0)
            {
                buffer[s] = 0;
                conn->inbuffer_ += buffer;
            }
            else if (s == 0)
            {
                logMessage(DEBUG, "client quit");
                conn->excepter_(conn);
                break;
            }
            else
            {
                if (errno == EINTR)
                    continue;
                else if (errno == EAGAIN || errno == EWOULDBLOCK)
                    break;
                else
                {
                    // 出错了
                    logMessage(DEBUG, "recv error: %d:%s", errno, strerror(errno));
                    conn->excepter_(conn);
                    break;
                }
            }
        }
        // 将本轮全部读取完毕
        std::vector<std::string> result;
        PackageSplit(conn->inbuffer_, &result);
        for (auto &message : result)
        {
            cb_(conn, message);
        }
        return 0;
    }
    int TcpSender(Connection *conn)
    {
        while(true)
        {
            ssize_t n = send(conn->sock_, conn->outbuffer_.c_str(), conn->outbuffer_.size(), 0);
            if(n > 0)
            {
                // 去除已经成功发送的数据
                conn->outbuffer_.erase(0, n);
            }
            else
            {
                if(errno == EINTR) continue;
                else if(errno == EAGAIN || errno == EWOULDBLOCK) break; //发完了，一定是outbuffer清空了吗？不一定(EPOLLOUT打开)
                else
                {
                    conn->excepter_(conn);
                    logMessage(DEBUG, "send error: %d:%s", errno, strerror(errno));
                    break;
                }
            }
        }

        return 0;
    }
    int TcpExcepter(Connection *conn)
    {
        // 0.
        if(!IsExists(conn->sock_)) return -1;
        
        // 所有的服务器异常，都会被归类到这里
        // 坑：一定要先从epoll中移除，然后再关闭fd
        // 1.
        Epoller::DelEvent(epfd_, conn->sock_);
        logMessage(DEBUG, "remove epoll event!");
        // 2.
        close(conn->sock_);
        logMessage(DEBUG, "close fd: %d", conn->sock_);
        // 3. delete conn;
        delete connections_[conn->sock_];
        logMessage(DEBUG, "delete connection object done");
        // 4. 
        connections_.erase(conn->sock_);
        logMessage(DEBUG, "erase connection from connections");

        return 0;
    }
    bool IsExists(int sock)
    {
        auto iter = connections_.find(sock);
        if (iter == connections_.end())
            return false;
        else
            return true;
    }
    // 打开或者关闭对于特定socket是否要关心读或者写
    //EnableReadWrite(sock, true, false);
    //EnableReadWrite(sock, true, true);
    void EnableReadWrite(int sock, bool readable, bool writeable)
    {
        uint32_t event = 0;
        event |= (readable ? EPOLLIN : 0);
        event |= (writeable ? EPOLLOUT : 0);
        Epoller::ModEvent(epfd_, sock, event);
    }
    // 根据就绪事件，将事件进行事件派发
    void Dispatcher()
    {
        int n = Epoller::LoopOnce(epfd_, revs_, revs_num);
        for (int i = 0; i < n; i++)
        {
            int sock = revs_[i].data.fd;
            uint32_t revent = revs_[i].events;
            if(revent & EPOLLHUP) revent |= (EPOLLIN|EPOLLOUT);
            if(revent & EPOLLERR) revent |= (EPOLLIN|EPOLLOUT);

            if (revent & EPOLLIN)
            {
                if (IsExists(sock) && connections_[sock]->recver_)
                    connections_[sock]->recver_(connections_[sock]);
            }
            if (revent & EPOLLOUT)
            {
                if (IsExists(sock) && connections_[sock]->sender_)
                    connections_[sock]->sender_(connections_[sock]);
            }
        }
    }
    void Run()
    {
        while (true)
        {
            Dispatcher();
        }
    }
    ~TcpServer()
    {
        if (listensock_ != -1)
            close(listensock_);
        if (epfd_ != -1)
            close(epfd_);
        delete[] revs_;
    }

private:
    static const int revs_num = 64;
    // 1. 网络socket
    int listensock_;
    // 2. epoll
    int epfd_;
    // 3. 将epoll和上层代码进行结合
    std::unordered_map<int, Connection *> connections_;
    // 4. 就绪事件列表
    struct epoll_event *revs_;
    // 5. 设置完整报文的处理方法
    callback_t cb_;
};

Epoller.hpp文件：

#pragma once
#include <iostream>
#include <cerrno>
#include <cstdlib>
#include <unistd.h>
#include <sys/epoll.h>
#include "Log.hpp"

class Epoller
{
public:
    static const int gsize = 128;
public:
    static int CreateEpoller()
    {
        int epfd = epoll_create(gsize);
        if (epfd < 0)
        {
            logMessage(FATAL, "epoll_create : %d : %s", errno, strerror(errno));
            exit(3);
        }
        return epfd;
    }
    static bool AddEvent(int epfd, 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);
        return n == 0;
    }
    static bool ModEvent(int epfd, int sock, uint32_t event)
    {
        struct epoll_event ev;
        ev.events = event;
        ev.data.fd = sock;
        int n = epoll_ctl(epfd, EPOLL_CTL_MOD, sock, &ev);
        return n == 0;
    }
    static bool DelEvent(int epfd, int sock)
    {
        int n = epoll_ctl(epfd, EPOLL_CTL_DEL, sock, nullptr);
        return n == 0;
    }
    static int LoopOnce(int epfd, struct epoll_event revs[], int num)
    {
        int n = epoll_wait(epfd, revs, num, -1);
        if(n == -1)
        {
            logMessage(FATAL, "epoll_wait : %d : %s", errno, strerror(errno));
        }
        return n;
    }
};

Log.hpp文件：

#pragma once

#include <cstdio>
#include <ctime>
#include <cstdarg>
#include <cassert>
#include <cassert>
#include <cstring>
#include <cerrno>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>

#define DEBUG 0
#define NOTICE 1
#define WARNING 2
#define FATAL 3

const char *log_level[] = {"DEBUG", "NOTICE", "WARINING", "FATAL"};

#define LOGFILE "serverTcp.log"

class Log
{
public:
    Log():logFd(-1)
    {}
    void enable()
    {
        umask(0);
        logFd = open(LOGFILE, O_WRONLY | O_CREAT | O_APPEND, 0666);
        assert(logFd != -1);
        dup2(logFd, 1);
        dup2(logFd, 2);
    }
    ~Log()
    {
        if(logFd != -1) 
        {
            fsync(logFd);
            close(logFd);
        }
    }
private:
    int logFd;
};

// logMessage(DEBUG, "%d", 10);
void logMessage(int level, const char *format, ...)
{
    assert(level >= DEBUG);
    assert(level <= FATAL);

    char *name = getenv("USER");

    char logInfo[1024];
    va_list ap; // ap -> char*
    va_start(ap, format);

    vsnprintf(logInfo, sizeof(logInfo) - 1, format, ap);

    va_end(ap); // ap = NULL

    // 每次打开太麻烦
    FILE *out = (level == FATAL) ? stderr : stdout;
    fprintf(out, "%s | %u | %s | %s\n",
            log_level[level],
            (unsigned int)time(nullptr),
            name == nullptr ? "unknow" : name,
            logInfo);

    fflush(out); // 将C缓冲区中的数据刷新到OS
    fsync(fileno(out));   // 将OS中的数据尽快刷盘

}

Protocol.hpp文件：

#pragma once

#include <iostream>
#include <vector>
#include <cstring>
#include <string>
#include <cstdio>

#define SEP 'X'
#define SEP_LEN sizeof(SEP)

#define CRLF "\r\n"
#define CRLF_LEN strlen(CRLF) // 坑：sizeof(CRLF)
#define SPACE " "
#define SPACE_LEN strlen(SPACE)

// bbbXcc
void PackageSplit(std::string &inbuffer, std::vector<std::string> *result)
{
    while (true)
    {
        std::size_t pos = inbuffer.find(SEP);
        if (pos == std::string::npos)
            break;
        result->push_back(inbuffer.substr(0, pos));
        inbuffer.erase(0, pos + SEP_LEN);
    }
}

struct Request
{
    int x;
    int y;
    char op;
};

struct Response
{
    int code;
    int result;
};

bool Parser(std::string &in, Request *req)
{
    // 1 + 1, 2 * 4, 5 * 9, 6 *1
    std::size_t spaceOne = in.find(SPACE);
    if (std::string::npos == spaceOne)
        return false;
    std::size_t spaceTwo = in.rfind(SPACE);
    if (std::string::npos == spaceTwo)
        return false;

    std::string dataOne = in.substr(0, spaceOne);
    std::string dataTwo = in.substr(spaceTwo + SPACE_LEN);
    std::string oper = in.substr(spaceOne + SPACE_LEN, spaceTwo - (spaceOne + SPACE_LEN));
    if (oper.size() != 1)
        return false;

    // 转成内部成员
    req->x = atoi(dataOne.c_str());
    req->y = atoi(dataTwo.c_str());
    req->op = oper[0];

    return true;
}

void Serialize(const Response &resp, std::string *out)
{
    // "exitCode_ result_"
    std::string ec = std::to_string(resp.code);
    std::string res = std::to_string(resp.result);

    *out = ec;
    *out += SPACE;
    *out += res;
    *out += CRLF;
}

Service.hpp文件：

#pragma once

#include "Protocol.hpp"
#include <functional>

using service_t = std::function<Response (const Request &req)>;

static Response calculator(const Request &req)
{
    Response resp = {0, 0};
    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 '/':
    { // x_ / y_
        if (req.y == 0)
            resp.code = -1; // -1. 除0
        else
            resp.result = req.x / req.y;
    }
    break;
    case '%':
    { // x_ / y_
        if (req.y == 0)
            resp.code = -2; // -2. 模0
        else
            resp.result = req.x % req.y;
    }
    break;
    default:
        resp.code = -3; // -3: 非法操作符
        break;
    }

    return resp;
}

Sock.hpp文件：

#pragma once

#include <iostream>
#include <fstream>
#include <string>
#include <vector>
#include <cstdio>
#include <cstring>
#include <signal.h>
#include <unistd.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <pthread.h>
#include <cerrno>
#include <cassert>

class Sock
{
public:
    static const int gbacklog = 20;

    static int Socket()
    {
        int listenSock = socket(PF_INET, SOCK_STREAM, 0);
        if (listenSock < 0)
        {
            exit(1);
        }
        int opt = 1;
        setsockopt(listenSock, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt));
        return listenSock;
    }
    static void Bind(int socket, uint16_t port)
    {
        struct sockaddr_in local; // 用户栈
        memset(&local, 0, sizeof local);
        local.sin_family = PF_INET;
        local.sin_port = htons(port);
        local.sin_addr.s_addr = INADDR_ANY;

        // 2.2 本地socket信息，写入sock_对应的内核区域
        if (bind(socket, (const struct sockaddr *)&local, sizeof local) < 0)
        {
            exit(2);
        }
    }
    static void Listen(int socket)
    {
        if (listen(socket, gbacklog) < 0)
        {
            exit(3);
        }
    }

    static int Accept(int socket, std::string *clientip, uint16_t *clientport)
    {
        struct sockaddr_in peer;
        socklen_t len = sizeof(peer);

        int serviceSock = accept(socket, (struct sockaddr *)&peer, &len);
        if (serviceSock < 0)
        {
            // 获取链接失败
            return -1;
        }
        if(clientport) *clientport = ntohs(peer.sin_port);
        if(clientip) *clientip = inet_ntoa(peer.sin_addr);
        return serviceSock;
    }
};

Util.hpp文件：

#pragma once

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

class Util
{
public:
    static void SetNonBlock(int fd)
    {
        int fl = fcntl(fd, F_GETFL);
        fcntl(fd, F_SETFL, fl | O_NONBLOCK);
    }
};

main.cc文件：

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

using namespace std;

static void usage(std::string process)
{
    cerr << "\nUsage: " << process << " port\n"
         << endl;
}
int BeginHandler(Connection *conn, std::string &message, service_t service)
{
    // 我们能保证，message一定是一个完整的报文，因为我们已经对它进行了解码
    Request req;
    // 反序列化，进行处理的问题
    if (!Parser(message, &req))
    {
        // 写回错误消息
        return -1;
        // 可以直接关闭连接
        // conn->excepter_(conn);
    }
    // 业务逻辑
    Response resp = service(req);

    std::cout << req.x << " " << req.op << " " << req.y << std::endl;
    std::cout << resp.code << " " << resp.result << std::endl;

    // 序列化
    std::string sendstr;
    Serialize(resp, &sendstr);

    // 处理完毕的结果，发送回给client
    conn->outbuffer_ += sendstr;
    conn->sender_(conn);
    if(conn->outbuffer_.empty()) conn->R_->EnableReadWrite(conn->sock_, true, false);
    else conn->R_->EnableReadWrite(conn->sock_, true, true);

    std::cout << "这里就是上次的业务逻辑啦 --- end" << std::endl;

    return 0;
}

// 1 + 1X2 + 3X5 + 6X8 -> 1 + 1
int HandlerRequest(Connection *conn, std::string &message)
{
    return BeginHandler(conn, message, calculator);
}

int main(int argc, char *argv[])
{
    if (argc != 2)
    {
        usage(argv[0]);
        exit(0);
    }
    // http.XXX("GET", "/aaa");
    unique_ptr<TcpServer> svr(new TcpServer(HandlerRequest, atoi(argv[1])));
    svr->Run();

    return 0;
}

Makefile文件：

server:main.cc
	g++ -o $@ $^ -std=c++11
.PHONY:clean
clean:
	rm -f server

2.2.epoll ET服务器源代码讲解

2.2.1.设计思路

epoll ET服务器：

在epoll ET服务器中，我们需要处理如下几种事件：

• 读事件：如果是监听套接字的读事件就绪则调用accept函数获取底层的连接，如果是其他套接字的读事件就绪则调用recv函数读取客户端发来的数据。
• 写事件：写事件就绪则将待发送的数据写入到发送缓冲区当中。
• 异常事件：当某个套接字的异常事件就绪时我们不做过多处理，直接关闭该套接字。
当epoll ET服务器监测到某一事件就绪后，就会将该事件交给对应的服务处理程序进行处理。

Reactor模式的五个角色：

在这个epoll ET服务器中，Reactor模式中的五个角色对应如下：

• 句柄：文件描述符。
• 同步事件分离器：I/O多路复用epoll。
• 事件处理器：包括读回调、写回调和异常回调。
• 具体事件处理器：读回调、写回调和异常回调的具体实现。
• 初始分发器：TcpServer类当中的Dispatcher函数。
Dispatcher函数要做的就是调用epoll_wait函数等待事件发生，当有事件发生后就将就绪的事件派发给对应的服务处理程序即可。

Connection类：

• 在Reactor的工作流程中说到，在注册事件处理器时需要将其与Handle关联，本质上就是需要将读回调、写回调和异常回调与某个文件描述符关联起来。
• 这样做的目的就是为了当某个文件描述符上的事件就绪时可以找到其对应的各种回调函数，进而执行对应的回调方法来处理该事件。
所以我们可以设计一个Connection类，该类当中的成员就包括一个文件描述符，以及该文件描述符对应的各种回调函数，此外还有一些其他成员，后面实现的时候再做详细论述。

TcpServer类：

• 在Reactor的工作流程中说到，当所有事件处理器注册完毕后，会使用同步事件分离器等待这些事件发生，当某个事件处理器的Handle变为Ready状态时，同步事件分离器会通知初始分发器，然后初始分发器会将Ready状态的Handle作为key来寻找其对应的事件处理器，并调用该事件处理器中对应的回调方法来响应该事件。
• 本质就是当事件注册完毕后，会调用epoll_wait函数来等待这些事件发生，当某个事件就绪时epoll_wait函数会告知调用方，然后调用方就根据就绪的文件描述符来找到其对应的各种回调函数，并调用对应的回调函数进行事件处理。
对此我们可以设计一个TcpServer类。

• 该类当中有一个成员函数叫做Dispatcher，这个函数就是所谓的初始分发器，在该函数内部会调用epoll_wait函数等待事件的发生，当事件发生后会告知Dispatcher已经就绪的事件。
• 当事件就绪后需要根据就绪的文件描述符来找到其对应的各种回调函数，由于我们会将每个文件描述符及其对应的各种回调都封装到一个Connection结构当中，所以实际我们就是需要根据文件描述符找到其对应的Connection结构。
• 我们可以使用C++ STL当中的unordered_map，来建立各个文件描述符与其对应的Connection结构之间的映射，这个unordered_map可以作为TcpServer类的一个成员变量，当需要找某个文件描述符的Connection结构时就可以通过该成员变量找到。
此外，在TcpServer类当中还有一些其他成员，后面实现的时候再做详细论述。

epoll ET服务器的工作流程：

• 这个epoll ET服务器在Reactor模式下的工作流程如下：

首先epoll ET服务器需要进行套接字的创建、绑定和监听。
• 然后定义一个TcpServer对象并初始化，初始化时要做的就是创建epoll模型。
• 紧接着需要为监听套接字创建对应的Connection结构，并调用TcpServer类中提供的AddConnection函数将监听套接字添加到epoll模型中，并建立监听套接字与其对应的Connection结构之间的映射关系。
• 之后就可以不断调用TcpServer类中的Dispatcher函数进行事件派发。
在事件处理过程中，会不断向Dispatcher当中新增或删除事件，而每个事件就绪时都会自动调用其对应的回调函数进行处理，所以我们要做的就是不断调用Dispatcher函数进行事件派发即可。

2.2.2.Connection结构

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

• 当某个文件描述符的读事件就绪时，我们会调用recv函数读取客户端发来的数据，但我们并不能保证我们读取到了一个完整的报文，因此需要将读取到的数据暂时存放到该文件描述符对应的inbuffer当中，当inbuffer当中可以分离出一个完整的报文后再将其分离出来进行数据处理，这里的inbuffer本质就是用来解决粘包问题的。
• 当处理完一个报文请求后，需要将响应数据发送给客户端，但我们并不能保证底层TCP的发送缓冲区中有足够的空间供我们写入，因此需要将要发送的数据暂时存放到该文件描述符对应的outbuffer当中，当底层TCP的发送缓冲区中有空间，即写事件就绪时，再依次发送outbuffer当中的数据。
• Connection结构当中设置回指指针R，便于快速找到我们定义的TcpServer对象，因为后续我们需要根据Connection结构找到这个TcpServer对象。比如当连接事件就绪时，需要调用TcpServer类当中的AddEvent函数将其添加到Dispatcher当中。
此外，Connection结构当中需要提供管理回调的成员函数，便于外部对Connection结构当中的各种回调进行设置。

2.2.3.TcpServer类

在TcpServer类当中有一个unordered_map成员，用于建立文件描述符和与其对应的Connection结构之间的映射，还有一个epfd成员，该成员是epoll模型对应的文件描述符。

在初始化TcpServer对象的时候就可以调用epoll_create函数创建epoll模型，并将该epoll模型对应的文件描述符用epfd成员记录下来，便于后续使用。
Reactor对象在析构的时候，需要调用close函数将该epoll模型进行关闭。

Dispatcher函数（事件分派器）：
TcpServer类当中的Dispatcher函数就是之前所说的初始分发器，这里我们更形象的将其称之为事件分派器。

• 事件分派器要做的就是调用epoll_wait函数等待事件发生。
• 当某个文件描述符上的事件发生后，先通过unordered_map找到该文件描述符对应的Connection结构，然后调用Connection结构当中对应的回调函数对该事件进行处理即可。

说明一下：

• 这里没有用switch或if语句对epoll_wait函数的返回值进行判断，而是借用for循环对其返回值进行了判断。
• 如果epoll_wait的返回值为-1则说明epoll_wait函数调用失败，此时不会进入到for循环内部进行事件处理。
• 如果epoll_wait的返回值为0则说明epoll_wait函数超时返回，此时也不会进入到for循环内部进行事件处理。
• 如果epoll_wait的返回值大于0则说明epoll_wait函数调用成功，此时才会进入到for循环内部调用对应的回调函数对事件进行处理。
• 事件处理时最好先对异常事件进行处理，因此代码中将异常事件的判断放在了最前面。

AddConnection函数：

TcpServer类当中的AddConnection函数是用于进行事件注册的。

• 在注册事件时需要传入一个文件描述符和一个事件集合，表示需要监视哪个文件描述符上的哪些事件。
• 还需要传入该文件描述符对应的Connection结构，表示当该文件描述符上的事件就绪后应该执行的回调方法。
• 在AddEvent函数内部要做的就是，调用epoll_ctl函数将该文件描述符及其对应的事件集合注册到epoll模型当中，然后建立该文件描述符与其对应的EventItem结构的映射关系。

EnableReadWrite函数：

Reactor类当中的EnableReadWrite函数，用于使能或使能某个文件描述符的读写事件。

• 调用EnableReadWrite函数时需要传入一个文件描述符，表示需要设置的是哪个文件描述符对应的事件。
• 还需要传入两个bool值，分别表示需要使能还是使能读写事件。
• EnableReadWrite函数内部会调用epoll_ctl函数修改将该文件描述符的监听事件。

2.2.4.回调函数

下面我们就可以实现一些回调函数，这里主要实现四个回调函数。

• Accepter：当连接事件到来时可以调用该回调函数获取底层建立好的连接。
• TcpRecver：当读事件就绪时可以调用该回调函数读取客户端发来的数据并进行处理。
• TcpSender：当写事件就绪时可以调用该回调函数向客户端发送响应数据。
• TcpExcepter：当异常事件就绪时可以调用该函数将对应的文件描述符进行关闭。
当我们为某个文件描述符创建Connection结构时，就可以调用Connection类提供的回调设置函数（SetRecver、SetSender、SetExcepter），将这些回调函数到Connection结构当中。

• 我们会将监听套接字对应的Connection结构当中的recver_设置为Accepter，因为监听套接字的读事件就绪就意味着连接事件就绪了，而监听套接字一般只关心读事件，因此监听套接字对应的sender_和excepter_可以设置为nullptr。
• 当Dispatcher监测到监听套接字的读事件就绪时，会调用监听套接字对应的Connection结构当中的recver_回调，此时就会调用Accepter回调获取底层建立好的连接。
• 而对于与客户端建立连接的套接字，我们会将其对应的Connection结构当中的recver_、sender_和excepter_分别设置为这里的TcpRecver、TcpSender和TcpExcepter。
• 当Dispatcher监测到这些套接字的事件就绪时，就会调用其对应的Connection结构当中对应的回调函数，也就是这里的TcpRecver、TcpSender和TcpExcepter。

Accepter回调：

Accepter回调用于处理连接事件，其工作流程如下：

1.调用accept函数获取底层建立好的连接。
2.将获取到的套接字设置为非阻塞，并为其创建Connection结构，填充Connection结构当中的各个字段，并注册该套接字相关的回调方法。
3.将该套接字及其对应需要关心的事件注册到Dispatcher当中。
下一次Dispatcher在进行事件派发时就会帮我们关注该套接字对应的事件，当事件就绪时就会执行该套接字对应的Connection结构中对应的回调方法。

需要注意的是，因为这里实现的ET模式下的epoll服务器，因此在获取底层连接时需要循环调用accept函数进行读取，并且监听套接字必须设置为非阻塞。

• 因为ET模式下只有当底层建立的连接从无到有或是从有到多时才会通知上层，如果没有一次性将底层建立好的连接全部获取，并且此后再也没有建立好的连接，那么底层没有读取完的连接就相当于丢失了，所以需要循环多次调用accept函数获取底层建立好的连接。
• 循环调用accept函数也就意味着，当底层连接全部被获取后再调用accept函数，此时就会因为底层已经没有连接了而被阻塞住，因此需要将监听套接字设置为非阻塞，这样当底层没有连接时accept就会返回，而不会被阻塞住。
accept获取到的新的套接字也需要设置为非阻塞，就是为了避免将来循环调用recv、send等函数时被阻塞。

设置文件描述符为非阻塞：

设置文件描述符为非阻塞时，需要先调用fcntl函数获取该文件描述符对应的文件状态标记，然后在该文件状态标记的基础上添加非阻塞标记O_NONBLOCK，最后调用fcntl函数对该文件描述符的状态标记进行设置即可。

监听套接字设置为非阻塞后，当底层连接不就绪时，accept函数会以出错的形式返回，因此当调用accept函数的返回值小于0时，需要继续判断错误码。

• 如果错误码为EAGAIN或EWOULDBLOCK，说明本次出错返回是因为底层已经没有可获取的连接了，此时底层连接全部获取完毕，这时我们可以返回0，表示本次Accepter调用成功。
• 如果错误码为EINTR，说明本次调用accept函数获取底层连接时被信号中断了，这时还应该继续调用accept函数进行获取。
• 除此之外，才说明accept函数是真正调用失败了，这时我们可以返回-1，表示本次Accepter调用失败。

问题：accept、recv和send等IO系统调用为什么会被信号中断？

答：IO系统调用函数出错返回并且将错误码设置为EINTR，表明本次在进行数据读取或数据写入之前被信号中断了，也就是说IO系统调用在陷入内核，但并没有返回用户态的时候内核跑去处理其他信号了。

• 在内核态返回用户态之前会检查信号的pending位图，也就是未决信号集，如果pending位图中有未处理的信号，那么内核就会对该信号进行处理。
• 但IO系统调用函数在进行IO操作之前就被信号中断了，这实际上是一个特例，因为IO过程分为“等”和“拷贝”两个步骤，而一般“等”的过程比较漫长，而在这个过程中我们的执行流其实是处于闲置的状态的，因此在“等”的过程中如果有信号产生，内核就会立即进行信号的处理。

写事件是按需打开的：

这里调用accept获取上来的套接字在添加到Dispatcher中时，只添加了EOPLLIN和EPOLLET事件，也就是说只让epoll帮我们关心该套接字的读事件。

• 这里之所以没有添加写事件，是因为当前我们并没有要发送的数据，因此没有必要让epoll帮我们关心写事件。
• 一般读事件是经常会被设置的，而写事件则是按序打开的，只有当我们有数据要发送时才会将写事件打开，并且在数据全部写入完毕后又会立即将写事件关闭。

注：

1.如果LT模式，一定是要先检测有没有对应的空间(先打开对写事件的关心，epoll会自动进行事件派发)，然后才写入。LT模式下，只要你打开了写入，我们的代码会自动进行调用TcpSender方法，进行发送。
2.如果是ET模式，也可以采用上面的方法。不过，一般ET我们追求高效，直接发送。通过发送是否全部发送完成，来决定是否要进行打开对写事件进行关心：

a.先调用send函数发送，发完就完了。
b.如果没有send发完，打开写事件关心，让epoll自动帮我们进行发送。
一般写事件关心，不能常打开,一定是需要的时候，在进行打开）不需要就要关闭对写事件的关心。

TcpRecver回调：

TcpRecver回调用于处理读事件，其工作流程如下：

1.循环调用recv函数读取数据，并将读取到的数据添加到该套接字对应EventItem结构的inbuffer当中。
2.对inbuffer当中的数据进行切割，将完整的报文切割出来，剩余的留在inbuffer当中。
3.对切割出来的完整报文进行反序列化。
4.业务处理。
5.业务处理后形成响应报文。
6.将响应报头添加到对应EventItem结构的outbuffer当中，并打开写事件。
下一次Dispatcher在进行事件派发时就会帮我们关注该套接字的写事件，当写事件就绪时就会执行该套接字对应的EventItem结构中写回调方法，进而将outbuffer中的响应数据发送给客户端。

数据读取：

• 循环调用recv函数将读取到的数据添加到inbuffer当中。

• 当recv函数的返回值小于0时同样需要进一步判断错误码，如果错误码为EAGAIN或EWOULDBLOCK则说明底层数据读取完毕了，如果错误码为EINTR则说明读取过程被信号中断了，此时还需要继续调用recv函数进行读取，否则就是读取出错了。
• 当读取出错时直接调用该套接字对应的excepter_回调，最终就会调用到下面将要实现的TcpExcepter回调，在我们会在TcpExcepter回调当中将该套接字进行关闭。

报文切割：

报文切割本质就是为了防止粘包问题，而粘包问题实际是涉及到协议定制的。

• 因为我们需要根据协议知道如何将各个报文进行分离，比如UDP分离报文采用的就是定长报头+自描述字段。
• 我们的目的是演示整个数据处理的过程，为了简单起见就不进行过于复杂的协议定制了，这里我们就以“X”作为各个报文之间的分隔符，每个报文的最后都会以一个“X”作为报文结束的标志。
• 因此现在要做的就是以“X”作为分隔符对inbuffer当中的字符串进行切割，这里将这个过程封装成一个PackageSplit函数。
• PackageSplit函数要做的就是对inbuffer当中的字符串进行切割，将切割出来的一个个报文放到vector当中，对于最后无法切出完整报文的数据就留在inbuffer当中即可。

反序列化：

在数据发送之前需要进行序列化Serialize，接收到数据之后需要对数据进行反序列化Parser。

• 序列化就是将对象的状态信息转换为可以存储或传输的形式（字节序列）的过程。
• 反序列化就是把字节序列恢复为原对象的过程。
实际反序列化也是与协议定制相关的，假设这里的epoll服务器向客户端提供的就是计算服务，客户端向服务器发来的都是需要服务器计算的计算表达式，因此可以用一个结构体来描述这样一个计算表达式，结构体当中包含两个操作数x和y，以及一个操作符op。

此时这里所谓的反序列化就是将一个计算表达式转换成这样一个结构体，

• 因此现在要做的就是将形如“1 + 2”这样的计算表达式转换成一个结构体，该结构体当中的x成员的值就是1，y的值就是2，op的值就是‘+’，这里将这个过程封装成一个Parser函数。

说明一下： 实际在做项目时不需要我们自己进行序列化和反序列化，我们一般会直接用JSON或XML这样的序列化反序列化工具。

业务处理：

业务处理就是服务器拿到客户端发来的数据后，对数据进行数据分析，最终拿到客户端想要的资源。

• 我们这里要做的业务处理非常简单，就是用反序列化后的数据进行数据计算，此时得到的计算结果就是客户端想要的。

形成响应报文：

在业务处理后我们已经拿到了客户端想要的数据，现在我们要做的就是形成响应报文，由于我们这里规定每个报文都以“X”作为报文结束的标志，因此在形成响应报文的时候，就需要在每一个计算结果后面都添加上一个“X”，表示这是之前某一个请求报文的响应报文，因为协议制定后就需要双方遵守。

将响应报文添加到outbuffer中：

响应报文构建完后需要将其添加到该套接字对应的outbuffer中，并打开该套接字的写事件，此后当写事件就绪时就会将outbuffer当中的数据发送出去。

TcpSender回调：

TcpSender回调用于处理写事件，其工作流程如下：

1.循环调用send函数发送数据，并将发送出去的数据从该套接字对应Connection结构的outbuffer中删除。
2.如果循环调用send函数后该套接字对应的outbuffer当中的数据被全部发送，此时就需要将该套接字对应的写事件关闭，因为已经没有要发送的数据了，如果outbuffer当中的数据还有剩余，那么该套接字对应的写事件就应该继续打开。

循环调用send函数发送数据的过程：

• 循环调用send函数将outbuffer中的数据发送出去。
• 当send函数的返回值小于0时也需要进一步判断错误码，如果错误码为EAGAIN或EWOULDBLOCK则说明底层TCP发送缓冲区已经被写满了，这时需要将已经发送的数据从outbuffer中移除。
• 如果错误码为EINTR则说明发送过程被信号中断了，此时还需要继续调用send函数进行发送，否则就是发送出错了。
• 当发送出错时也直接调用该套接字对应的excepter_回调，最终就会调用到下面将要实现的TcpExcepter回调，在我们会在TcpExcepter回调当中将该套接字进行关闭。
• 如果最终outbuffer当中的数据全部发送成功，则将outbuffer清空即可。

TcpExcepter回调：

errorer回调用于处理异常事件。

• 对于异常事件就绪的套接字我们这里不做其他过多的处理，简单的调用close函数将该套接字关闭即可。
• 但是在关闭该套接字之前，需要先调用DelEvent函数将该套接字从epoll模型中删除，并取消该套接字与其对应的Connection结构的映射关系。
• 由于在Dispatcher当中是先处理的异常事件，为了避免该套接字被关闭后继续进行读写操作，然后因为读写操作失败再次调用errorer回调重复关闭该文件描述符，因此在关闭该套接字后将其Connection当中的文件描述符值设置为-1。
• 在调用TcpRecver和TcpSender回调执行读写操作之前，都会判断该Connection结构当中的文件描述符值是否有效，如果无效则不会进行后续操作。

2.2.5.套接字相关

这里可以编写一个Socket类，对套接字相关的接口进行一定程度的封装，为了让外部能够直接调用Socket类当中封装的函数，于是将这些函数定义成了静态成员函数。

2.2.6.Reactor模式和Proactor模式

• Reactor：半异步半同步，Reactor (tcp）服务器是Linux服务最常用的（几乎没有之一），既负责事件派发，又负责IO（负责业务逻辑的处理）
• Proactor：前摄模式，其他平台可能出现的模式。只负责负责事件派发，就绪的事件推送给后端的进程、线程池，不关心负责IO（不负责业务逻辑的处理）文章来源地址https://www.toymoban.com/news/detail-481182.html

到了这里，关于Linux - 第25节 - Linux高级IO（三）的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！