上篇文章：C/C++ Linux网络编程7 - epoll解决客户端并发连接问题-CSDN博客

代码仓库：橘子真甜 (yzc-YZC) - Gitee.com

目录

一. Reactor模式说明

1.1 reactor简介

1.2 ET 与 LT 模式说明

二. reactro服务器实现

1.1 socket.hpp

1.2 connItem.hpp⭐

1.3 tcpServer.hpp

a setEvent

b addConnlist

c removeConn

d init

e dispather事件派发器 

f accepter/recver/sender

1.4 tcpServer.cc和简单任务代码

三. 测试与总结

3.1 测试

3.2 总结

一. Reactor模式说明

1.1 reactor简介

        上篇文章中，我们使用epoll实现了一个tcp服务器，得益于epoll的高效IO多路复用，服务器的并发管理和QPS都是比较好的。

        但是我们的代码仍有部分问题

1 由于我们的面向fd的，随着不同的事件增多，拓展比较麻烦

2 网络IO并没有和数据处理解耦（上篇文章通过回调函数进行简单解耦）

3 读写数据会发生阻塞，即便想使用多线程进行非阻塞处理也难以拓展

        而reactor是面向事件触发的，即判断就绪事件的是EPOLLIN/EPOLLOUT/EPOLLERR等执行提前注册好fd的回调函数。由于我们提前注册好了回调函数，无论是添加新功能还是拓展多线程都比较方便。

1.2 ET 与 LT 模式说明

        ET（边缘触发）和LT（水平触发）是epoll的两种wait触发方式。

        epoll默认的方式是 LT 水平触发方式，这种方式epoll_wait只要发现关心fd的内核中有数据可读可写就会唤醒通过就绪队列返回。如果用户层不去或者处理数据较慢，epollwait会不断的唤醒直到数据被处理完毕。

        而ET 边缘触发方式需要用户关心fd时候，epollctl时候加上 EPOLLET标明这个fd关心的方式的ET模式，ET模式是 内核数据发送变化的时候才去通知用户层读取数据，然后等到下一次数据变化时候再去通知用户层。

        内核数据变化的情况：数据从无到有，从不可读到可读，数据增多等

        乍一看，貌似ET模式是不是没太必要？ET模式下如果用户层不去处理数据，内核也没有新数据到来，这些数据不就被遗忘了吗？的确是这样的，所以ET模式一般要求我们 设置fd为非阻塞，然后循环一次性读取全部内核数据。

        为何要设置为非阻塞并一次性读取完毕？如果阻塞读取，内核一共10字节数据，ET通知我之后不通知了。用户层一次性只能读5个数据，读后5个字节的时候由于是阻塞读只有ET唤醒才会读，但是此时由于内核数据不再变化，永远不会唤醒，而这个read会一直阻塞

        ET模式通过强制用户层一次性读取 内核全部数据，这样epollwait就可以一次唤醒就能处理所有数据了。

        如果使用LT模式，epollwait可能会多次唤醒，浪费性能。特别是数据比较多的时候，LT模式会有大量的无效唤醒。

        因此，epoll服务器推荐使用ET模式减少epollwait唤醒从而提供服务器的性能

二. reactro服务器实现

1.1 socket.hpp

        由于解耦的好处，我们能够直接使用之前的代码

#pragma once
#include 
#include 
#include 
#include 
#include 
#include 

#include 

const int gbacklog = 5;
class mySocket
{
public:
    // 1.构建tcp socketfd
    static int creatSockfd()
    {
        // 创建socketfd
        int sockfd = socket(AF_INET, SOCK_STREAM, 0);
        assert(sockfd > 0);

        // 设置端口复用
        int opt = 1;
        setsockopt(sockfd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt));
        return sockfd;
    }

    // 2.bind绑定端口
    static void Bind(int sockfd, int port)
    {
        struct sockaddr_in serveraddr;
        memset(&serveraddr, 0, sizeof(serveraddr));
        // 设置地址的信息（协议，ip，端口）
        serveraddr.sin_family = AF_INET;
        serveraddr.sin_addr.s_addr = htonl(INADDR_ANY); // 绑定任意网卡ip，通常我们访问某一个IP地址是这个服务器的公网网卡IP地址
        serveraddr.sin_port = htons(port);              // 注意端口16位，2字节需要使用htons。不可使用htonl
        if (bind(sockfd, (const sockaddr *)(&serveraddr), sizeof(serveraddr)) < 0)
        {
            perror("sock bind err");
            exit(-1);
        }
        std::cout << "sock bind success" << std::endl;
    }

    // 3. listen监听，让打开的sock这个"文件"去监听来自网络的请求。用于获取新的网络连接
    static void Listen(int sockfd, int n)
    {
        if (listen(sockfd, n) == -1)
        {
            perror("sock listen err");
            exit(-1);
        }
        std::cout << "sock listen success" << std::endl;
    }

    // 4 accept创建sockfd用于传输数据
    static int Accept(int listenfd, std::string &clientIp, uint16_t &clientPort)
    {
        // 获取新fd用于通信
        struct sockaddr_in clientaddr;
        memset(&clientaddr, 0, sizeof(clientaddr));

        socklen_t len = sizeof(clientaddr);
        int sockfd = accept(listenfd, (struct sockaddr *)&clientaddr, &len);

        
        // 成功了，可以获取对方的ip和端口
        clientIp = inet_ntoa(clientaddr.sin_addr);
        clientPort = ntohs(clientaddr.sin_port);
        return sockfd;
    }
};

1.2 connItem.hpp⭐

        这部分代码有什么作用？reactor模式是基于事件触发的，要求事件触发后调用就绪fd对应的回调函数，既然如此，我咋知道我调用的回调函数是哪一个呢？

        所有需要一个结构体来管理 fd和对应的回调函数，以及接收发送数据的用户缓冲区。

#pragma once
#include 

// 回调函数的函数指针
struct connItem;
using func_t = std::function;
struct connItem
{
    // 构造函数
    connItem(int sockfd = -1)
        : _sockfd(sockfd), _reader(nullptr), _sender(nullptr), _execpter(nullptr) {}

    // 注册回调函数
    void connRegister(func_t reader = nullptr, func_t sender = nullptr, func_t execpter = nullptr)
    {
        _reader = reader;
        _sender = sender;
        _execpter = execpter;
    }

    int _sockfd;
    func_t _reader;
    func_t _sender;
    func_t _execpter;

    std::string _inbuffer;
    std::string _outbuffer;
};

1.3 tcpServer.hpp

        由于新增了connItem结构体，还有使用reactor + ET模式。所以对于之前的epoll服务器来说，新代码需要更改的比较多

代码框架如下：

#include 
#include 
#include 
#include 
#include 
#include 
#include 

#include 
#include 

#include "socket.hpp"
#include "connItem.hpp"

const int fdnums = 100000;

// 将一个函数设置为非阻塞
bool SetNonBlock(int sockfd)
{
    if (sockfd < 0)
        return false;

    int flags = fcntl(sockfd, F_GETFL, 0);
    if (flags < 0)
        return false;

    int n = fcntl(sockfd, F_SETFL, flags | O_NONBLOCK);
    return n > 0;
}


namespace YZC
{
    // 设置默认端口和最大backlog
    const int defaultPort = 8080;
    const int maxBacklog = 128;

    class tcpServer
    {
    public:
        tcpServer(int port = defaultPort)
            : _port(port), _epfd(-1) {}

        void init()
        {
            // epoll_create创建epollfd,初始化返回epoll_event数组
            _epfd = epoll_create(1);
            if (_epfd < 0)
            {
                printf("epoll_create err
");
                exit(errno);
            }
            _events = new struct epoll_event[fdnums];

            _listensock = mySocket::creatSockfd();
            mySocket::Bind(_listensock, _port);
            mySocket::Listen(_listensock, maxBacklog);

            // epollctl关心 listensock,注意要设置好event
            struct epoll_event ev;
            ev.data.fd = _listensock;
            ev.events = EPOLLIN | EPOLLET;
            epoll_ctl(_epfd, EPOLL_CTL_ADD, _listensock, &ev);
        }

        // 事件派发器
        void Dispather()
        {
        }

    private:
        void accepter(connItem *conn)
        {
        }

        void sender(connItem *conn)
        {
        }

        void recver(connItem *conn)
        {
        }

    private:
        // accpet一个连接后，epoll关心关心该事件以及注册对应的读写异常方法
        void addConnList(int sockfd, uint32_t event, func_t recver, func_t sender, func_t ececpter)
        {
        }

        // 添加/修改/删除一个sockfd epoll关心事件属性
        void setEvent(int sockfd, uint32_t event, int flag)
        {
        }

        // 连接关闭后，释放对应的资源
        void removeConn(int sockfd)
        {
        }

    private:
        int _listensock; // 监听sock
        int _port;       // 端口port

        int _epfd;                                       // epoll fd
        struct epoll_event *_events;                     // epoll返回数组
        std::unordered_map _connlist{}; // 管理fd-conn的哈希表

        func_t _service; // 该服务器的业务处理函数
    };
}

a setEvent

        用于设置epoll关心fd的状态，关心/删除/修改。首先我们要定义一下关心的方式

类外定义好枚举类型

enum
{
    EVENT_ADD,
    EVENT_DEL,
    EVENT_MOD
};

根据传入的flag执行对应的处理

        // 添加/修改/删除一个sockfd epoll关心事件属性
        void setEvent(int sockfd, uint32_t event, int flag)
        {
            epoll_event ev;
            ev.data.fd = sockfd;
            ev.events = event;

            if (flag == EVENT_ADD)
                epoll_ctl(_epfd, EPOLL_CTL_ADD, sockfd, &ev);
            else if (flag == EVENT_MOD)
                epoll_ctl(_epfd, EPOLL_CTL_MOD, sockfd, &ev);
            else if (flag == EVENT_DEL)
                epoll_ctl(_epfd, EPOLL_CTL_DEL, sockfd, nullptr);
        }

b addConnlist

        用于accpet后，关心这个fd和管理fd-conn

        // accpet一个连接后，epoll关心关心该事件以及注册对应的读写异常方法
        void addConnList(int sockfd, uint32_t event, func_t recver, func_t sender, func_t execpter)
        {
            // ET模式，设置fd为非阻塞
            SetNonBlock(sockfd);

            // epoll关心该fd
            setEvent(sockfd, event, EVENT_ADD);

            // 构建conn，注册好回调方法
            connItem *conn = new connItem(sockfd);
            conn->connRegister(recver, sender, execpter);

            // 插入哈希表管理fd-conn,使用下标是因为用户层确保fd不会覆盖，连接断开会close fd 和移除哈希表
            _connlist[sockfd] = conn;
        }

c removeConn

        // 连接关闭后，释放对应的资源
        void removeConn(int sockfd)
        {
            auto it = _connlist.find(sockfd);
            if (it != _connlist.end())
            {
                // 1.epoll移除fd
                setEvent(sockfd, 0, EVENT_DEL);

                // 2.删除sock对应conn
                delete it->second;
                it->second = nullptr;

                // 3.移除哈希表中的映射关系,删除迭代器的效率是O(1)。如果使用key需要再次查找
                _connlist.erase(it);

                // 4.close fd
                close(sockfd);
            }
        }

d init

        由于使用了conn管理，需要重新更新init初始化。在这里，我们还能使用 setsockopt来设置端口复用，这样即便服务端崩掉也能快速重新启动bind。

直接修改socket.hpp中的代码即可

    // 1.构建tcp socketfd
    static int creatSockfd()
    {
        // 创建socketfd
        int sockfd = socket(AF_INET, SOCK_STREAM, 0);
        assert(sockfd > 0);

        // 设置端口复用
        int opt = 1;
        setsockopt(sockfd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt));
        return sockfd;
    }

然后进行初始化

        void init()
        {
            // epoll_create创建epollfd,初始化返回epoll_event数组
            _epfd = epoll_create(1);
            if (_epfd < 0)
            {
                printf("epoll_create err
");
                exit(errno);
            }
            _events = new struct epoll_event[fdnums];

            _listensock = mySocket::creatSockfd();
            mySocket::Bind(_listensock, _port);
            mySocket::Listen(_listensock, maxBacklog);

            // epollctl关心 listensock,注意要设置好event
            // struct epoll_event ev;
            // ev.data.fd = _listensock;
            // ev.events = EPOLLIN | EPOLLET;
            // epoll_ctl(_epfd, EPOLL_CTL_ADD, _listensock, &ev);

            // 注册listensock对应读写异常函数
            addConnList(_listensock, EPOLLIN | EPOLLET, [this](connItem *conn)
                        { this->accepter(conn); }, nullptr, nullptr);
        }

e dispather事件派发器 

      该函数是Reactor是中心，在这里循环处理epoll_wait返回的fd，并通过回调函数执行提前注册好的任务。

        // 事件派发器
        void Dispather()
        {
            while (true)
            {
                // 1. epoll_wait返回就绪事件，高性能服务器一般设置为阻塞等待
                int n = epoll_wait(_epfd, _events, fdnums, -1);

                for (int i = 0; i < n; i++)
                {
                    int sockfd = _events[i].data.fd;
                    uint32_t event = _events[i].events;

                    // 执行fd对应回调函数
                    if (_connlist.count(sockfd))
                    {
                        // 执行读事件
                        if ((event & EPOLLIN) && _connlist[sockfd]->_reader != nullptr)
                            _connlist[sockfd]->_reader(_connlist[sockfd]);

                        // 执行写事件
                        if ((event & EPOLLOUT) && _connlist[sockfd]->_reader != nullptr)
                            _connlist[sockfd]->_reader(_connlist[sockfd]);

                        // 暂时不考虑异常
                    }
                }
            }
        }

f accepter/recver/sender

        这三个函数是fd对应的回调函数，分别是获取新连接，接收远端数据，发送数据到远端。

首先来处理accepter。

注意我们使用的ET模式，需要循环非阻塞读取数据。

所以遇到异常 EAGAIN或者EWOULDBLOCK表示这次数据读取/发送完毕，需要break。

遇到EINTER表示读取数据中断，需要continue重新读写数据

        void accepter(connItem *conn)
        {
            while (true)
            {
                std::string clientip;
                uint16_t clientport;
                int clientsockfd = mySocket::Accept(conn->_sockfd, clientip, clientport);
                if (clientsockfd > 0)
                {
                    // 关心这个新的的fd
                    addConnList(clientsockfd, EPOLLIN | EPOLLET, [this](connItem *conn)
                                { this->recver(conn); }, [this](connItem *conn)
                                { this->sender(conn); }, nullptr);

                    printf("Get a new link, info [%s:%d] clientsock[%d]
", clientip.data(), clientport, clientsock);
                }
                else
                {
                    // 注册该fd对应的回调函数
                    if (errno == EAGAIN || errno == EWOULDBLOCK)
                    {
                        // 标明读取完毕，需要退出
                        break;
                    }
                    else if (errno == EINTR)
                    {
                        continue;
                    }
                    else
                    {
                        // 对端关闭或者异常了
                        removeConn(conn->_sockfd);
                    }
                }
            }
        }

然后是recver。同理需要处理异常EAGAIN和EINTER。

不过recver接收数据之后，需要将数据传递给其他模块进行数据处理

        void recver(connItem *conn)
        {
            char buffer[1024];
            while (true)
            {
                int count = recv(conn->_sockfd, buffer, sizeof(buffer) - 1, 0);
                if (count > 0)
                {
                    buffer[count] = 0;
                    conn->_inbuffer += buffer;
                }
                else if (count == 0)
                {
                    removeConn(conn->_sockfd);
                    return;
                }
                else
                {
                    // 同理需要处理EAGAIN和EINTER
                    if (errno == EAGAIN || errno == EWOULDBLOCK)
                    {
                        // 无数据可读
                        break;
                    }
                    else if (errno == EINTR)
                    {
                        continue;
                    }
                    else
                    {
                        //  关闭套接字和取消epoll关心，然后退出
                        removeConn(conn->_sockfd);
                        return;
                    }
                }
            }

            // 接收数据之后，进行解析处理。这里就是业务的代码了，一般交给其他模块处理
            _service(conn);
        }

最后处理sender，用于发送协议处理后的数据

        void sender(connItem *conn)
        {
            while (true)
            {
                int count = send(conn->_sockfd, conn->_outbuffer.c_str(), conn->_outbuffer.size(), 0);

                if (count > 0)
                {
                    // 清空发送的数据，可以进一步优化
                    conn->_outbuffer.erase(0, count);
                    // 数据发送完毕
                    if (conn->_outbuffer.empty())
                    {
                        // 此时不可以直接更改事件的关系，因为数据可能还在内核，没有发送到网络
                        break;
                    }
                }
                else if (count == 0)
                {
                    // 没有数据发送
                    removeConn(conn->_sockfd);
                    return;
                }
                else
                {
                    // 同理需要处理EAGAIN和EINTER
                    if (errno == EAGAIN || errno == EWOULDBLOCK)
                        break;
                    else if (errno == EINTR)
                        continue;
                    else
                    {
                        // 关闭套接字和取消epoll关心，然后退出
                        removeConn(conn->_sockfd);
                        return;
                    }
                }
            }

            // 到这里，事件结束了，数据才是真的发送出去了通重新关心该事件的读写
            if (conn->_outbuffer.empty())
                setEvent(conn->_sockfd, EPOLLIN | EPOLLET, EVENT_MOD); // 无数据可写
            else
                setEvent(conn->_sockfd, EPOLLIN | EPOLLOUT | EPOLLET, EVENT_MOD); // 有数据可写
        }

1.4 tcpServer.cc和简单任务代码

任务代码

void serviceIO(connItem *conn)
{
    printf("client --> server:%s
", conn->_inbuffer.c_str());
    // 尝试直接发送数据
    conn->_outbuffer = conn->_inbuffer;
    conn->_inbuffer.clear();
    conn->_sender(conn);
}

void serviceHTTP(connItem *conn)
{
    conn->_inbuffer.clear();
    // 这里仅做简单的数据收发
    std::string outbuffer;
    std::string body = "
hello world
";
    outbuffer =
        "HTTP/1.1 200 OK
"
        "Content-Type: text/html; charset=utf-8
"
        "Content-Length: " +
        std::to_string(body.size()) + "
"
                                      "Server: Apache/2.4.41
"
                                      "Date: Mon, 18 Dec 2023 08:32:10 GMT
"
                                      "X-Frame-Options: DENY
"
                                      "X-Content-Type-Options: nosniff
"
                                      "Referrer-Policy: strict-origin-when-cross-origin
"
                                      "
" // 空行分隔头部和正文
        + body;

    // 无脑向客户端发送一个简单http响应
    conn->_outbuffer = outbuffer;
    conn->_sender(conn);
}

tcpserver.cc

#include "tcpServer.hpp"
#include 
#include 
using namespace std;

// tcp 服务器，启动方式与udp server一样
//./tcpServer + local_port    //我们将本主机的所有ip与端口绑定

static void Usage(string proc)
{
    cout << "
Usage:
	" << proc << " lock_port

";
}

int main(int argc, char *argv[])
{
    if (argc != 2)
    {
        Usage(argv[0]);
    }
    uint16_t serverport = atoi(argv[1]);

    unique_ptr tsvr(new YZC::tcpServer(serviceHTTP, serverport));
    tsvr->init();
    tsvr->Dispather();
    return 0;
}

三. 测试与总结

3.1 测试

        测试通信结果如下，可以看到能够正常处理多个客户端的并发请求。

同样使用wrk来测试一下我们服务器的QPS，对比普通epoll是否有提升？？

服务器配置：2G2核

上篇文章的测试结果如下：

下面是wrk测试reactor epoll ET模式服务器的结果

[root@study wrk]# ./wrk -c 1000 -d 10s -t 10 http://47.105.37.157:8080
Running 10s test @ http://47.105.37.157:8080
  10 threads and 1000 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency   140.64ms  231.11ms   2.00s    89.14%
    Req/Sec     1.67k   848.73     5.38k    85.44%
  164840 requests in 10.07s, 41.66MB read
  Socket errors: connect 0, read 0, write 0, timeout 140
Requests/sec:  16371.18
Transfer/sec:      4.14MB

[root@study wrk]# ./wrk -c 10000 -d 10s -t 10 http://47.105.37.157:8080
Running 10s test @ http://47.105.37.157:8080
  10 threads and 10000 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency   124.95ms  281.36ms   2.00s    89.04%
    Req/Sec     1.45k     1.16k   11.24k    65.67%
  140981 requests in 10.10s, 35.63MB read
  Socket errors: connect 0, read 0, write 0, timeout 1536
Requests/sec:  13963.52
Transfer/sec:      3.53MB

[root@study wrk]# ./wrk -c 25000 -d 10s -t 50 http://47.105.37.157:8080
Running 10s test @ http://47.105.37.157:8080
  50 threads and 25000 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency   318.82ms  352.69ms   1.99s    87.76%
    Req/Sec   726.34      2.24k   28.55k    95.80%
  77320 requests in 42.42s, 19.54MB read
  Socket errors: connect 0, read 4, write 0, timeout 2591
Requests/sec:   1822.90
Transfer/sec:    471.75KB

[root@study wrk]# ./wrk -c 55555 -d 10s -t 100 http://47.105.37.157:8080
Running 10s test @ http://47.105.37.157:8080
  100 threads and 55555 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency   755.27ms  511.24ms   2.00s    63.02%
    Req/Sec   305.20    675.18    13.34k    92.80%
  349055 requests in 1.28m, 88.21MB read
  Socket errors: connect 8253, read 4414, write 20963, timeout 71586
Requests/sec:   4553.60
Transfer/sec:      1.15MB

可以看到，epoll + ET + reactor模式更适合处理高压/复杂数据的情况。并且通过代码就能看到，reactor模式对于代码的维护和拓展都支持的比较好

最后的对比表格如下：

3.2 总结

        当代Linux服务器的大多选择epoll ET + Reactor 模式，原因是更够高效并发处理连接，同时代码易拓展和维护。

        有没有更好的选择或者优化？

优化：可以引用多线程，主从reactor，线程池等机制。这样就能减轻单一reactor的压力。或者使用协程来对任务处理

更好的选择：异步IO的 io_uring，完全由协程替代的网络模型