Linux高并发服务器开发第六章：项目执行流程讲解

1. 线程池

线程池是由服务器预先创建的一组子线程，线程池中的线程数量应该和 CPU 数量差不多。线程池中的所有子线程都运行着相同的代码。当有新的任务到来时，主线程将通过某种方式选择线程池的某一个子线程来为之服务。相比与动态的创建子线程，选择一个已经存在的子线程的代价显然要小很多。至于主线程选择哪个子线程来为新任务服务，则有多种方式：

• 主线程使用某种算法来主动选择子线程。最简单、最常用的算法是随机算法和Round Robin（轮流选取）算法，但更优秀、更智能的算法将使任务在各个工作线程中更均匀地分配，从而减轻服务器的整体压力。
• 主线程和所有子线程通过一个共享的工作队列来同步，子线程都睡眠在该工作队列上。当有新的任务到来时，主线程将任务添加到工作队列中。这将唤醒正在等待任务的子线程，不过只有一个子线程将获得新任务的“接管权”，它可以从工作队列中取出任务并执行之，而其他子线程将继续睡眠在工作队列上。

线程池代码webserver/code/pool/threadpool.h

#ifndef THREADPOOL_H
#define THREADPOOL_H

#include 
#include 
#include 
#include 
#include 
class ThreadPool {
public:
    explicit ThreadPool(size_t threadCount = 8): pool_(std::make_shared<Pool>()) {
            assert(threadCount > 0);

            // 创建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_; // 池子
};


#endif //THREADPOOL_H
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

创建了一个线程池，默认为8个子线程，线程通过拿到池子的锁，去操作池子中的任务队列。

• 当任务队列中有任务时，从对头取出一个任务，然后释放锁，接着子线程执行任务。
• 当池子未关闭且任务队列中没有任务时，通过条件变量等待任务的到来。
• 当池子关闭后，直接释放该线程。

函数AddTask(F&& task)，获得池子的锁之后，向任务队列中添加一个任务对象，并通知子线程去执行任务。
通过互斥锁+条件变量来实现线程同步。只有获得锁，才能对pool_中的任务队列进行操作。

---

2. 代码分析

首先我们查看main()函数，创建了一个WebServer的对象server，传入构造函数需要的参数，也就是说把整个Web服务器封装成了一个类WebServer，当调用Start()方法即可启动服务器。
了解完主函数中的内容之后，接着查看webserver.h，大致看一看WebServer类中有哪些成员变量和方法。

#ifndef WEBSERVER_H
#define WEBSERVER_H

#include 
#include        // fcntl()
#include       // close()
#include 
#include 
#include 
#include 
#include 

#include "epoller.h"
#include "../log/log.h"
#include "../timer/heaptimer.h"
#include "../pool/sqlconnpool.h"
#include "../pool/threadpool.h"
#include "../pool/sqlconnRAII.h"
#include "../http/httpconn.h"

class WebServer {
public:
    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);

    ~WebServer();
    void Start();

private:
    bool InitSocket_(); 
    void InitEventMode_(int trigMode);
    void AddClient_(int fd, sockaddr_in addr);
  
    void DealListen_();
    void DealWrite_(HttpConn* client);
    void DealRead_(HttpConn* client);

    void SendError_(int fd, const char*info);
    void ExtentTime_(HttpConn* client);
    void CloseConn_(HttpConn* client);

    void OnRead_(HttpConn* client);
    void OnWrite_(HttpConn* client);
    void OnProcess(HttpConn* client);

    static const int MAX_FD = 65536; // 最大的文件描述符的个数

    static int SetFdNonblock(int fd); // 设置文件描述符非阻塞

    int port_;  // 端口
    bool openLinger_; // 是否打开优雅关闭
    int timeoutMS_;  /* 毫秒MS */
    bool isClose_; // 是否关闭
    int listenFd_; // 监听的文件描述符
    char* srcDir_; // 资源的目录
    
    uint32_t listenEvent_; // 监听的文件描述符的事件
    uint32_t connEvent_; // 连接的文件描述符的事件
   
    std::unique_ptr<HeapTimer> timer_; // 定时器
    std::unique_ptr<ThreadPool> threadpool_; // 线程池
    std::unique_ptr<Epoller> epoller_; // epoll对象
    std::unordered_map<int, HttpConn> users_; // 保存的是客户端连接的信息，其中first是文件描述符，second是对应的客户端信息
};


#endif //WEBSERVER_H
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

注意⚠️：一个客户端连接到服务器以后，会封装成一个HttpConn对象，HttpConn对象中包含客户端连接的相关信息，并把HttpConn对象保存到users_中。

---

接着查看webserver.cpp，了解服务器的具体实现过程，首先看WebServer类的构造函数：

srcDir_ = getcwd(nullptr, 256); // 获取当前的工作路径
strncat(srcDir_, "/resources/", 16);
1
2

初始化资源保存的路径。

HttpConn::userCount = 0; 
HttpConn::srcDir = srcDir_;
1
2

将连接的用户总数量初始化为0，并设置资源的目录。同时这两个变量是静态变量，被所有的连接对象共享。

InitEventMode_(trigMode);
1

初始化事件的模式，设置监听的文件描述符和通信的文件描述符的模式，主要是涉及到epoll是ET还是LT模式。

if(!InitSocket_()) { isClose_ = true;}
1

接着查看WebServer::InitSocket_(),

/* Create listenFd */
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);
    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); // 将监听的文件描述符listenFd_添加到epoller_上
    if(ret == 0) {
        LOG_ERROR("Add listen error!");
        close(listenFd_);
        return false;
    }
    SetFdNonblock(listenFd_);
    LOG_INFO("Server port:%d", port_);
    return true;
}

// 设置文件描述符非阻塞
int WebServer::SetFdNonblock(int fd) {
    assert(fd > 0);
    int flag = fcntl(fd, F_GETFD, 0);
    flag |= O_NONBLOCK;
    return fcntl(fd, F_SETFL, flag);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

回到webserver.cpp中

if(!InitSocket_()) { isClose_ = true;}
1

当套接字初始化成功后，则整个构造函数就执行完成了。

---

接下来，就要执行server.Start()了，我们来看Start()函数是如何执行的：

int eventCnt = epoller_->Wait(timeMS);
1

当服务器开启时，调用epoller_->Wait(timeMS);检测客户端的事件到来，返回值eventCnt表示检测到有多少个事件产生了动作。
通过for循环遍历，处理相应的事件。

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");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

接着，我们开始看这些处理事件的函数是如何执行的
（1）当接收到事件的文件描述符是listenFd_时，就执行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;} // 如果没有新的客户端连接，直接退出while循环
        else if(HttpConn::userCount >= MAX_FD) {
            SendError_(fd, "Server busy!");
            LOG_WARN("Clients is full!");
            return;
        }
        AddClient_(fd, addr); // 添加客户端
    } while(listenEvent_ & EPOLLET); // 如果监听文件描述符设置成了ET模式，需要通过while循环一次性将连接的所有用户添加到epoller_，
                                     // 监听这些连接用户的fd的EPOLLIN事件。
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

DealListen_()中最主要的工作就是执行了AddClient_(fd, addr);添加客户端到epoller_。

void WebServer::AddClient_(int fd, sockaddr_in addr) {
    assert(fd > 0);
    users_[fd].init(fd, addr); // 初始化users_的HttpConn
    if(timeoutMS_ > 0) {
        timer_->add(fd, timeoutMS_, std::bind(&WebServer::CloseConn_, this, &users_[fd]));
    }
    epoller_->AddFd(fd, EPOLLIN | connEvent_); // 将新连接到服务器的客户端fd添加到epoller_上，并且监听其EPOLLIN事件
    SetFdNonblock(fd); // 设置非阻塞
    LOG_INFO("Client[%d] in!", users_[fd].GetFd());
}
1
2
3
4
5
6
7
8
9
10

（2）当接收到的是错误事件时，就执行CloseConn_(&users_[fd]);

void WebServer::CloseConn_(HttpConn* client) {
    assert(client);
    LOG_INFO("Client[%d] quit!", client->GetFd());
    epoller_->DelFd(client->GetFd()); // 从epoll_中删除该文件描述符
    client->Close(); // 将isClose_置为true，关闭文件描述符fd_，连接的用户userCount-1
}
1
2
3
4
5
6

（3）当有读事件到来时，调用DealRead_(&users_[fd]);

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

把任务添加到任务队列中，就通知线程池中的子线程去处理任务，执行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);
}
1
2
3
4
5
6
7
8
9
10
11
12
13

其中，ret = client->read(&readErrno); 读取客户端的数据时，调用了以下函数

ssize_t HttpConn::read(int* saveErrno) {
    ssize_t len = -1;
    do {
        len = readBuff_.ReadFd(fd_, saveErrno);
        if (len <= 0) {
            break;
        }
    } while (isET);
    return len;
}
1
2
3
4
5
6
7
8
9
10

这里运用了do...while循环，假如是ET模式，则一次性将数据全部读出来，返回读取到的字节数。
在HttpConn类中有个readBuff_成员，该成员的类型为自定义的Buffer类型，表示读缓存区，把读取到数据保存到readBuff_的vector buffer_中。
其中底层的ReadFd()函数如下：

ssize_t Buffer::ReadFd(int fd, int* saveErrno) {
    char buff[65535]; // 临时的数组，保证能够把所有的数据都读出来

    struct iovec iov[2];
    const size_t writable = WritableBytes();

    /* 分散读，保证数据全部读完 */
    iov[0].iov_base = BeginPtr_() + writePos_;
    iov[0].iov_len = writable;
    iov[1].iov_base = buff;
    iov[1].iov_len = sizeof(buff);

    const ssize_t len = readv(fd, iov, 2);
    if(len < 0) {
        *saveErrno = errno;
    }
    else if(static_cast<size_t>(len) <= writable) {
        writePos_ += len;
    }
    else {
        writePos_ = buffer_.size();
        Append(buff, len - writable); // buff临时数组，len-writable是临时数组中的数据个数
    }
    return len;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

其中当开辟的缓存区无法读取完所有数据时，就需要用临时数组暂时进行存储，然后动态开辟缓存区，再将临时数组中的数据拷贝到缓存区。上方代码Append(buff, len - writable);就是将临时数组中的数据拷贝到缓存区中，它会调用下边这个函数：

void Buffer::Append(const char* str, size_t len) {
    assert(str);
    EnsureWriteable(len);
    std::copy(str, str + len, BeginWrite());
    HasWritten(len);
}
1
2
3
4
5
6

在读取数据的时候，调用readv(fd, iov, 2);进行分散读，使用struct iovec iov[2]来指定读取到的数据储存的位置，其中iov[1]是指定的一个临时数组，保证所有的数据全部读完。
最终，我们读取到了数据，然后回到webserver.cpp中的void WebServer::OnRead_(HttpConn* client) 中，接着就是执行OnProcess(client);，进行业务逻辑的处理。

void WebServer::OnProcess(HttpConn* client) {
    if(client->process()) {
        epoller_->ModFd(client->GetFd(), connEvent_ | EPOLLOUT);
    } else {
        epoller_->ModFd(client->GetFd(), connEvent_ | EPOLLIN);
    }
}
1
2
3
4
5
6
7

我们发现，其实真正处理业务逻辑是通过client->process()，这个函数如下：

bool HttpConn::process() {
    request_.Init();
    if(readBuff_.ReadableBytes() <= 0) {
        return false;
    }
    else if(request_.parse(readBuff_)) {
        LOG_DEBUG("%s", request_.path().c_str());
        response_.Init(srcDir, request_.path(), request_.IsKeepAlive(), 200);
    } else {
        response_.Init(srcDir, request_.path(), false, 400);
    }

    response_.MakeResponse(writeBuff_);
    /* 响应头 */
    iov_[0].iov_base = const_cast<char*>(writeBuff_.Peek());
    iov_[0].iov_len = writeBuff_.ReadableBytes();
    iovCnt_ = 1;

    /* 文件 */
    if(response_.FileLen() > 0  && response_.File()) {
        iov_[1].iov_base = response_.File();
        iov_[1].iov_len = response_.FileLen();
        iovCnt_ = 2;
    }
    LOG_DEBUG("filesize:%d, %d  to %d", response_.FileLen() , iovCnt_, ToWriteBytes());
    return true;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

上方代码中的request_.parse(readBuff_)对读取到的数据进行解析，在了解解析的底层代码之前，我们首先需要知道什么是有限状态机，接下来我就来给大家讲解一下。
有限状态机
逻辑单元内部的一种高效编程方法：有限状态机（finite state machine）。
有的应用层协议头部包含数据包类型字段，每种类型可以映射为逻辑单元的一种执行状态，服务器可以根据它来编写相应的处理逻辑。
如下是一种状态独立的有限状态机：

STATE_MACHINE( Package _pack ) 
{ 
	PackageType _type = _pack.GetType(); 
	switch( _type ) 
	{ 
		case type_A: 
			process_package_A( _pack ); 
			break; 
		case type_B: 
			process_package_B( _pack ); 
			break; 
	} 
}
1
2
3
4
5
6
7
8
9
10
11
12
13

这是一个简单的有限状态机，只不过该状态机的每个状态都是相互独立的，即状态之间没有相互转移。状态之间的转移是需要状态机内部驱动，如下代码：

STATE_MACHINE() 
{ 
	State cur_State = type_A;
	while( cur_State != type_C )
	{ 
		Package _pack = getNewPackage(); 
		switch( cur_State ) 
		{
			case type_A: 
				process_package_state_A( _pack ); 
				cur_State = type_B; 
				break; 
			case type_B: 
				process_package_state_B( _pack ); 
				cur_State = type_C; 
				break; 
		} 
	} 
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

该状态机包含三种状态：type_A、type_B 和 type_C，其中 type_A 是状态机的开始状态，type_C 是状态机的结束状态。状态机的当前状态记录在 cur_State 变量中。在一趟循环过程中，状态机先通过 getNewPackage 方法获得一个新的数据包，然后根据 cur_State 变量的值判断如何处理该数据包。数据包处理完之后，状态机通过给 cur_State 变量传递目标状态值来实现状态转移。那么当状态机进入下一趟循环时，它将执行新的状态对应的逻辑。
知道状态机的基本概念之后，我们就来看解析数据的底层代码：

bool HttpRequest::parse(Buffer& buff) {
    const char CRLF[] = "\r\n";
    if(buff.ReadableBytes() <= 0) {
        return false;
    }
    while(buff.ReadableBytes() && state_ != FINISH) {
        // 根据\r\n为结束标志，获取一行数据结束的位置
        const char* lineEnd = search(buff.Peek(), buff.BeginWriteConst(), CRLF, CRLF + 2);
        std::string line(buff.Peek(), lineEnd);
        switch(state_)
        {
        case REQUEST_LINE:
            if(!ParseRequestLine_(line)) {
                return false;
            }
            ParsePath_();
            break;    
        case HEADERS:
            ParseHeader_(line);
            if(buff.ReadableBytes() <= 2) {
                state_ = FINISH;
            }
            break;
        case BODY:
            ParseBody_(line);
            break;
        default:
            break;
        }
        if(lineEnd == buff.BeginWrite()) { break; }
        buff.RetrieveUntil(lineEnd + 2);
    }
    LOG_DEBUG("[%s], [%s], [%s]", method_.c_str(), path_.c_str(), version_.c_str());
    return true;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

通过不断循环，解析请求报文中的数据，直到state_为FINISH时，才跳出循环，代表解析完成。解析的过程的过程也比较简单，就是不断读取每一行数据，然后通过当前状态，通过有限状态机去调用的函数做相应的处理逻辑，同时在相应的处理函数中，会去做状态的改变。当解析成功以后，就会执行response_.Init(srcDir, request_.path(), request_.IsKeepAlive(), 200);，即初始化响应的数据。
然后执行response_.MakeResponse(writeBuff_);去创建响应的状态行和响应头部的数据保存到writeBuff_的buffer_成员中，而响应体的数据则是被映射到了内存中。因此，接下来我们去写数据的时候，采用分散写的方式。
如果处理业务逻辑成功，则执行epoller_->ModFd(client->GetFd(), connEvent_ | EPOLLOUT);，将通信的文件描述符设置为EPOLLOUT，监听它是否可写的事件。至此，DealRead_(&users_[fd]); 处理读操作执行完成。接着，主线程就会去监听该文件描述符的可写事件。
（4）当监听到事件可写时，执行DealWrite_(&users_[fd]);处理写操作

void WebServer::DealWrite_(HttpConn* client) {
    assert(client);
    ExtentTime_(client);
    threadpool_->AddTask(std::bind(&WebServer::OnWrite_, this, client));
}
1
2
3
4
5

把任务添加到任务队列中，就通知线程池中的子线程去处理任务，执行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);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

在执行ret = client->write(&writeErrno);时，其实才是真正执行了写操作，调用了以下函数：

ssize_t HttpConn::write(int* saveErrno) {
    ssize_t len = -1;
    do {
        len = writev(fd_, iov_, iovCnt_);
        if(len <= 0) {
            *saveErrno = errno;
            break;
        }
        if(iov_[0].iov_len + iov_[1].iov_len  == 0) { break; } /* 传输结束 */
        else if(static_cast<size_t>(len) > iov_[0].iov_len) {
            iov_[1].iov_base = (uint8_t*) iov_[1].iov_base + (len - iov_[0].iov_len);
            iov_[1].iov_len -= (len - iov_[0].iov_len);
            if(iov_[0].iov_len) {
                writeBuff_.RetrieveAll();
                iov_[0].iov_len = 0;
            }
        }
        else {
            iov_[0].iov_base = (uint8_t*)iov_[0].iov_base + len; 
            iov_[0].iov_len -= len; 
            writeBuff_.Retrieve(len);
        }
    } while(isET || ToWriteBytes() > 10240);
    return len;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

在写数据的时候，调用writev(fd_, iov_, iovCnt_)去分散的写，当写完成后，解除内存映射。
至此，其实我们整个服务器执行的流程就结束了，我再简述一下整个流程：
在主函数中定义了一个WebServer对象，首先会去调用WebServer类的构造函数，构造函数中初始化了资源存放的路径、用户数、连接池、事件的模式，建立Socket通信，还对日志做了初始化操作。
服务器启动以后，主线程通过while循环不断的通过epoller_->Wait(timeMS)去监听是否有事件到达，如果发现有事件到达，就会去处理相应的事件。
1）如果获取到的fd是监听文件描述符时，就调用DealListen_()去处理监听的操作，调用accept()接受新的客户端的连接， 然后初始化一个HttpConn的连接对象，并将客户端连接的文件描述符添加到epoller_中，并监听它是否有可读的数据到达。
2）如果有数据到达时，就调用DealRead()函数去处理读操作。DealRead()函数中将真正的读操作的任务交给线程池，添加到任务队列中，通知子线程去执行任务，工作线程拿到任务后，执行Onread_函数。Onread_首先去读取客户端数据，然后执行OnProcess(client)中的client->process()处理业务逻辑，从读缓存区readBuff_中解析HTTP请求request_.parse(readBuff_)，解析完成后，生成response_的响应数据。当业务逻辑处理完成后，就会将通信的文件描述符设置为EPOLLOUT，监听它是否可写的事件。至此，DealRead_(&users_[fd]); 处理读操作执行完成。接着，主线程就会去监听该文件描述符的可写事件。
3）当主线程通过epoller_->Wait(timeMS)检测到该文件描述符可写后，就执行DealWrite()函数处理写操作。DealRead()函数中将真正的写操作的任务交给线程池，添加到任务队列中，通知子线程去执行任务，工作线程拿到任务后，执行Onwrite_函数。
4）如果检测到错误事件时，就关闭连接。

---

3. 异步日志模块

日志功能的本质就是往文件中记录数据，这就涉及到文件IO。如果实现为同步日志，当需要写入的数据比较多，则开销比较大，服务器在写日志时，客户端此时无法访问服务器，程序就会停在原地等待日志系统的写完成后，才能继续向下执行。如果实现为异步日志，则将需要写的日志加入到一个队列中，子线程从队列中去取得任务，然后由子线程完成往日志系统中写数据的操作，这样主线程就不会被阻塞在原地，可以继续向下执行。

---

4. EPOLLONESHOT事件

即使可以使用 ET 模式，一个 socket 上的某个事件还是可能被触发多次。这在并发程序中就会引起一个问题。比如一个线程在读取完某个 socket 上的数据后开始处理这些数据，而在数据的处理过程中该 socket 上又有新数据可读（EPOLLIN 再次被触发），此时另外一个线程被唤醒来读取这些新的数据。于是就出现了两个线程同时操作一个 socket 的局面。一个 socket 连接在任一时刻都只被一个线程处理，可以使用 epoll 的 EPOLLONESHOT 事件实现。对于注册了 EPOLLONESHOT 事件的文件描述符，操作系统最多触发其上注册的一个可读、可写或者异常事件，且只触发一次，除非我们使用 epoll_ctl 函数重置该文件描述符上注册的 EPOLLONESHOT 事件。这样，当一个线程在处理某个 socket 时，其他线程是不可能有机会操作该 socket 的。但反过来思考，注册了 EPOLLONESHOT 事件的 socket 一旦被某个线程处理完毕， 该线程就应该立即重置这个 socket 上的 EPOLLONESHOT 事件，以确保这个 socket 下一次可读时，其 EPOLLIN 事件能被触发，进而让其他工作线程有机会继续处理这个 socket。

---

5. 服务器压力测试

Webbench 是 Linux 上一款知名的、优秀的 web 性能压力测试工具。它是由 Lionbridge 公司开发。

• 测试处在相同硬件上，不同服务的性能以及不同硬件上同一个服务的运行状况。
• 展示服务器的两项内容：每秒钟响应请求数和每秒钟传输数据量。

基本原理：Webbench 首先 fork 出多个子进程，每个子进程都循环做 web 访问测试。子进程把访问的结果通过 pipe 告诉父进程，父进程做最终的统计结果。

wenbench的使用方法：
在项目的目录下有个webbench-1.5的文件夹，通过cd webbench-1.5进入webbench-1.5，默认只有Makefile、socket.c、webbench.c三个文件，然后执行make进行编译，生成一个可执行文件webbench，然后就可以使用它了。

webbench -c 1000 -t 30 http://192.168.110.129:10000/index.html 
参数：
	-c 表示客户端数 
	-t 表示时间
1
2
3
4