Linux网络编程——基于ET模式下的Reactor

一、前言

上篇文章中我们已经讲解了多路转接剩下的两个接口：poll和epoll，并且知道了epoll的两种工作模式分别是 LT模式和ET模式，下来我们就实现的是一个简洁版的 Reactor，即半同步半异步I/O，在linux网络中，最常用最频繁的一种网络IO计划模式
    Reactor计划模式的概念：
Reactor计划模式是一种为处理惩罚并发操纵的事件处理惩罚模式。它将事件的检测和相应分离，使得事件的监听者可以注册对特定类型事件的爱好，并在这些事件发生时得到通知。这种模式非常得当于需要处理惩罚大量并发连接的应用程序，比如Web服务器。
  二、代码

1、首先对于网络来说必须有套接字的创建、绑定、监听和连接，将它们封装起来

1. //Sock.hpp
2. #pragma once
4. #include <iostream>
5. #include <string>
6. #include <cstring>
7. #include <unistd.h>
8. #include <sys/types.h>
9. #include <sys/socket.h>
10. #include <netinet/in.h>
11. #include <arpa/inet.h>
12. #include "Log.hpp"
13. #include "Err.hpp"
15. const static int backlog = 32;
16. const static int defaultsock = -1;
18. class Sock
19. {
20. public:
21.     Sock():listensock_(defaultsock)
22.     {}
23.     ~Sock()
24.     {
25.         if(listensock_ != defaultsock) close(listensock_);
26.     }
27. public:
28.     void Socket()
29.     {
30.         // 1. 创建socket文件套接字对象
31.         listensock_ = socket(AF_INET, SOCK_STREAM, 0);
32.         if (listensock_ < 0)
33.         {
34.             logMessage(FATAL, "create socket error");
35.             exit(SOCKET_ERR);
36.         }
37.         logMessage(NORMAL, "create socket success: %d", listensock_);
39.         int opt = 1;
40.         setsockopt(listensock_, SOL_SOCKET, SO_REUSEADDR|SO_REUSEPORT, &opt, sizeof(opt));
41.     }
43.     void Bind(int port)
44.     {
45.         // 2. bind绑定自己的网络信息
46.         struct sockaddr_in local;
47.         memset(&local, 0, sizeof(local));
48.         local.sin_family = AF_INET;
49.         local.sin_port = htons(port);
50.         local.sin_addr.s_addr = INADDR_ANY;
51.         if (bind(listensock_, (struct sockaddr *)&local, sizeof(local)) < 0)
52.         {
53.             logMessage(FATAL, "bind socket error");
54.             exit(BIND_ERR);
55.         }
56.         logMessage(NORMAL, "bind socket success");
57.     }
59.     void Listen()
60.     {
61.         // 3. 设置socket 为监听状态
62.         if (listen(listensock_, backlog) < 0)
63.         {
64.             logMessage(FATAL, "listen socket error");
65.             exit(LISTEN_ERR);
66.         }
67.         logMessage(NORMAL, "listen socket success");
68.     }
70.     int Accept(std::string *clientip, uint16_t *clientport, int *err)
71.     {
72.         struct sockaddr_in peer;
73.         socklen_t len = sizeof(peer);
74.         int sock = accept(listensock_, (struct sockaddr *)&peer, &len);
75.         *err = errno;
76.         if (sock < 0){}
77.             // logMessage(ERROR, "accept error, next");
78.         else
79.         {
80.             // logMessage(NORMAL, "accept a new link success, get new sock: %d", sock); // ?
81.             *clientip = inet_ntoa(peer.sin_addr);
82.             *clientport = ntohs(peer.sin_port);
83.         }
85.         return sock;
86.     }
87.     int Fd()
88.     {
89.         return listensock_;
90.     }
91.     void Close()
92.     {
93.         if(listensock_ != defaultsock) close(listensock_);
94.     }
95. private:
96.     int listensock_;
97. };

复制代码

2、接着我们还想要将日记文件带进来

1. //日志文件，生成和输出日志信息
2. //Log.hpp
3. #pragma once
5. #include <iostream>
6. #include <string>
7. #include <cstdarg>
8. #include <ctime>
9. #include <unistd.h>
11. #define DEBUG   0
12. #define NORMAL  1
13. #define WARNING 2
14. #define ERROR   3
15. #define FATAL   4
17. const char * to_levelstr(int level)
18. {
19.     switch(level)
20.     {
21.         case DEBUG : return "DEBUG";
22.         case NORMAL: return "NORMAL";
23.         case WARNING: return "WARNING";
24.         case ERROR: return "ERROR";
25.         case FATAL: return "FATAL";
26.         default : return nullptr;
27.     }
28. }
30. void logMessage(int level, const char *format, ...)
31. {
32. #define NUM 1024
33.     char logprefix[NUM];//用于存储日志前缀，包含日志级别，时间戳和进程PID
34.     snprintf(logprefix, sizeof(logprefix), "[%s][%ld][pid: %d]",//格式化输出，并将结果按格式存储到指定的字符数组中，
35.         to_levelstr(level), (long int)time(nullptr), getpid());//这一行是可变数量的参数
37.     char logcontent[NUM];//用于存储日志内容
38.     va_list arg;
39.     va_start(arg, format);//初始化arg指向第一个可选参数
40.     vsnprintf(logcontent, sizeof(logcontent), format, arg);
42.     std::cout << logprefix << logcontent << std::endl;
43. }

复制代码

3、还有对错误代码举行定义

1. //Err.hpp
2. #pragma once
4. #include <iostream>
5. //枚举错误
6. enum
7. {
8.     USAGE_ERR = 1,//参数使用不当
9.     SOCKET_ERR,//创建套接字失败
10.     BIND_ERR,//绑定套接字失败
11.     LISTEN_ERR,//设置监听失败
12.     EPOLL_CREATE_ERR//创建epoll失败
13. };

复制代码

4、我们要知道ET模式下，我们必须将文件描述符设置为非壅闭的，以是我们还需要一个设置文件描述符非壅闭接口

1. #pragma once
3. #include <iostream>
4. #include <unistd.h>
5. #include <fcntl.h>
6. //给文件描述符设置非阻塞模式
7. class Util
8. {
9. public:
10.     static bool SetNonBlock(int fd)
11.     {
12.         int fl = fcntl(fd, F_GETFL);
13.         if (fl < 0) return false;
14.         fcntl(fd, F_SETFL, fl | O_NONBLOCK);
15.         return true;
16.     }
17. };

复制代码

5、接着想在网络中添加的网络计算器的业务逻辑，为了方便对转发的数据大概报文举行完整读取，我们还需要订定传输的协议

1. //制定数据的传输格式，即协议
2. //Protocol.hpp
3. #pragma once
5. #include <iostream>
6. #include <cstring>
7. #include <string>
8. #include <sys/types.h>
9. #include <sys/socket.h>
10. #include <jsoncpp/json/json.h>
12. #define SEP " "
13. #define SEP_LEN strlen(SEP) // 不敢使用sizeof()
14. #define LINE_SEP "\r\n"
15. #define LINE_SEP_LEN strlen(LINE_SEP) // 不敢使用sizeof()
17. enum
18. {
19.     OK = 0,
20.     DIV_ZERO,
21.     MOD_ZERO,
22.     OP_ERROR
23. };
25. // "x op y" -> "content_len"\r\n"x op y"\r\n
26. // "exitcode result" -> "content_len"\r\n"exitcode result"\r\n
27. std::string enLength(const std::string &text)//encode编码，将文本內容加工成上面的形式
28. {
29.     std::string send_string = std::to_string(text.size());
30.     send_string += LINE_SEP;
31.     send_string += text;
32.     send_string += LINE_SEP;
34.     return send_string;
35. }
37. // "content_len"\r\n"exitcode result"\r\n
38. bool deLength(const std::string &package, std::string *text)//decode解码
39. {
40.     auto pos = package.find(LINE_SEP);
41.     if (pos == std::string::npos)//没有找到\r\n
42.         return false;
43.     std::string text_len_string = package.substr(0, pos);//获取文本长度
44.     int text_len = std::stoi(text_len_string);//从字符串转成整型
45.     *text = package.substr(pos + LINE_SEP_LEN, text_len);
46.     return true;
47. }
49. // 没有人规定我们网络通信的时候，只能有一种协议！！
50. // 我们怎么让系统知道我们用的是哪一种协议呢？？
51. // "content_len"\r\n"协议编号"\r\n"x op y"\r\n
53. class Request//请求
54. {
55. public:
56.     Request() : x(0), y(0), op(0)//默认构造函数
57.     {
58.     }
59.     Request(int x_, int y_, char op_) : x(x_), y(y_), op(op_)//构造函数
60.     {
61.     }
62.     // 1. 自己写
63.     // 2. 用现成的
64.     bool serialize(std::string *out)//序列化
65.     {
66. #ifdef MYSELF//自己的序列化
67.         *out = "";
68.         // 结构化 -> "x op y";
69.         std::string x_string = std::to_string(x);
70.         std::string y_string = std::to_string(y);
72.         *out = x_string;
73.         *out += SEP;
74.         *out += op;
75.         *out += SEP;
76.         *out += y_string;
77. #else
78.         Json::Value root;
79.         root["first"] = x;
80.         root["second"] = y;
81.         root["oper"] = op;
83.         Json::FastWriter writer;
84.         // Json::StyledWriter writer;
85.         *out = writer.write(root);
86. #endif
87.         return true;
88.     }
90.     // "x op yyyy";
91.     bool deserialize(const std::string &in)//反序列化
92.     {
93. #ifdef MYSELF
94.         // "x op y" -> 结构化
95.         auto left = in.find(SEP);
96.         auto right = in.rfind(SEP);
97.         if (left == std::string::npos || right == std::string::npos)
98.             return false;
99.         if (left == right)
100.             return false;
101.         if (right - (left + SEP_LEN) != 1)
102.             return false;
104.         std::string x_string = in.substr(0, left); // [0, 2) [start, end) , start, end - start
105.         std::string y_string = in.substr(right + SEP_LEN);
107.         if (x_string.empty())
108.             return false;
109.         if (y_string.empty())
110.             return false;
111.         x = std::stoi(x_string);
112.         y = std::stoi(y_string);
113.         op = in[left + SEP_LEN];
114. #else
115.         Json::Value root;
116.         Json::Reader reader;
117.         reader.parse(in, root);
119.         x = root["first"].asInt();
120.         y = root["second"].asInt();
121.         op = root["oper"].asInt();
122. #endif
123.         return true;
124.     }
126. public:
127.     // "x op y"
128.     int x;
129.     int y;
130.     char op;
131. };
133. class Response
134. {
135. public:
136.     Response() : exitcode(0), result(0)
137.     {
138.     }
139.     Response(int exitcode_, int result_) : exitcode(exitcode_), result(result_)
140.     {
141.     }
142.     bool serialize(std::string *out)
143.     {
144. #ifdef MYSELF
145.         *out = "";
146.         std::string ec_string = std::to_string(exitcode);
147.         std::string res_string = std::to_string(result);
149.         *out = ec_string;
150.         *out += SEP;
151.         *out += res_string;
152. #else
153.         Json::Value root;
154.         root["exitcode"] = exitcode;
155.         root["result"] = result;
157.         Json::FastWriter writer;
158.         *out = writer.write(root);
159. #endif
160.         return true;
161.     }
162.     bool deserialize(const std::string &in)
163.     {
164. #ifdef MYSELF
165.         // "exitcode result"
166.         auto mid = in.find(SEP);
167.         if (mid == std::string::npos)
168.             return false;
169.         std::string ec_string = in.substr(0, mid);
170.         std::string res_string = in.substr(mid + SEP_LEN);
171.         if (ec_string.empty() || res_string.empty())
172.             return false;
174.         exitcode = std::stoi(ec_string);
175.         result = std::stoi(res_string);
176. #else
177.         Json::Value root;
178.         Json::Reader reader;
179.         reader.parse(in, root);
181.         exitcode = root["exitcode"].asInt();
182.         result = root["result"].asInt();
183. #endif
184.         return true;
185.     }
187. public:
188.     int exitcode; // 0：计算成功，!0表示计算失败，具体是多少，定好标准
189.     int result;   // 计算结果
190. };
192. // "content_len"\r\n"x op y"\r\n     "content_len"\r\n"x op y"\r\n"content_len"\r\n"x op
193. //从输入缓冲区 inbuffer 中解析出一个完整的报文，并将其存储在 text 指针指向的位置。
194. bool ParseOnePackage(std::string &inbuffer, std::string *text)
195. {
196.     *text = "";//清空 text 指向的字符串，确保不会残留旧数据。
197.     // 分析处理
198.     auto pos = inbuffer.find(LINE_SEP);//查找 inbuffer 中的第一个 \r\n
199.     if (pos == std::string::npos)//如果找不到，则返回 false 表示当前缓冲区中没有完整的报文头部信息。
200.         return false;
202.     std::string text_len_string = inbuffer.substr(0, pos);//提取从开始到第一个 \r\n 之前的部分作为文本长度字符串。
203.     int text_len = std::stoi(text_len_string);//将该字符串转换为整数 text_len，表示实际消息内容的长度。
204.     int total_len = text_len_string.size() + 2 * LINE_SEP_LEN + text_len;//整个报文的总长度
206.     if (inbuffer.size() < total_len)
207.         return false;
209.     // 至少有一个完整的报文
210.     *text = inbuffer.substr(0, total_len);//读取
211.     inbuffer.erase(0, total_len);//从 inbuffer 中移除已处理的报文部分，以便后续继续处理剩余的数据。
212.     return true;
213. }

复制代码

6、接着封装epoll，包罗它的添加事件、等候事件就绪和其他操纵

2. //Epoller.hpp
3. #pragma once
5. #include <iostream>
6. #include <string>
7. #include <cstring>
8. #include <sys/epoll.h>
9. #include "Err.hpp"
10. #include "Log.hpp"
12. const static int defaultepfd = -1;
13. const static int size = 128;
15. class Epoller
16. {
17. public:
18.     Epoller():epfd_(defaultepfd)
19.     {}
20.     ~Epoller()
21.     {
22.         if(epfd_ != defaultepfd) close(epfd_);
23.     }
24. public:
25.     void Create()
26.     {
27.         epfd_ = epoll_create(size);
28.         if(epfd_ < 0)
29.         {
30.             logMessage(FATAL, "epoll_create error, code: %d, errstring: %s", errno, strerror(errno));
31.             exit(EPOLL_CREATE_ERR);
32.         }
33.     }
34.     // user -> kernel
35.     bool AddEvent(int sock, uint32_t events)
36.     {
37.         struct epoll_event ev;
38.         ev.events = events;
39.         ev.data.fd = sock;
41.         int n = epoll_ctl(epfd_, EPOLL_CTL_ADD, sock, &ev);
42.         return n == 0;
43.     }
44.     // kernel -> user
45.     int Wait(struct epoll_event revs[], int num, int timeout)
46.     {
47.         int n = epoll_wait(epfd_, revs, num, timeout);
48.         return n;
49.     }
50.    
51.     bool Control(int sock, uint32_t event, int action)
52.     {
53.         int n = 0;
54.         if(action == EPOLL_CTL_MOD)
55.         {
56.             struct epoll_event ev;
57.             ev.events = event;
58.             ev.data.fd = sock;
59.             n = epoll_ctl(epfd_, action, sock, &ev);
60.         }
61.         else if(action == EPOLL_CTL_DEL)
62.         {
63.             n = epoll_ctl(epfd_, action, sock, nullptr);
64.         }
65.         else n = -1;
67.         return n == 0;
68.     }
70.     void Close()
71.     {
72.         if(epfd_ != defaultepfd) close(epfd_);
73.     }
74.    
75. private:
76.     int epfd_;
77. };

复制代码

7、接着就是重头戏了，实现封装服务器

1. #pragma once
3. #include <iostream>
4. #include <cassert>
5. #include <functional>
6. #include <unordered_map>
7. #include "Log.hpp"
8. #include "Sock.hpp"
9. #include "Err.hpp"
10. #include "Epoller.hpp"
11. #include "Util.hpp"
12. #include "Protocol.hpp"
14. namespace tcpserver
15. {
16.     class Connection;//声明
17.     class TcpServer;
19.     static const uint16_t defaultport = 8080;
20.     static const int num = 64;
22.     using func_t = std::function<void(Connection *)>;
23.     // using hander_t = std::function<void(const std::string &package)>;
25.     class Connection//封装套接字，使得他们都有着自己的缓冲区空间
26.     {
27.     public:
28.         Connection(int sock, TcpServer *tsp) : sock_(sock), tsp_(tsp)
29.         {
30.         }
31.         void Register(func_t r, func_t s, func_t e)//注册方法，创建该结构体的时候，注册它的三个方法，即数据就绪之后怎么处理
32.         {
33.             recver_ = r;
34.             sender_ = s;
35.             excepter_ = e;
36.         }
37.         ~Connection()
38.         {
39.         }
40.         void Close()
41.         {
42.             close(sock_);
43.         }
45.     public:
46.         int sock_;
47.         std::string inbuffer_;  // 输入缓冲区，但是这里只能处理字符串类信息，图片视频类就难解决了
48.         std::string outbuffer_; // 输出缓冲区，发送也是一样，因为不能保证自己对一个文本发送到哪里了
50.         func_t recver_;   // 从sock_读
51.         func_t sender_;   // 向sock_写
52.         func_t excepter_; // 处理sock_ IO的时候上面的异常事件
54.         TcpServer *tsp_; // 回执指针，可以指向TcpServer，可以被省略
56.         uint64_t lasttime;
57.     };
59.     class TcpServer // Reactor反应堆模式
60.     {
61.     private:
62.         void Recver(Connection *conn)//对于收到的消息如果不做处理，后面再收到时就会看到一直追加在原数据后面
63.         {
64.             conn->lasttime = time(nullptr);
66.             char buffer[1024];
67.             while (true)//循环将文件描述符的东西全部读上来
68.             {
69.                 ssize_t s = recv(conn->sock_, buffer, sizeof(buffer) - 1, 0);
70.                 if (s > 0)
71.                 {
72.                     buffer[s] = 0;
73.                     conn->inbuffer_ += buffer; // 将读到的数据入队列
74.                     logMessage(DEBUG, "\n%s", conn->inbuffer_);
75.                     service_(conn);
76.                 }
77.                 else if (s == 0)
78.                 {
79.                     if (conn->excepter_)
80.                     {
81.                         conn->excepter_(conn);//如果此时异常了，且异常被设置了，此时只需要回调它的异常就可以了
82.                         return;
83.                     }
84.                 }
85.                 else
86.                 {
87.                     if (errno == EAGAIN || errno == EWOULDBLOCK)//底层没数据了
88.                         break;
89.                     else if (errno == EINTR)//信号被中断了
90.                         continue;
91.                     else//出错了
92.                     {
93.                         if (conn->excepter_)
94.                         {
95.                             conn->excepter_(conn);
96.                             return;
97.                         }
98.                     }
99.                 }
100.             } // while
101.         }
102.         void Sender(Connection *conn)
103.         {
104.             conn->lasttime = time(nullptr);
105.             while (true)
106.             {
107.                 ssize_t s = send(conn->sock_, conn->outbuffer_.c_str(), conn->outbuffer_.size(), 0);
108.                 if (s > 0)
109.                 {
110.                     if (conn->outbuffer_.empty())//发完了
111.                     {
112.                         // EnableReadWrite(conn, true, false);
113.                         break;
114.                     }
115.                     else//没有发完，发出去多少清除多少
116.                         conn->outbuffer_.erase(0, s);
117.                 }
118.                 else
119.                 {
120.                     if (errno == EAGAIN || errno == EWOULDBLOCK)//发送缓冲区满了
121.                         break;
122.                     else if (errno == EINTR)//被信号中断
123.                         continue;
124.                     else//发送出错
125.                     {
126.                         if (conn->excepter_)
127.                         {
128.                             conn->excepter_(conn);
129.                             return;
130.                         }
131.                     }
132.                 }
133.             }
134.             // 如果没有发送完毕，需要对对应的sock开启对写事件的关系， 如果发完了，我们要关闭对写事件的关心！
135.             if (!conn->outbuffer_.empty())
136.                 conn->tsp_->EnableReadWrite(conn, true, true);
137.             else
138.                 conn->tsp_->EnableReadWrite(conn, true, false);
139.         }
140.         void Excepter(Connection *conn)
141.         {
142.             logMessage(DEBUG, "Excepter begin");
143.             epoller_.Control(conn->sock_, 0, EPOLL_CTL_DEL);
144.             conn->Close();
145.             connections_.erase(conn->sock_);
147.             logMessage(DEBUG, "关闭%d 文件描述符的所有的资源", conn->sock_);
149.             delete conn;
150.         }
152.         void Accepter(Connection *conn)
153.         {
154.             for (;;)//因为监听套接字已经被设置为ET了，所以无论来多少连接也只是通知一次，这就倒逼程序员一次拿完所有的连接，所以我们需要循环读取，直到失败
155.             {
156.                 std::string clientip;
157.                 uint16_t clientport;
158.                 int err = 0;
159.                 int sock = sock_.Accept(&clientip, &clientport, &err);
160.                 if (sock > 0)
161.                 {
162.                     AddConnection(
163.                         sock, EPOLLIN | EPOLLET,
164.                         std::bind(&TcpServer::Recver, this, std::placeholders::_1),
165.                         std::bind(&TcpServer::Sender, this, std::placeholders::_1),
166.                         std::bind(&TcpServer::Excepter, this, std::placeholders::_1));
168.                     logMessage(DEBUG, "get a new link, info: [%s:%d]", clientip.c_str(), clientport);
169.                 }
170.                 else//当底层没有连接时，使用accept拿的时候，错误码会被设置
171.                 {
172.                     if (err == EAGAIN || err == EWOULDBLOCK)//表示底层没有连接了
173.                         break;
174.                     else if (err == EINTR)//表示正在读的时候被信号中断了
175.                         continue;
176.                     else//这才是真正出错了
177.                         break;
178.                 }
179.             }
180.         }
181.         
182.         void AddConnection(int sock, uint32_t events, func_t recver, func_t sender, func_t excepter)//管理Connection对象
183.         {
184.             // 1. 首先要为该sock创建Connection，并初始化，并添加到connections_
185.             if (events & EPOLLET)//如果事件设置了ET模式，那么就将该套接字设置为非阻塞
186.                 Util::SetNonBlock(sock);
187.             Connection *conn = new Connection(sock, this); //构建对象
188.             // 2. 给对应的sock设置对应回调处理方法
189.             conn->Register(recver, sender, excepter);
191.             // 2. 其次将sock与它要关心的事件"写透式"注册到epoll中，让epoll帮我们关心
192.             bool r = epoller_.AddEvent(sock, events);//一般这里是不会出什么问题的
193.             assert(r);//断言一下
194.             (void)r;
196.             // 3. 将kv添加到connections_
197.             connections_.insert(std::pair<int, Connection *>(sock, conn));
199.             logMessage(DEBUG, "add new sock : %d in epoll and unordered_map", sock);
200.         }
201.         bool IsConnectionExists(int sock)//判断文件描述符是否在我们对应的connection中
202.         {
203.             auto iter = connections_.find(sock);
204.             return iter != connections_.end();
205.         }
206.         void Loop(int timeout)
207.         {
208.             int n = epoller_.Wait(revs_, num_, timeout); // 获取已经就绪的事件
209.             for (int i = 0; i < n; i++)
210.             {
211.                 int sock = revs_[i].data.fd;
212.                 uint32_t events = revs_[i].events;
214.                 // 将所有的异常问题，全部转化 成为读写问题
215.                 if (events & EPOLLERR)
216.                     events |= (EPOLLIN | EPOLLOUT);//如果出现异常，就设置它的读事件和写事件就绪，读写本来就要做异常处理
217.                 if (events & EPOLLHUP)
218.                     events |= (EPOLLIN | EPOLLOUT);
219.                 // listen事件就绪
220.                 if ((events & EPOLLIN) && IsConnectionExists(sock) && connections_[sock]->recver_)//读事件就绪并且connection对应的套接字存在，并且recver对象也存在
221.                     connections_[sock]->recver_(connections_[sock]);
222.                 if ((events & EPOLLOUT) && IsConnectionExists(sock) && connections_[sock]->sender_)
223.                     connections_[sock]->sender_(connections_[sock]);
224.             }
225.         }
227.     public:
228.         TcpServer(func_t func, uint16_t port = defaultport) : service_(func), port_(port), revs_(nullptr)
229.         {
230.         }
231.         void InitServer()
232.         {
233.             // 1. 创建socket
234.             sock_.Socket();
235.             sock_.Bind(port_);
236.             sock_.Listen();
237.             // 2. 构建Epoll
238.             epoller_.Create();
239.             // 3. 将目前唯一的一个sock，添加到epoller中， 之前需要先将对应的fd设置成为非阻塞
240.       
241.             // listensock_也是一个socket啊，也要看做成为一个Connection
242.             AddConnection(sock_.Fd(), EPOLLIN | EPOLLET,
243.                           std::bind(&TcpServer::Accepter, this, std::placeholders::_1), nullptr, nullptr);
245.             revs_ = new struct epoll_event[num];//初始化一段缓冲区用来保存已经就绪的事件
246.             num_ = num;
247.         }
248.         void EnableReadWrite(Connection *conn, bool readable, bool writeable)
249.         {
250.             uint32_t event = (readable ? EPOLLIN : 0) | (writeable ? EPOLLOUT : 0) | EPOLLET;
251.             epoller_.Control(conn->sock_, event, EPOLL_CTL_MOD);
252.         }
253.         // 事件派发器
254.         void Dispatcher()
255.         {
256.             int timeout = 1000;
257.             while (true)
258.             {
259.                 Loop(timeout);
260.                 // logMessage(DEBUG, "time out ...");
262.                 // 遍历connections_，计算每一个链接的已经有多长时间没有动了
263.                
264.             }
265.         }
266.         ~TcpServer()
267.         {
268.             sock_.Close();
269.             epoller_.Close();
270.             if (nullptr == revs_)
271.                 delete[] revs_;
272.         }
274.     private:
275.         uint16_t port_;
276.         Sock sock_;
277.         Epoller epoller_;
278.         std::unordered_map<int, Connection *> connections_;
279.         struct epoll_event *revs_;//存储多个就绪时间的空间，指针
280.         int num_;//表示用来保存就绪事件的缓冲区一共有多少个元素
281.         // hander_t handler_;
282.         func_t service_;
283.     };
284. }

复制代码

  这里有着很多细节，需要强调点如下：
  就读事件来说，在我们之前实现无论是select、poll还是epoll的时候，对底层数据到位之后，读取时候的处理惩罚都是特殊不完善的。先看我们之前怎么处理惩罚的，如下：

1. else if(revs[i].events&EPOLLIN)//读事件就绪
2.             {
3.                 char buffer[64];
4.                 ssize_t s=recv(fd,buffer,sizeof(buffer)-1,0);
5.                 if(s>0)//读取成功
6.                 {
7.                     buffer[s]='\0';
8.                     std::cout<<"echo#"<<buffer<<std::endl;
9.                 }
10.                 else if(s==0)//对端连接关闭
11.                 {
12.                     std::cout<<"Client quit"<<std::endl;
13.                     close(fd);
14.                     DelEvent(fd);//将文件描述符从模型中删除
15.                 }
16.                 else{//读取错误
17.                     std::cerr<<"recv error!"<<std::endl;
18.                     close(fd);
19.                     DelEvent(fd);
20.                 }
21.             }

复制代码

         我们要知道在epoll模式下，如果底层数据已经就绪，我们假设把本轮的数据全部读完，这时候也不一定可以或许读到一个完整的请求，本轮读完是ET的要求，能不能完整读完是跟协议要求干系的。纵然循环读，读完也不行。如果我们并不能读到一个完整的请求，那么我们就只能 将自己读到的数据临时放在buffer中，但是临时生存在buffer中的话，这个文件描述符生存了，其他的文件描述符数据也就绪了怎么办，而且在当前读完之后，整个的这个代码区间全部都被开释掉了(因为它是栈上的空间)，以是再想下一次读的时候buffer早都被开释完了。
          以是光在栈上的缓冲区远远不够，对于每一个文件描述符，我们都得给他们的输入输出的用户层缓冲区全都带上。以是要对每一个套接字举行封装，每一个套接字都要有着自己的缓冲区空间，读取的时候，将数据暂存到自己的缓冲区中，没读完下次再读，这也才不会和其他的数据胶葛。在发送的时候也是同理。
          我们在这次的Reactor代码中对每一个文件描述符都封装了Connection，系统中出现大量的Connection对象，需要被管理起来，先描述再组织，描述已经到位就是Connection结构体，接下来用unordered_map<int ,Connection*>管理起来,，纵然用kv将文件描述符和我们所创建的Connection结构体结合起来  ，  以是将来我们将文件描述符添加到epoll模型中的时候，同时将文件描述符的Connect对象也添加进unordered_map中，这样将来就可以通过unordered_map检察该文件描述符所对应的各种事件和输入输出缓冲区了。
  8、Main函数

1. //Main.cc
2. #include "TcpServer.hpp"
3. #include <memory>
5. using namespace tcpserver;
7. static void usage(std::string proc)
8. {
9.     std::cerr << "Usage:\n\t" << proc << " port"
10.               << "\n\n";
11. }
13. bool cal(const Request &req, Response &resp)
14. {
15.     // req已经有结构化完成的数据啦，你可以直接使用
16.     resp.exitcode = OK;
17.     resp.result = OK;
19.     switch (req.op)
20.     {
21.     case '+':
22.         resp.result = req.x + req.y;
23.         break;
24.     case '-':
25.         resp.result = req.x - req.y;
26.         break;
27.     case '*':
28.         resp.result = req.x * req.y;
29.         break;
30.     case '/':
31.     {
32.         if (req.y == 0)
33.             resp.exitcode = DIV_ZERO;
34.         else
35.             resp.result = req.x / req.y;
36.     }
37.     break;
38.     case '%':
39.     {
40.         if (req.y == 0)
41.             resp.exitcode = MOD_ZERO;
42.         else
43.             resp.result = req.x % req.y;
44.     }
45.     break;
46.     default:
47.         resp.exitcode = OP_ERROR;
48.         break;
49.     }
51.     return true;
52. }
54. void calculate(Connection *conn)
55. {
56.     std::string onePackage;
57.     while (ParseOnePackage(conn->inbuffer_, &onePackage))
58.     {
59.         std::string reqStr;
60.         if (!deLength(onePackage, &reqStr))
61.             return;
62.         std::cout << "去掉报头的正文：\n"
63.                   << reqStr << std::endl;
65.         // 2. 对请求Request，反序列化
66.         // 2.1 得到一个结构化的请求对象
67.         Request req;
68.         if (!req.deserialize(reqStr))
69.             return;
71.         Response resp;
72.         cal(req, resp);
74.         std::string respStr;
75.         resp.serialize(&respStr);
76.         // 5. 然后我们在发送响应
77.         // 5.1 构建成为一个完整的报文
78.         conn->outbuffer_ += enLength(respStr);//添加到自己的发送缓冲区中
80.         std::cout << "--------------result: " << conn->outbuffer_ << std::endl;
81.     }
82.     // 直接发
83.     if (conn->sender_)
84.         conn->sender_(conn);
86.     // // 如果没有发送完毕，需要对对应的sock开启对写事件的关系， 如果发完了，我们要关闭对写事件的关心！
87.     // if (!conn->outbuffer_.empty())
88.     //     conn->tsp_->EnableReadWrite(conn, true, true);
89.     // else
90.     //     conn->tsp_->EnableReadWrite(conn, true, false);
91. }
93. int main(int argc, char *argv[])
94. {
95.     if (argc != 2)
96.     {
97.         usage(argv[0]);
98.         exit(USAGE_ERR);
99.     }
100.     uint16_t port = atoi(argv[1]);
102.     std::unique_ptr<TcpServer> tsvr(new TcpServer(calculate, port));
103.     tsvr->InitServer();
104.     tsvr->Dispatcher();
106.     return 0;
107. }

复制代码

这就是全部代码了。

---

感谢阅读！ 

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。