阅读量:0
项目复习：从封装epoll_server到实现reactor服务器(part2)
- 项目复习：从封装epoll_server到实现reactor服务器(part2)
仓库：https://github.com/Yufccode/Reactor-based-HyperWebServer/
这一部分不多说，直接进入代码编写，后面有需要解释的特性，再解释。
基本结构搭建好
tcp_server.hpp
#ifndef __YUFC_TCP_SERVER__ #define __YUFC_TCP_SERVER__ #include <iostream> #include <string> class tcp_server { private: public: }; #endif 为什么上面我们写的epoll的recv是不正确的？
我们好好的看一下这一份代码。
void epoll_recver(int sock) { // 读取这个普通套接字里面的内容 // 1. 读取数据 // 2. 处理数据 char buffer[10240]; size_t n = recv(sock, buffer, sizeof(buffer) - 1, 0); if (n > 0) { // 假设这里就是读到了一个完整的报文 buffer[n] = 0; __handler_request(buffer); // 进行回调！ } else if (n <= 0) { // 对端关闭文件描述符 // 让epoll不再关注这个文件描述符 // 一定要先从epoll中去掉，才能close文件描述符 bool res = __epoll::control_epoll(__epoll_fd, EPOLL_CTL_DEL, sock, 0); assert(res); // 保证是成功的，因为一般来说都是成功的，所以直接assert (void)res; close(sock); if (n == 0) logMessage(NORMAL, "client %d quit, me quit too ...", sock); else if (n < 0) logMessage(NORMAL, "client recv %d error, close error sock", sock); } } 调用recv的时候读取的是完整的报文吗？不知道，所以我们buffer[10240]里面的也是不完整的报文。
此时能够直接交付给__handler_request()做我们的业务逻辑吗？不可以！这样是错的！
但是，我们上节课写的epoll服务器，也没有对应的缓冲区让我们暂时存放，等报文完整才__handler_request()，所以，上面的epoll_recver是错误的。
而且，一个epoll_server可能有成百上千的sock，每一个sock，都需要缓冲区才行！
sock要封装了，要维护缓冲区
所以要维护！
tcp_server.hpp
class connection { public: using func_t = std::function<void(connection*)>; public: connection(); ~connection(); public: int __sock; // io的文件描述符 func_t __recv_callback; func_t __send_callback; func_t __except_callback; std::string __in_buffer; // 输入缓冲区（暂时没有处理二进制流） std::string __out_buffer; // 输出缓冲区 tcp_server* tsvr; // 回指指针 }; 每一个文件描述符，都要封装！因为每一个文件描述符，都有可能有三种时间，读事件，写事件，异常事件。
简单来说就是，我们通过封装func_t，后面这个sock需要读的时候，就去调用__recv_callback。后面的同理！
这三个函数对象，都要从用户层传过来！
当然，输入输出缓冲区也是需要的！目前的用string做缓冲区，暂时没有办法处理二进制流，但是文本是可以的。
tcp_server* tsvr; 这个指针也很好理解，毕竟我这个connector总要知道是哪个服务器在调用我吧。
然后初始化什么的这里就不说了，看代码就行。然后我也写了一个void set_callback(func_t recv_cb, func_t send_cb, func_t except_cb)来设置这个sock封装(connector)的回调。
封装epoll(1)
这次封装和上次就不同了，这次我们把epoll的所有动作都封装到epoll.hpp里面去，包括epollfd什么的，我们的tcp_server就不管这个epoll里面的事情了，tcp_server只负责调用接口！
class tcp_server { public: const static int gport = 8080; private: int __listen_sock; int __port; __epoll __poll; // 这里直接维护一个epoll 然后，其实在很多真正的开发场景中，多路转接的方法都是可以选的，一般就是写一个虚基类class poll {}，里面的函数就是纯虚的，然后去继承epoll, select, poll 这些。
所以对于tcp_server, 我们多路转接都叫poll比较好，因为也不一定要用epoll。
继续先写tcp_server
tcp_server(int port = gport) : __port(gport) { // 1. 创建listensock __listen_sock = Sock::Socket(); Sock::Bind(__listen_sock, __port); Sock::Listen(__listen_sock); // 2. 创建多路转接对象 __poll.create_poll(); // 3. 封装listensock, 因为以后在tcpserver里面，已经不会再有裸露的sock了，都是要封装的 connection* conn = new connection(__listen_sock); conn->set_callback(read_callback, nullptr, nullptr); conn->__tsvr = this; // 让conn对象指向自己 } 这里需要理解一下！首先，我们知道，有两种sock，一种是listensock，一种是其他sock。
但是对于listensock来说，他只需要关注读事件啊！因此，conn->set_callback(read_callback, nullptr, nullptr);。
tcp_server要对所有的connection进行管理
因为处理监听套接字，未来我们会存在大量的socket，每一个sock都必须被封装成一个connction。
当服务器中存在大量的connection的时候，tcp_server就需要将所有的connection进行管理！
那就用一个哈希表来存！
std::unordered_map<int, connection*> __connection_map; 文件描述符是唯一的，所以O(1)直接找到对应的connection！
所以我们不仅要将sock添加到poll中，还要将sock加到映射表中去！
未来，poll会告诉我哪个sock就绪了，然后我就可以通过这个sock立刻找到对应的connection对象，然后缓冲区，回调，都有了！这句话很重要！
未来，poll会告诉我哪个sock就绪了，然后我就可以通过这个sock立刻找到对应的connection对象，然后缓冲区，回调，都有了！这句话很重要！
未来，poll会告诉我哪个sock就绪了，然后我就可以通过这个sock立刻找到对应的connection对象，然后缓冲区，回调，都有了！这句话很重要！
编写__add_connection：把套接字加入到服务中
什么意思呢？
加入到服务中有三件事：
- 封装成connection
- 加入到poll中
- 加入到connection的哈希表中
void __add_connection(int sock, func_t recv_cb, func_t send_cb, func_t except_cb) { // 不同种类的套接字都可以调用这个方法 // 1. 构建conn对象，封装sock connection* conn = new connection(sock); conn->set_callback(recv_cb, send_cb, except_cb); conn->__tsvr = this; // 让conn对象指向自己 // 2. 添加sock到poll中 __poll. // 3. 把封装好的conn放到map里面去 __connection_map.insert({ sock, conn }); } 所以我们需要在poll里面添加一个方法，把sock添加到epoll中，这个很简单，写epoll的时候已经写过了。
poll.hpp
class __epoll { private: int __epoll_fd; public: const static int gnum = 128; public: __epoll() { } ~__epoll() { } public: void create_poll() { __epoll_fd = epoll_create(gnum); if (__epoll_fd < 0) exit(5); } bool add_sock_to_poll(int sock, uint32_t events) { struct epoll_event ev; ev.events = events; ev.data.fd = sock; int n = epoll_ctl(__epoll_fd, EPOLL_CTL_ADD, sock, &ev); return n == 0; } public: }; 一个要注意的点
任何多路转接的服务器，一般默认只会打开对读取时间的关心，写入事件回按需进行打开。
所以，这里要这么写
void __add_connection(int sock, func_t recv_cb, func_t send_cb, func_t except_cb) { // 不同种类的套接字都可以调用这个方法 // 1. 构建conn对象，封装sock connection* conn = new connection(sock); conn->set_callback(recv_cb, send_cb, except_cb); conn->__tsvr = this; // 让conn对象指向自己 // 2. 添加sock到poll中 __poll.add_sock_to_poll(sock, EPOLLIN | EPOLLET); // 3. 把封装好的conn放到map里面去 __connection_map.insert({ sock, conn }); } 注意这一行：__poll.add_sock_to_poll(sock, EPOLLIN | EPOLLET);
默认打开关心读事件+设置ET模式！
设置了ET模式，我们要设置sock是非阻塞
所以在sock.hpp里面加
static bool SetNonBlock(int sock) { int fl = fcntl(sock, F_GETFL); if (fl < 0) return false; fcntl(sock, F_SETFL, fl | O_NONBLOCK); return true; } 所以，在__add_connection里面，第一步，先把套接字设置成非阻塞
void __add_connection(int sock, func_t recv_cb, func_t send_cb, func_t except_cb) { // 不同种类的套接字都可以调用这个方法 // 0. ！先把sock弄成非阻塞！ Sock::SetNonBlock(sock); listensock的recv方法是accept, 其他都不需要，怎么写？
tcp_server(int port = gport) : __port(gport) { // 1. 创建listensock __listen_sock = Sock::Socket(); Sock::Bind(__listen_sock, __port); Sock::Listen(__listen_sock); // 2. 创建多路转接对象 __poll.create_poll(); // 3. 添加listensock到tcp_server::__connection_map和poll中 __add_connection(__listen_sock, std::bind(&tcp_server::__accepter, this, std::placeholders::_1), nullptr, nullptr); } 这个绑定需要注意！因为__accepter写在类里面了，记得有个this，std::placeholders::_1就是我们的参数connection*，是__accepter要传的。
理解重要概念：派发
tcp_server.hpp
public: public: void dispather() { // 捞取所有就绪事件到revs数组中 } 在dispather中，我们需要先获取到所有就绪的事件！
所以我们要写一个，从epoll中捞取所有就绪事件的接口。
poll.hpp
int wait_poll(struct epoll_event revs[], int num) { return epoll_wait(__epoll_fd, revs, num, __timeout); } 所以dispather就是这样了
public: void dispather() { while (true) { loop_once(); } } void loop_once() { // 捞取所有就绪事件到revs数组中 int n = __poll.wait_poll(__revs, __revs_num); for (int i = 0; i < n; i++) { // 此时就可以去处理已经就绪事件了！ } } 已经拿到就绪的事件了，如何处理
其实就是
void loop_once() { // 捞取所有就绪事件到revs数组中 int n = __poll.wait_poll(__revs, __revs_num); for (int i = 0; i < n; i++) { // 此时就可以去处理已经就绪事件了！ } } 这里面的for循环怎么去写了。
注意细节
for (int i = 0; i < n; i++) { // 此时就可以去处理已经就绪事件了！ int sock = __revs[i].data.fd; uint32_t revents = __revs[i].events; if (revents & EPOLLIN) { // 这个事件读就绪了 // 1. 先判断这个套接字是否在这个map中存在 } } 首先，为什么要判断这个套接字在不在map中呢？因为服务器会有大量的异常情况，可能走到这里的时候，虽然事件就绪了，但是sock早就被关掉了，释放了，这个时候就要做区分。
写事件也是一样的处理方式
for (int i = 0; i < n; i++) { // 此时就可以去处理已经就绪事件了！ int sock = __revs[i].data.fd; uint32_t revents = __revs[i].events; if (revents & EPOLLIN) { // 这个事件读就绪了 // 1. 先判断这个套接字是否在这个map中存在 if (is_sock_in_map(sock) && __connection_map[sock]->__recv_callback != nullptr) __connection_map[sock]->__recv_callback(__connection_map[sock]); } if (revents & EPOLLOUT) { if (is_sock_in_map(sock) && __connection_map[sock]->__send_callback != nullptr) __connection_map[sock]->__send_callback(__connection_map[sock]); } } 如果这个sock在map里面（表示这个sock是合法的），与此同时，这个回调不是空指针（被设置过）
执行这个回调！
进行初步的测试
略
完善__accepter函数
tcp_server.hpp
void __accepter(connection* conn) { // logMessage(DEBUG, "accepter is called"); // 此时的listensock一定已经就绪了！ // v1 std::string client_ip; uint16_t client_port; int accept_errno = -1; int sock = Sock::Accept(conn->__sock, &client_ip, &client_port, &accept_errno); // accept回来的sock就是正常io的sock if (sock < 0) { } // 将sock托管给poll和tcpserver __add_connection(sock, ); } accept回来得到的sock，就是普通的io套接字，是要交给tcpserver托管的，要交给poll的。
所以，读回调，写回调，异常回调，我们在代码中可以先写好。
一共四个回调
所以可以总结了，一共有四个回调！
tcp_server.hpp
void __accepter(connection* conn); void __recver(connection* conn); void __sender(connection* conn); void __excepter(connection* conn); __accepter是给监听套接字用的！
后面三个是给普通的io套接字用的！
所以：
// 将sock托管给poll和tcpserver __add_connection(sock, std::bind(&tcp_server::__recver, this, std::placeholders::_1), std::bind(&tcp_server::__sender, this, std::placeholders::_1), std::bind(&tcp_server::__excepter, this, std::placeholders::_1)); 此时的__accepter还有问题吗
void __accepter(connection* conn) { // logMessage(DEBUG, "accepter is called"); // 此时的listensock一定已经就绪了！ // v1 std::string client_ip; uint16_t client_port; int accept_errno = -1; int sock = Sock::Accept(conn->__sock, &client_ip, &client_port, &accept_errno); // accept回来的sock就是正常io的sock if (sock < 0) { } // 将sock托管给poll和tcpserver __add_connection(sock, std::bind(&tcp_server::__recver, this, std::placeholders::_1), std::bind(&tcp_server::__sender, this, std::placeholders::_1), std::bind(&tcp_server::__excepter, this, std::placeholders::_1)); } 注意，所有文件描述符都是ET模式，所以：你怎么保证，底层只有一个链接就绪呢？
所以需要while，把这些东西包起来！
void __accepter(connection* conn) { // logMessage(DEBUG, "accepter is called"); // 此时的listensock一定已经就绪了！ // v1 while (true) { std::string client_ip; uint16_t client_port; int accept_errno = 0; int sock = Sock::Accept(conn->__sock, &client_ip, &client_port, &accept_errno); // accept回来的sock就是正常io的sock if (sock < 0) { } // 将sock托管给poll和tcpserver __add_connection(sock, std::bind(&tcp_server::__recver, this, std::placeholders::_1), std::bind(&tcp_server::__sender, this, std::placeholders::_1), std::bind(&tcp_server::__excepter, this, std::placeholders::_1)); } } 因为sock已经被设置成非阻塞了，所以就算一直循环到没有数据，也不会阻塞。
此时，问题来了：如何区分，这个accept的推出是因为读取出错了，还是因为底层已经读完了（没有数据了）呢？所以要加以区分。
所以sock.hpp的accept可以多加上一个参数，表示errno这个码，我们就能在上层区分了
static int Accept(int listensock, std::string* ip, uint16_t* port, int* accept_errno); 所以这种形式才是对的
tcp_server.hpp
void __accepter(connection* conn) { // logMessage(DEBUG, "accepter is called"); // 此时的listensock一定已经就绪了！ // v1 while (true) { std::string client_ip; uint16_t client_port; int accept_errno = 0; int sock = Sock::Accept(conn->__sock, &client_ip, &client_port, &accept_errno); // accept回来的sock就是正常io的sock if (sock < 0) { if (accept_errno == EAGAIN || accept_errno == EWOULDBLOCK) // 并不是出错了，是因为没链接了 break; else if (accept_errno == EINTR) continue; // 概率非常低 else { logMessage(WARNING, "accept error, %d : %s", accept_errno, strerror(accept_errno)); break; } } // 将sock托管给poll和tcpserver __add_connection(sock, std::bind(&tcp_server::__recver, this, std::placeholders::_1), std::bind(&tcp_server::__sender, this, std::placeholders::_1), std::bind(&tcp_server::__excepter, this, std::placeholders::_1)); } } void __recver(connection* conn) { logMessage(DEBUG, "__recver called"); } void __sender(connection* conn) { logMessage(DEBUG, "__sender called"); } void __excepter(connection* conn) { logMessage(DEBUG, "__excepter called"); } 此时进行一个测试
发现__recver的回调成功了！
编写__recver
void __recver(connection* conn) { // 非阻塞读取，所以要循环读取 // v1. 先面向字节流 const int num = 1024; while (true) { char buffer[num]; ssize_t n = recv(conn->__sock, buffer, sizeof(buffer) - 1, 0); if (n < 0) { if (errno == EAGAIN || errno == EWOULDBLOCK) // 读取完毕了(正常的break) break; else if (errno == EINTR) continue; else { logMessage(ERROR, "recv error, %d:%s", errno, strerror(errno)); conn->__except_callback(conn); // 异常了，调用异常回调 break; } } // 读取成功了 buffer[n] = 0; conn->__in_buffer += buffer; // 放到缓冲区里面就行了 } // end while logMessage(DEBUG, "recv done, the [(%d)]inbuffer: %s", conn->__sock, conn->__in_buffer.c_str()); } 和accept其实差不太多，反正读取要while读取，因为是ET非阻塞，要读完。
读取进来的数据，直接丢到缓冲区里面去！如果测试，我们就能看到缓冲区是叠加的！
解耦上层业务
现在我们把报文都读进来了，然后当然我们要分割这些报文，问题是，分割好之后干什么？
如果是web服务器，分割好的都是一个一个http报文，所以我们应该分析http报文，作出对应的回应。
如果是其他服务器，业务逻辑就不同了。因为我们要解耦合，把上层业务逻辑分割出去。
using business_func_t = std::function<void(connection*, std::string request)>; // 上层的业务逻辑 分割出一个一个报文之后，当成放到request里面，传给business_func_t，让他去对这些报文做处理。
所以这个，交给上层处理！上层的代码 web_server.hpp 我就不给大家讲解了，可以直接看代码。
如何响应
当响应构建好之后，要发给客户端，也就是要调用 __sender 了。那么怎么让服务器发送呢？
我们构建ET模式的时候，只关心了读事件，写事件压根儿没被打开，所以现在是发不了的！那怎么发?
两个步骤：
- 需要完整的发送的逻辑
- 要触发发送这个动作
完成发送的逻辑
void __sender(connection* conn) { while (true) { ssize_t n = send(conn->__sock, conn->__out_buffer.c_str(), conn->__out_buffer.size(), 0); if (n > 0) { conn->__out_buffer.erase(0, n); if (conn->__out_buffer.empty()) break; // 发完了 } else { if (errno == EAGAIN || errno == EWOULDBLOCK) break; else if (errno == EINTR) continue; else { logMessage(ERROR, "send error, %d:%s", errno, strerror(errno)); conn->__except_callback(conn); break; } } } // 走到这里，要么就是发完，要么就是发送条件不满足，下次发送 } 和前面一样，不重复解释。
触发发送
一旦开启EPOLLOUT，epoll回立刻自动触发一次事件就绪，如果后续保持发送的开启，epoll会一直发。
所以需要这么一个函数
void enable_read_write(bool readable, bool writable) { } 所以此时：__sender里面要补充。
// 走到这里，要么就是发完，要么就是发送条件不满足，下次发送 if (conn->__out_buffer.empty()) enable_read_write(conn, true, false); 如果此次调用__sender已经把数据发完了，就要关闭写事件！enable_read_write(conn, true, false);。
谁打开的写事件？业务逻辑那里触发！
web_server.hpp
static void Respones(yufc::connection* conn, std::string& request) { // 处理http报文 std::string target = web_server::analyze_http_mesg(request); // 2. 构建一个Http构建一个响应 std::string HttpResponse = web_server::build_http_response_mesg(target); // 3. 放到conn到发送缓冲区中 conn->__out_buffer += HttpResponse; // 4. 调用send conn->__tsvr->enable_read_write(conn, true, true); } __excepter处理
void __excepter(connection* conn) { if (!is_sock_in_map(conn->__sock)) return; // 1. 从epoll中移除 if (!__poll.delete_from_epoll(conn->__sock)) assert(false); // 2. 从map中移除 __connection_map.erase(conn->__sock); // 3. close sock close(conn->__sock); // 4. delete conn delete conn; } 至此，服务器全部搞定！
