【项目学习】C++实现高并发服务器——代码学习（一）Reactor高并发模型

项目来源：WebServer

上一篇：环境搭建
本文介绍以下功能的代码实现

• 利用IO复用技术Epoll与线程池实现多线程的Reactor高并发模型；

一、IO复用技术

IO多路复用使得程序能同时监听多个文件描述符，能够提高程序的性能，Linux下实现IO多路复用的系统调用主要有select. poll和epoll。

I/O 多路复用

1、epoll函数族

epoll_creat： 该函数生成一个epoll专用的文件描述符

#include 
int epoll_creae(int size);  //epoll上能关注的最大描述符数

参数：

• size : 必须大于0

返回值：

• -1:失败
• > 0∶文件描述符， *** 作epoll实例的

epoll_ctl：用于控制某个epoll文件描述符事件，可以注册、修改、删除

#include 
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

参数：

• epfd : epoll实例对应的文件描述符
• op :要进行什么 *** 作 EPOLL_CTL_ADD:添加 EPOLL_CTL_MOD:修改 EPOLL_CTL_DEL:删除
• fd :要检测的文件描述符
• event :检测文件描述符什么事情

struct epoll_event {
uint32_t	events; / * Epoll events * /
epoll_data_t data; / t user data variable * /
} ;

typedef union epoll_data {
void 	fptr;
int		fd;
uint32_t u32;
uint64_t u64;
}epoll_data_t;

常见的Epoll检测事件:

• EPOLLIN
• EPOLLOUT
• EPOLLERR

epoll_wait：等待IO事件发生 - 可以设置阻塞的函数

#include 
int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);

• efds：epoll_create函数的返回值
• events：传出参数【数组】满足监听条件的哪些fd结构体
• maxevents：数组元素的总个数(1024) struct epoll_events [1024]:
• timeout :阻塞时间
  • 0:不阻塞
  • -1 :阻塞，直到检测到fd数掘发生变化，解除阻塞->0:阻塞的时长（毫秒)

2、LT和ET模式

epoll对文件描述符的 *** 作方式有两种工作模式：LT模式(Level Trigger，水平触发) 和ET模式(Edge Trigger，边缘触发)。

• LT模式：当epoll_wait检测到其上有事件发生并将此事件通知应用程序后，应用程序可以不立即处理该事件，这样，当应用程序下一次调用epoll_wait时，epoll_wait还会向应用程序通告此事件，直到该事件被处理。
  

  
  

  (（缓冲区剩余未读尽的数据会导致epoll_wait返回.
  ）

  • a.用户不读数据，数据一直在缓冲区，epoll会一直通知
  • b.用户只读了一部分数据，epoll会通知
  • c.缓冲区的数据读完了，不通知
• ET模式：当epoll_wait检测到其上有事件发生并将此事件通知应用程序后，应用程序必须立即处理该事件，因为后续的epoll_wait调用将不在向应用程序通告此事件。
  

  
  

  • a.用户不读数据，数据一致在缓冲区中，epoll下次检测的时候就不通知了
  • b.用户只读了一部分数据，epoll不通知
  • c.缓冲区的数据读完了，不通知

3、代码实现

定义Epoller类

class Epoller {
public:
    explicit Epoller(int maxEvent = 1024);

    ~Epoller();

    //使用epoll_ctl取添加add，修改mod，删除del
    bool AddFd(int fd, uint32_t events);

    bool ModFd(int fd, uint32_t events);

    bool DelFd(int fd);
    
    int Wait(int timeoutMs = -1);

    int GetEventFd(size_t i) const;

    uint32_t GetEvents(size_t i) const;
        
private:
    int epollFd_;//epoll_create()创建一个epoll对象，返回值为epollFd_

    std::vector<struct epoll_event> events_;//检测到的事件集合 
};

#endif //EPOLLER_H

AddFd()函数

bool Epoller::AddFd(int fd, uint32_t events) {//文件描述符，事件
    if(fd < 0) return false;
    epoll_event ev = {0};
    ev.data.fd = fd;
    ev.events = events;
    return 0 == epoll_ctl(epollFd_, EPOLL_CTL_ADD, fd, &ev);
}

对文件描述符进行添加（ EPOLL_CTL_ADD） *** 作。

ModFd和DelFd同理

Wait()函数

int Epoller::Wait(int timeoutMs) {
    return epoll_wait(epollFd_, &events_[0], static_cast<int>(events_.size()), timeoutMs);//文件描述符 地址 大小 超时时间
}

调用epoll_wait实现

二、线程池 1、线程池原理

线程池是由服务器预先创建的一组子线程，线程池中的线程数量应该和CPU数量差不多。

线程池中的所有子线程都运行着相同的代码。

当有新的任务到来时，主线程将通过某种方式选择线程池中的某一个子线程来为之服务。

相比与动态的创建子线程，选择一个已经存在的子线程的代价显然要小得多。

至于主线程选择哪个子线程来为新任务服务，则有多种方式:

• 主线程使用某种算法来主动选择子线程。
  

  
  

  最简单、最常用的算法是随机算法和Round Robin(轮流选取）算法，但更优秀、更智能的算法将使任务在各个工作线程中更均匀地分配，从而减轻服务器的整体压力。
  

  
  

• 主线程和所有子线程通过一个共享的工作队列来同步，子线程都睡眠在该工作队列上。
  

  
  

  当有新的任务到来时，主线程将任务添加到工作队列中。
  

  
  

  这将唤醒正在等待任务的子线程，不过只有一个子线程将获得新任务的"接管权""，它可以从工作队列中取出任务并执行之，而其他子线程将继续睡眠在工作队列上。
  

  
  

线程池的一般模型为:

线程池中的线程数量最直接的限制因素是中央处理器(CPU)的处理器(processors/cores)的数量N:如果你的CPU是4-cores的，对于CPU密集型的任务(如视频剪辑等消耗CPU计算资源的任务)来说，那线程池中的线程数量最好也设置为4(或者+1防止其他因素造成的线程阻塞）﹔对于I0密集型的任务，一般要多于CPU的核数，因为线程间竞争的不是CPU的计算资源而是IO，lO的处理一般较慢，多于cores数的线程将为CPU争取更多的任务，不至在线程处理IO的过程造成CPU空闲导致资源浪费。

• 空间换时间，浪费服务器的硬件资源，换取运行效率。
  

  
  

• 池是一组资源的集合，这组资源在服务器启动之初就被完全创建好并初始化，这称为静态资源。
  

  
  

• 当服务器进入正式运行阶段，开始处理客户请求的时候，如果它需要相关的资源，可以直接从池中获取，无需动态分配。
  

  
  

• 当服务器处理完一个客户连接后，可以把相关的资源放回池中，无需执行系统调用释放资源。
  

  
  

2、代码实现

class ThreadPool {
public:
    explicit ThreadPool(size_t threadCount = 8): pool_(std::make_shared<Pool>()) {
            assert(threadCount > 0);
            //assert 宏的原型定义在 assert.h 中，其作用是如果它的条件返回错误，则终止程序执行。


            //创建threadCount个子线程
            for(size_t i = 0; i < threadCount; i++) {
                std::thread([pool = pool_] {
                    std::unique_lock<std::mutex> locker(pool->mtx);
                    while(true) {
                        //判断任务队列不为空
                        if(!pool->tasks.empty()) {
                            //从任务队列里取第一个任务
                            auto task = std::move(pool->tasks.front());
                            pool->tasks.pop();
                            //加锁，解锁
                            locker.unlock();
                            //任务执行的代码
                            task();
                            locker.lock();
                        } 
                        //判断池子是否关闭
                        else if(pool->isClosed) break;
                        else pool->cond.wait(locker);//阻塞
                    }
                }).detach();//线程分离
            }
    }

    ThreadPool() = default;

    ThreadPool(ThreadPool&&) = default;
    
    ~ThreadPool() {
        if(static_cast<bool>(pool_)) {
            {
                std::lock_guard<std::mutex> locker(pool_->mtx);
                pool_->isClosed = true;
            }
            pool_->cond.notify_all();
        }
    }

    template<class F>
    void AddTask(F&& task) {//从池子里添加一个任务
        {
            std::lock_guard<std::mutex> locker(pool_->mtx);
            pool_->tasks.emplace(std::forward<F>(task));
        }
        pool_->cond.notify_one();//唤醒一个线程
    }

private:
//定义了一个池子里面的信息，结构体
    struct Pool {
        std::mutex mtx;//互斥锁
        std::condition_variable cond;//条件变量
        bool isClosed;//是否关闭
        std::queue<std::function<void()>> tasks;//队列，保存任务
    };
    std::shared_ptr<Pool> pool_;//线程池
};

三、Reactor模式

服务器程序通常需要处理三类事件:I/O事件、信号及定时事件。

有两种高效的事件处理模式: Reactor和Proactor，同步I/O模型通常用于实现Reactor模式，异步I/O模型通常用于实现 Proactor模式。

Reactor模式

要求主线程（I/O处理单元）只负责监听文件描述符上是否有事件发生，有的话就立即将该事件通知工作线程〈逻辑单元)，将socket 可读可写事件放入请求队列，交给工作线程处理。

除此之外，主线程不做任何其他实质性的工作。

读写数据，接受新的连接，以及处理客户请求均在工作线程中完成。

使用同步I/O (以epoll_wait为例）实现的Reactor模式的工作流程是:

1. 主线程往epoll内核事件表中注册socket 上的读就绪事件。
   

   
   

2. 主线程调用epoll_wait 等待socket上有数据可读。
   

   
   

3. 当socket上有数据可读时，epoll_wait通知主线程。
   

   
   

   主线程则将socket可读事件放入请求队列。
   

   
   

4. 睡眠在请求队列上的某个工作线程被唤醒，它从socket 读取数据，并处理客户请求，然后往epoll 内核事件表中注册该socket 上的写就绪事件。
   

   
   

5. 当主线程调用epoll_wait等待socket可写。
   

   
   

6. 当socket可写时，epoll_wait通知主线程。
   

   
   

   主线程将socket可写事件放入请求队列。
   

   
   

7. 睡眠在请求队列上的某个工作线程被唤醒，它往 socket上写入服务器处理客户请求的结果

模拟Proactor模式

使用同步VO方式模拟出 Proactor模式。

原理是:主线程执行数据读写 *** 作，读写完成之后，主线程向工作线程通知这一"“完成事件”。

那么从工作线程的角度来看，它们就直接获得了数据读写的结果，接下来要做的只是对读写的结果进行逻辑处理。

使用同步I/O模型(以epoll_wait为例）)模拟出的 Proactor模式的工作流程如下:

1. 主线程往epoll内核事件表中注册socket上的读就绪事件。
   

   
   

2. 主线程调用epoll_wait等待socket上有数据可读。
   

   
   

3. 当socket 上有数据可读时，epoll_wait 通知主线程。
   

   
   

   主线程从socket循环读取数据，直到没有更多数据可读，然后将读取到的数据封装成一个请求对象并插入请求队列。
   

   
   

4. 睡眠在请求队列上的某个工作线程被唤醒，它获得请求对象并处理客户请求，然后往epoll内核事件表中注册socket 上的写就绪事件。
   

   
   

5. 主线程调用epoll_wait 等待socket可写。
   

   
   

6. 当socket可写时，epoll_wait通知主线程。
   

   
   

   主线程往socket上写入服务器处理客户请求的结果。
   

   
   

   
   

四、代码实现

主函数

int main() {
    /* 守护进程 后台运行 */
    //daemon(1, 0); 

    WebServer server(
        1316, 3, 60000, false,             /* 端口 ET模式 timeoutMs超时  优雅退出  */
        3306, "root", "root", "webserver", /* Mysql配置 */
        12, 6, true, 1, 1024);             /* 连接池数量 线程池数量 日志开关 日志等级 日志异步队列容量 */
    server.Start();
} 

1、构造函数

对端口、ET模式、timeoutMs超时、退出模式、Mysql配置、连接池数量、线程池数量、日志开关、日志等级、日志异步队列容量进行设置

WebServer::WebServer(
            int port, int trigMode, int timeoutMS, bool OptLinger,
            int sqlPort, const char* sqlUser, const  char* sqlPwd,
            const char* dbName, int connPoolNum, int threadNum,
            bool openLog, int logLevel, int logQueSize):
            port_(port), openLinger_(OptLinger), timeoutMS_(timeoutMS), isClose_(false),
            timer_(new HeapTimer()), threadpool_(new ThreadPool(threadNum)), epoller_(new Epoller())
    {
    srcDir_ = getcwd(nullptr, 256);//获取当前工作目录
    assert(srcDir_);
    ///home/xyh/WebServer-master/resources/
    strncat(srcDir_, "/resources/", 16);//拼接目录，资源路径
    HttpConn::userCount = 0;
    HttpConn::srcDir = srcDir_;
    SqlConnPool::Instance()->Init("localhost", sqlPort, sqlUser, sqlPwd, dbName, connPoolNum);

    //初始化事件的模式
    InitEventMode_(trigMode);
    //初始化套接字
    if(!InitSocket_()) { isClose_ = true;}//初始化失败，关闭服务器
    //日志相关
    if(openLog) {
        Log::Instance()->init(logLevel, "./log", ".log", logQueSize);
        if(isClose_) { LOG_ERROR("========== Server init error!=========="); }
        else {
            LOG_INFO("========== Server init ==========");
            LOG_INFO("Port:%d, OpenLinger: %s", port_, OptLinger? "true":"false");
            LOG_INFO("Listen Mode: %s, OpenConn Mode: %s",
                            (listenEvent_ & EPOLLET ? "ET": "LT"),
                            (connEvent_ & EPOLLET ? "ET": "LT"));
            LOG_INFO("LogSys level: %d", logLevel);
            LOG_INFO("srcDir: %s", HttpConn::srcDir);
            LOG_INFO("SqlConnPool num: %d, ThreadPool num: %d", connPoolNum, threadNum);
        }
    }
}

1. 获取资源目录
2. 初始化事件的模式
3. 初始化套接字
4. 日志相关

初始化套接字InitSocket_()

bool WebServer::InitSocket_() {
    int ret;
    struct sockaddr_in addr;
    if(port_ > 65535 || port_ < 1024) {
        LOG_ERROR("Port:%d error!",  port_);
        return false;
    }
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = htonl(INADDR_ANY);//转换网络字节
    addr.sin_port = htons(port_);
    struct linger optLinger = { 0 };
    if(openLinger_) {
        /* 优雅关闭: 直到所剩数据发送完毕或超时 */
        optLinger.l_onoff = 1;
        optLinger.l_linger = 1;
    }

    listenFd_ = socket(AF_INET, SOCK_STREAM, 0);//创建一个socket，返回监听文件描述符
    if(listenFd_ < 0) {
        LOG_ERROR("Create socket error!", port_);
        return false;
    }

    ret = setsockopt(listenFd_, SOL_SOCKET, SO_LINGER, &optLinger, sizeof(optLinger));
    if(ret < 0) {
        close(listenFd_);
        LOG_ERROR("Init linger error!", port_);
        return false;
    }
    int optval = 1;
    /* 端口复用 */
    /* 只有最后一个套接字会正常接收数据。

 */
    ret = setsockopt(listenFd_, SOL_SOCKET, SO_REUSEADDR, (const void*)&optval, sizeof(int));
    if(ret == -1) {
        LOG_ERROR("set socket setsockopt error !");
        close(listenFd_);
        return false;
    }

    ret = bind(listenFd_, (struct sockaddr *)&addr, sizeof(addr));//绑定
    if(ret < 0) {
        LOG_ERROR("Bind Port:%d error!", port_);
        close(listenFd_);
        return false;
    }

    ret = listen(listenFd_, 6);
    if(ret < 0) {
        LOG_ERROR("Listen port:%d error!", port_);
        close(listenFd_);
        return false;
    }
    ret = epoller_->AddFd(listenFd_,  listenEvent_ | EPOLLIN);
    if(ret == 0) {
        LOG_ERROR("Add listen error!");
        close(listenFd_);
        return false;
    }
    SetFdNonblock(listenFd_);//设置文件描述符非阻塞
    LOG_INFO("Server port:%d", port_);
    return true;
}

1. 创建一个socket，返回监听文件描述符
2. 端口复用
3. 绑定bind();
4. 监听listen();
5. 调用epoll添加监听epoller_->AddFd();
6. 设置文件描述符非阻塞

2、Start()；

void WebServer::Start() {
    int timeMS = -1;  /* epoll wait timeout == -1 无事件将阻塞 */
    if(!isClose_) { LOG_INFO("========== Server start =========="); }
    while(!isClose_) {//只要服务器不关闭就一直循环
        //超时连接相关todo
        if(timeoutMS_ > 0) {
            timeMS = timer_->GetNextTick();
        }
        
        int eventCnt = epoller_->Wait(timeMS);//检测到有多少个
        for(int i = 0; i < eventCnt; i++) {
            /* 处理事件 */
            //获取检测的文件描述符和事件
            int fd = epoller_->GetEventFd(i);
            uint32_t events = epoller_->GetEvents(i);
            //如果检测到的是监听的，就处理监听文件
            if(fd == listenFd_) {
                DealListen_();//接收客户端连接
            }
            else if(events & (EPOLLRDHUP | EPOLLHUP | EPOLLERR)) {
                assert(users_.count(fd) > 0);
                CloseConn_(&users_[fd]);//出现错误就关闭连接
            }
            else if(events & EPOLLIN) {
                assert(users_.count(fd) > 0);
                DealRead_(&users_[fd]);//处理读 *** 作
            }
            else if(events & EPOLLOUT) {
                assert(users_.count(fd) > 0);
                DealWrite_(&users_[fd]);//处理写 *** 作
            } else {
                LOG_ERROR("Unexpected event");
            }
        }
    }
}

eventCnt = epoller_->Wait(timeMS);
主线程调用epoll_wait等待socket上有数据可读。

处理事件

• DealListen_();//接收客户端连接
• CloseConn_(&users_[fd]);//出现错误就关闭连接
• DealRead_(&users_[fd]);//处理读 *** 作
• DealWrite_(&users_[fd]);//处理写 *** 作

2.1 DealListen_();接收客户端连接

void WebServer::DealListen_() {
    struct sockaddr_in addr;
    socklen_t len = sizeof(addr);//保存连接的客户端信息
    do {
        int fd = accept(listenFd_, (struct sockaddr *)&addr, &len);
        if(fd <= 0) { return;}
        else if(HttpConn::userCount >= MAX_FD) {//超出最大连接数量
            SendError_(fd, "Server busy!");
            LOG_WARN("Clients is full!");
            return;
        }
        AddClient_(fd, addr);//连接成功，添加客户端
    } while(listenEvent_ & EPOLLET);//ET模式需要循环取
}

2.2 CloseConn_(&users_[fd]);关闭连接

void WebServer::CloseConn_(HttpConn* client) {
    assert(client);
    LOG_INFO("Client[%d] quit!", client->GetFd());
    epoller_->DelFd(client->GetFd());
    client->Close();
}

2.3 DealRead_(&users_[fd]);//处理读 *** 作

void WebServer::DealRead_(HttpConn* client) {
    assert(client);
    ExtentTime_(client);
    threadpool_->AddTask(std::bind(&WebServer::OnRead_, this, client));
}

2.3.1 WebServer::OnRead

void WebServer::OnRead_(HttpConn* client) {
    assert(client);
    int ret = -1;
    int readErrno = 0;
    ret = client->read(&readErrno);//读取客户端数据
    if(ret <= 0 && readErrno != EAGAIN) {
        CloseConn_(client);
        return;
    }
    //处理业务逻辑
    OnProcess(client);
}

2.3.1.1 处理业务逻辑 OnProcess(client);

void WebServer::OnProcess(HttpConn* client) {
    if(client->process()) {
        epoller_->ModFd(client->GetFd(), connEvent_ | EPOLLOUT);//修改文件描述符
    } else {
        epoller_->ModFd(client->GetFd(), connEvent_ | EPOLLIN);
    }
}

2.3.2 添加线程 threadpool_->AddTask

    void AddTask(F&& task) {//从池子里添加一个任务
        {
            std::lock_guard<std::mutex> locker(pool_->mtx);
            pool_->tasks.emplace(std::forward<F>(task));
        }
        pool_->cond.notify_one();//唤醒一个线程
    }

2.4 DealWrite_(&users_[fd]);//处理写 *** 作 WebServer::OnWrite

void WebServer::OnWrite_(HttpConn* client) {
    assert(client);
    int ret = -1;
    int writeErrno = 0;
    ret = client->write(&writeErrno);
    if(client->ToWriteBytes() == 0) {
        /* 传输完成 */
        if(client->IsKeepAlive()) {
            OnProcess(client);
            return;
        }
    }
    else if(ret < 0) {
        if(writeErrno == EAGAIN) {
            /* 继续传输 */
            epoller_->ModFd(client->GetFd(), connEvent_ | EPOLLOUT);
            return;
        }
    }
    CloseConn_(client);
}
}