【项目日志】仿mudou的高并发服务器 --- 实现缓冲区模块，通用范例Any模块 ...

   一个人知道本身为什么而活，      就可以忍受任何一种生活。        --- 尼采 ---       

---

   ✨✨✨项目地点在这里 ✨✨✨
  ✨✨✨https://gitee.com/penggli_2_0/TcpServer✨✨✨
  

---

  
1 主从Reactor模子

这个项目标目标是实现一个可以高效处理哀求的服务器，那么对于如许的一个服务器要如何实现呢？
这里接纳主从Reactor模子的方案：

• 主Reactor模子负责对监听套接字进行管理，进行获取连接操作。
  
• 附属Reactor负责对连接的数据进行处理，而且为了保证线程安全，每一个附属Reactor都要绑定一个线程，这个连接的使命都在这个线程中完成。
  

简单来说就是如许的一个模子：

此中管理连接的对象是Connection对象，这个类是高并发服务器的焦点部门，这里包含了对套接字的处理，事件等待等一系列操作！而Connection对象中对于这些操作的管理也要通过其他的对象来进行！
主要的工作就是搭建起主从Reactor模子，但是这个模子不是一下子就可以写出来的。为了实现主从Reactor模子，我们先需要实现一些基础类，对基础功能进封装，然后对这些功能进行整合，最终实现主从Reactor模子！
2 基础功能封装

2.1 缓冲区 Buffer模块

Buffer模块是⼀个缓冲区模块，用于实现通信中用户态的接收缓冲区和发送缓冲区功能：

• 需要支持字符串读取/写入
  
• 需要支持char*缓冲区读取/写入
  
• 需要正常按行读取 — 方便解析http哀求
  

Buffer模块成员变量很简单：

• vector<char>容器_buffer： 对内存空间进行管理
  
• uint64_t _reader_idx 读偏移：进行读取位置在_buffer容器中的偏移量，即读取的起始位置。
  
• uint64_t _writer_idx 写偏移：下一次写入位置在_buffer容器中的偏移量，即写入的起始位置。
  

接下来实现一下基础功能：

• 构造函数：初始化读/写偏移为0 ，容器_buffer初始化一个巨细 BUFFER_DEFAULT_SIZE。
  
• 获取当前写入起始地点：_buffer空间的起始地点加入写偏移量即写入起始地点。
  
• 获取当前读取起始地点： _buffer空间的起始地点加入读偏移量即读取起始地点。
  
• 获取缓冲区末尾空闲空间巨细：写偏移之后的空闲空间 ，总体空间巨细减去写偏移就是写偏移之后的空间巨细。
  
• 获取缓冲区起始空闲空间巨细：读偏移之前的空闲空间，其实就是读偏移的巨细。
  
• 获取可读数据巨细：写偏移减去读偏移就就之间可读空间的巨细！
  
• 读/写偏移向后移动：
  * 先根据len判断是否小于可读数据巨细 len必须小于可读数据巨细，然后移动读偏移。
  * 向后移动必须小于当前后边的空闲空间巨细，写入数据必须小于缓冲区剩余空间巨细，不敷就进行扩容！
  
• 确保可写空间足够 :
  * 末尾空闲空间足够 直接返回。
  * 末尾空间不敷，但算上起始位置的空闲空间巨细足够 ，将数据移动到起始位置。
  * 如果总空间不敷 ，进行扩容，扩容到足够空间即可
  
• 写入数据：首先保证有足够空间，然后将数据数据拷贝进去。可以继续计划出针对string的写入、针对Buffer的写入以及写入 + 压入数据
  
• 读取数据：首先读取巨细len必须小于可读取数据巨细，然后拷贝数据出来。同样可以计划出针对string的读取、针对Buffer的读取以及读取+弹出数据。
  
• 清空缓冲区：将偏移量归零即可！
  
• 读取一行数据：先找到换行符，然后进行读取。
  

1. // 缓冲区Buffer类
2. class Buffer
3. {
4. private:
5.     std::vector<char> _buffer;
6.     uint64_t _reader_idx;
7.     uint64_t _writer_idx;
9. private:
10.     char *Begin() { return &*_buffer.begin(); }
12. public:
13.     Buffer() : _buffer(BUFFER_DEFAULT_SIZE), _reader_idx(0), _writer_idx(0) {}
14.     // 获取当前写入起始地址
15.     char *WritePos() { return Begin() + _writer_idx; }
16.     // 获取当前读取起始地址
17.     char *ReadPos() { return Begin() + _reader_idx; }
18.     // 获取读取位置之前的空闲空间
19.     uint64_t HeadIdleSize() { return _reader_idx; }
20.     // 获取写入位置之后的空闲空间
21.     uint64_t TailIdleSize() { return _buffer.size() - _writer_idx; }
22.     // 获取可读数据大小
23.     uint64_t ReadAbleSize() { return _writer_idx - _reader_idx; }
24.     // 读/写偏移移动
25.     void MoveReadOffset(uint64_t len)
26.     {
27.         if (len <= 0)
28.             return;
29.         assert(len <= ReadAbleSize());
30.         _reader_idx += len;
31.     }
32.     void MoveWriteOffset(uint64_t len)
33.     {
34.         if (len <= 0)
35.             return;
36.         assert(len <= TailIdleSize());
37.         _writer_idx += len;
38.     }
39.     // 确保可写空间足够 --- 整体空闲空间足够了就移动数据，否则进行扩容
40.     void EnsureWriteSpace(uint64_t len)
41.     {
42.         // 当len小于写入位置之后的空闲空间 直接进行写入
43.         if (len <= TailIdleSize())
44.         {
45.             return;
46.         }
48.         // len 小于总共的空闲空间
49.         else if (len <= TailIdleSize() + HeadIdleSize())
50.         {
51.             // 记录总共的数据
52.             uint64_t sz = ReadAbleSize();
53.             // 移动数据
54.             std::copy(ReadPos(), ReadPos() + sz, Begin());
55.             // 更新写入读取位置
56.             _reader_idx = 0;
57.             _writer_idx = sz;
58.         }
59.         // 需要扩容
60.         else
61.         {
62.             // 在写入位置之后扩充len个大小
63.             _buffer.resize(_writer_idx + len);
64.         }
65.     }
66.     // ----------写入数据------------
67.     void Write(const void *data, uint64_t len)
68.     {
69.         // 写入的数据不能超过可写的总空间
70.         // 确保可以正常写入
71.         EnsureWriteSpace(len);
72.         // 进行拷贝
73.         const char *d = reinterpret_cast<const char *>(data);
74.         std::copy(d, d + len, WritePos());
75.     }
76.     // 写入Buffer
77.     void WriteBuffer(Buffer &buffer)
78.     {
79.         // 直接调用Write
80.         return Write(buffer.ReadPos(), buffer.ReadAbleSize());
81.     }
82.     // 写入字符串
83.     void WriteString(const std::string &str)
84.     {
85.         // 直接调用Write
86.         return Write(&str[0], str.size());
87.     }
88.     // 写入 + 偏移
89.     void WriteAndPush(const void *data, uint64_t len)
90.     {
91.         if (len <= 0)
92.             return;
93.         // 进行写入
94.         Write(data, len);
95.         // 更新写入偏移量
96.         MoveWriteOffset(len);
97.     }
98.     void WriteStringAndPush(const std::string &str)
99.     {
101.         // 直接调用Write
102.         WriteString(str);
103.         LOG(DEBUG, "%s\n", WritePos());
104.         // 更新偏移量
105.         MoveWriteOffset(str.size());
106.         LOG(DEBUG, "当前可读数据大小：%ld\n", ReadAbleSize());
107.     }
108.     void WriteBufferAndPush(Buffer &buffer)
109.     {
110.         // 直接调用Write
111.         Write(buffer.ReadPos(), buffer.ReadAbleSize());
112.         MoveWriteOffset(buffer.ReadAbleSize());
113.     }
115.     //---------读取数据----------
116.     void Read(void *buf, uint64_t len)
117.     {
118.         // len 必须小于可读数据大小
119.         assert(len <= ReadAbleSize());
120.         // 进行拷贝
121.         std::copy(ReadPos(), ReadPos() + len, (char *)buf);
122.     }
123.     // 读取字符串
124.     std::string ReadAsString(uint64_t len)
125.     {
126.         // len 必须小于可读数据大小
127.         assert(len <= ReadAbleSize());
128.         std::string str;
129.         str.resize(len);
130.         // 直接调用Read
131.         Read(&str[0], len);
132.         return str;
133.     }
134.     // 读取 + Pop
135.     void ReadAndPop(void *buf, uint64_t len)
136.     {
137.         // 进行读取
138.         Read(buf, len);
139.         // 更新读取偏移量
140.         MoveReadOffset(len);
141.     }
142.     std::string ReadAsStringAndPop(uint64_t len)
143.     {
144.         // len 必须小于可读数据大小
145.         assert(len <= ReadAbleSize());
146.         // 进行读取
147.         std::string str = ReadAsString(len);
148.         // 更新偏移量
149.         MoveReadOffset(len);
150.         return str;
151.     }
153.     //-----------读取一行数据--------------
154.     char *FindCRLF()
155.     {
156.         return (char *)memchr(ReadPos(), '\n', ReadAbleSize());
157.     }
158.     std::string GetLine()
159.     {
160.         // 先寻找换行符
161.         char *pos = FindCRLF();
162.         if (pos == nullptr)
163.             return "";
164.         // 读取要带走'\n'
165.         std::string str = ReadAsString(pos - ReadPos() + 1);
166.         return str;
167.     }
168.     std::string GetLineAndPop()
169.     {
170.         std::string str = GetLine();
171.         MoveReadOffset(str.size());
172.         return str;
173.     }
175.     // 清除数据
176.     void Clear()
177.     {
178.         // 偏移量归零即可
179.         _reader_idx = 0;
180.         _writer_idx = 0;
181.     }
182. };

复制代码

2.2 通用范例 Any类

每⼀个Connection对连接进⾏管理，最终都不可避免需要涉及到应⽤层协议的处理，因此在Connection中需要设置协议处理的上下⽂来控制处理节奏。但是应⽤层协议千千万，为了降低耦合度，这个协议接收解析上下文就不能有显着的协议倾向，它可以是恣意协议的上下文信息，因此就需要⼀个通⽤的范例来保存各种差别的数据结构：

• 一个连接必须拥有一个哀求接收与解析的上下文！
  
• 上下文的范例大概结构不能固定！因为服务器的协议支持的协议许多，差别的协议，可能都有差别的上下文结构！
  

所以必须拥有一个容器，能够保存各种差别的范例！那么就要实现一个any类

• 假如使用模版类方法，那么实例化对象的时间肯定要指明容器保存的数据范例！而我们需要的是any可以接收恣意范例Any a ; a = 10 ; a = "abc"!
  
• 但是可以嵌套一下，在Any类中计划一个类，专门用于保存各种范例的数据，而Any类保存的是固定类的对象。
  
• 对于这个固定类依旧不能使用模版。但这里这里可以采用多态，计划一个子类，这是一个模版类。
  如许可以通过父类指针读取子类数据！
  
• Any类中，保存的是holder类的指针，当Any类需要保存一个数据时，只需要提供placeholder子类实例化一个
  特定范例的对象出来，让子类对象保存数据！
  

1. // 通用类型 Any类
2. class Any
3. {
4. private:
5.     class holder
6.     {
7.     public:
8.         holder() {}
9.         virtual ~holder() {};
10.         virtual const std::type_info &type() = 0;
11.         virtual holder *clone() = 0;
12.     };
13.     template <class T>
14.     class placeholder : public holder
15.     {
16.     public:
17.         placeholder(const T &val) : _val(val) {}
19.         virtual const std::type_info &type() override
20.         {
21.             return typeid(T);
22.         }
23.         virtual holder *clone() override
24.         {
25.             return new placeholder(_val);
26.         }
28.         T _val;
29.     };
30.     // 通过这个父类指针访问子类的成员
31.     holder *_content = nullptr;
33. public:
34.     Any()
35.     {
36.     }
37.     template <class T>
38.     // 拷贝构造
39.     Any(const T &val) : _content(new placeholder<T>(val))
40.     {
41.     }
42.     // 拷贝构造
43.     Any(const Any &other) : _content(other._content ? other._content->clone() : nullptr) {}
44.     // 交换数据
45.     Any &swap(Any &other)
46.     {
47.         std::swap(_content, other._content);
48.         return *this;
49.     }
51.     // 赋值重载
52.     Any &operator=(Any &other)
53.     {
54.         // 根据 other建立一个临时对象 ，进行资源的交换
55.         Any(other).swap(*this);
56.         return *this;
57.     }
58.     template <class T>
59.     Any &operator=(const T &val)
60.     {
61.         // 根据 val 建立一个临时对象，进行资源的交换
62.         Any(val).swap(*this);
63.         return *this;
64.     }
65.     template <class T>
66.     T *Get()
67.     {
68.         if (typeid(T) != _content->type())
69.             return nullptr;
70.         return &((reinterpret_cast<placeholder<T> *>(_content))->_val);
71.     }
73.     ~Any()
74.     {
75.         delete _content;
76.     }
77. };

复制代码

2.3 套接字 Socket模块

Socket模块是对套接字操作封装的⼀个模块，主要实现的socket的各项操作。是连接模块Connection 与监听模块Accpter的基础！对于套接字的操作我们已经在熟悉不过了：

• 构造函数 析构函数
  
• 创建套接字
  bool Create()
  
• 绑定地点信息
  
• 开始监听
  
• 向服务器发起连接
  
• 获取新连接
  
• 接收数据
  
• 发送数据
  
• 关闭套接字
  
• 创建一个服务端连接：首先创建套接字，然后将历程绑定地点信息，开启监听状态，留意设置为非阻塞读取（为了配合多路转接IO，不需要IO接口进行等待，统统等待都由多路转接负责）；同时启动地点重用 ，保护客户端
  
• 创建一个客户端连接：创建套接字，连接服务器。
  
• 设置套接字选项 — 开启地点端口重用
  
• 设置套接字阻塞属性 — 设置为非阻塞读取
  

1. // 套接字Socket类
2. class Socket
3. {
4. private:
5.     int _sockfd; // 套接字文件描述符
7. private:
8.     // 创建套接字
9.     bool Create()
10.     {
11.         // int socket(int domain, int type, int protocol);
12.         // IPV4     数据流IO
13.         _sockfd = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
14.         if (_sockfd < 0)
15.         {
16.             LOG(ERROR, "create socket failed!\n");
17.             return false;
18.         }
19.         LOG(INFO, "Sockfd:%d create success \n", _sockfd);
20.         return true;
21.     }
22.     // 绑定地址信息
23.     bool bind(const std::string &ip, uint16_t port)
24.     {
25.         struct sockaddr_in addr;
26.         addr.sin_family = AF_INET;                    // 使用IPv4版本地址
27.         addr.sin_port = htons(port);                  // 主机端口转网络端口
28.         addr.sin_addr.s_addr = inet_addr(ip.c_str()); // 将ip字符串转化为网络ip
29.         // 进行绑定
30.         socklen_t len = sizeof(struct sockaddr_in);
31.         int n = ::bind(_sockfd, (struct sockaddr *)&addr, len);
32.         if (n < 0)
33.         {
34.             LOG(ERROR, "bind failed!\n");
35.             return false;
36.         }
37.         return true;
38.     }
39.     // 建立监听套接字
40.     bool Listen(int backlog = MAX_LIETENSIZE)
41.     {
42.         //                        全连接队列的大小
43.         int n = ::listen(_sockfd, backlog);
44.         if (n < 0)
45.         {
46.             LOG(ERROR, "listen failed!\n");
47.             return false;
48.         }
49.         return true;
50.     }
51.     // 向服务器发起连接
52.     bool Connect(const std::string &ip, uint16_t port)
53.     {
54.         struct sockaddr_in addr;
55.         addr.sin_family = AF_INET;                    // 使用IPv4版本地址
56.         addr.sin_port = htons(port);                  // 主机端口转网络端口
57.         addr.sin_addr.s_addr = inet_addr(ip.c_str()); // 将ip字符串转化为网络ip
58.         // 进行绑定
59.         socklen_t len = sizeof(struct sockaddr_in);
60.         int n = ::connect(_sockfd, (struct sockaddr *)&addr, len);
61.         if (n < 0)
62.         {
63.             LOG(ERROR, "connect failed!\n");
64.             return false;
65.         }
66.         return true;
67.     }
69.     // 非阻塞启动地址重用
70.     void ReuseAddress(uint16_t port, const std::string &ip)
71.     {
72.         int val = 1;
73.         socklen_t len = sizeof(int);
74.         // 将端口号设置为可重用
75.         if (setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, (void *)&val, len) == -1)
76.         {
77.             throw std::runtime_error("Failed to set SO_REUSEPORT");
78.         }
80.         // 将IP地址设置为可重用
81.         if (setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, (void *)&val, len) == -1)
82.         {
83.             throw std::runtime_error("Failed to set SO_REUSEADDR");
84.         }
85.     }
86.     void NonBlock()
87.     {
88.         int flag = ::fcntl(_sockfd, F_GETFL, 0);
89.         ::fcntl(_sockfd, F_SETFL, flag | O_NONBLOCK);
90.     }
92. public:
93.     // 构造函数
94.     Socket() : _sockfd(-1) {};
95.     Socket(int sockfd) : _sockfd(sockfd) {}
96.     // 析构函数
97.     ~Socket() { Close(); }
98.     // 返回套接字文件描述符
99.     int Sockfd() { return _sockfd; }
101.     // 接收数据
102.     ssize_t Recv(void *buf, size_t len, int flag = 0)
103.     {
104.         ssize_t n = ::recv(_sockfd, buf, len, flag);
105.         if (n <= 0)
106.         {
107.             // 被信号中断         没有数据了- 非阻塞读取完毕
108.             if (errno == EINTR || errno == EAGAIN)
109.                 return 0; // 没有读取到数据
111.             LOG(ERROR, "recv failed!\n");
112.             return -1;
113.         }
114.         return n; // 实际发送的长度
115.     }
116.     ssize_t NonBlockRecv(void *buf, size_t len)
117.     {
118.         //       MSG_DONTWAIT  非阻塞式读取
119.         if (len <= 0)
120.             return 0;
121.         return Recv(buf, len, MSG_DONTWAIT);
122.     }
123.     // 发送数据
124.     ssize_t Send(void *buf, size_t len, int flag = 0)
125.     {
126.         if (len <= 0)
127.             return 0;
128.         ssize_t n = ::send(_sockfd, buf, len, flag);
129.         if (n <= 0)
130.         {
131.             // 被信号中断         没有数据了- 非阻塞读取完毕
132.             if (errno == EINTR || errno == EAGAIN)
133.                 return 0; // 没有读取到数据
134.             LOG(ERROR, "send failed!\n");
135.             return -1;
136.         }
137.         LOG(DEBUG, "Send :%s\n", (char *)buf);
138.         return n; // 实际发送的数据大小
139.     }
140.     ssize_t NonBlockSend(void *buf, size_t len)
141.     {
142.         //       MSG_DONTWAIT  非阻塞式发送
143.         if (len <= 0)
144.             return 0;
145.         return Send(buf, len, MSG_DONTWAIT);
146.     }
148.     // 创建服务端套接字
149.     bool CreateServer(uint16_t port, const std::string &ip = "0.0.0.0", bool flag = 0)
150.     {
152.         // 创建套接字
153.         if (Create() == false)
154.             return false;
155.         // 设置为非阻塞
156.         if (flag)
157.             NonBlock();
158.         // 将地址与端口设置为可重用
159.         ReuseAddress(port, ip);
160.         // 绑定地址信息
161.         if (bind(ip, port) == false)
162.             return false;
163.         // 进行监听
164.         if (Listen() == false)
165.             return false;
167.         return true;
168.     }
169.     // 创建客户端套接字
170.     bool CreateClient(uint16_t port, const std::string &ip)
171.     {
172.         // 创建套接字
173.         if (Create() == false)
174.             return false;
175.         // 连接服务器
176.         if (Connect(ip, port) == false)
177.             return false;
178.         return true;
179.     }
180.     // 服务器获取连接
181.     int Accept()
182.     {
183.         int newfd = ::accept(_sockfd, nullptr, nullptr);
184.         if (newfd < 0)
185.         {
186.             LOG(ERROR, "accept failed!\n");
187.             return -1;
188.         }
189.         return newfd;
190.     }
191.     //  关闭套接字
192.     void Close()
193.     {
194.         LOG(INFO, "Close sockfd: %d\n", _sockfd);
195.         if (_sockfd != -1)
196.         {
197.             ::close(_sockfd);
198.         }
199.     }
200. };

复制代码

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