从零造的一个网络轮子
从零造的一个网络轮子
- 序 大概是一些碎碎念
- 一 类图及各类简述
- 二 各个类详述及部分代码
- 2.1.main() & FairySunOfNetBaseStart() & FairySunOfNetBaseOver() & 两个辅助函数
- 2.2.CServiceNoBlock
- 2.3.CServiceEpoll
- 2.4.CClientLinkManage
- 2.5.CClientLink
- 2.6.CSocketObj
- 2.7.CSecondBuffer
- 2.8.CThreadManage
- 2.9.CLog & enum EMsgLevel & struct CLogMsg & CDoubleBufLogQueue
- 2.10.union CObj & CObjectPool CMemoryPool
- 2.11.CTime
- 2.12.Barrier
- 2.13.struct CNetMsgHead
- 结束
过去一年了，这玩意陆陆续续改了不少次，填了不少坑
比如说改变了线程的启动方式，一开始使用一个全局的单例来管理各个线程，现在是各个线程的启动由自身对象所持有的线程池启动，原来的方法会导致一些问题，
还有日志的处理，思考是要不要单例类，最后还是选择了单例，一个进程写一个日志文件应该没问题吧
那个丑陋的CThreadManager也被注释掉了
还有一些细小的改动，头文件的相互包含或者其他？记不太清了
不过主体没什么大变化的，改动的主要是启动、结束、另外还有一个新增的线程池，其他基本没什么，后续看看代码传到哪吧。
2021年12月9日23:53:06
序 大概是一些碎碎念
1.第一次见到网络轮子大概是在某个网课上，不过那个课里的代码，嗯，继承的地方有点多，搞得我有点乱，而且毕竟是别人代码，总是感觉迷迷糊糊的。
2.本文所述的轮子大概有2300行，文中大概会穿插部分代码，完整代码(毕竟不好放)可能之后会上传到某个地方吧。
3.之前看网课的时候以为自己什么都会了，真正自己写起来，一点一点定位问题，一点一点查资料的时候，才发现很多东西都不明白。
4.当然也有些思路和有些代码是从那个网课里拿过来的，不过大部分还是自己一点点搞出来的，保守点75%?.
5.主要内容就是写一些实现思路，和大概全部的实现细节，比如说遇到了哪些问题，如果解决，解决过程，又如某些地方为什么这么做，避免了什么，大概是这样的。
5.大概实现了日志、线程管理、内存池、内存池为基础的对象池、心跳、、服务端以及客户端统一的管理这些。
7.也有一些不是太好的地方，比如说一个成员函数，隔着两三层包含关系，调用了另一个类的同名成员函数，这可能会造成一些困惑？
8.另：毕业论文底稿，仅供学习参考，请勿进行任何形式转载。
一 类图及各类简述
首先，在整个类图中，绿色部分是分别在客户端和服务端可直接使用的三个类，蓝色部分的两个类则是两个全局单例对象。
1.CServiceNoBlock：是一个服务端的类，所有socket全都通过非阻塞模式进行accept、recv、send等行为，每次循环都会对所有socket进行操作，类似于轮询。
2.CServiceEpoll：继承自CServiceNoBlock，行为和CServiceNoBlock基本一样，主要区别在于：在对套接字进行轮询之前，使用epoll取得那些需要recv或send的套接字，只对这些epoll返回的套接字进行操作，这样避免了每次轮询所有套接字，效率上会有优势。
3.CClientLinkManage：CClientLink代表一个从客户端到服务端的连接，而CClientLinkManage则是一个管理这些连接的类，可在该类建立CClientLink并管理，同时可以在CClientLinkManage中对某个指定连接发送或处理数据。
4.CClientLink：代表一个从客户端主动向服务端发起连接的套接字，该对象被CClientLinkManage建立和管理。
5.CSocketObj：仅代表一个socket连接，不区分是从客户端主动连接还是服务端被动接收。
6.CSecondBuffer：如其名，作为一个第二缓冲区。
①位于send之前，要被发送的数据暂存于此，等待被发送线程统一发送。
②位于recv之后，recv的数据会被暂存于此，等待用户处理和取出。
7.CThreadManage：一个全局单例对象，存在目的在于为了在程序结束时，等待所有线程结束，所以如果需要，每次启动线程时，都应该调用该对象提供一个接口来启动线程。
8.CLog：一个全局单例对象，存在目的是为了提供日志接口，并在程序结束时处理好日志文件。
9.enum EMsgLevel：日志消息的等级。
10.struct CLogMsg：存储日志消息的结构
11.CDoubleBufLogQueue：一个双缓冲队列，各个线程提交日志时使用的结构，通过双缓冲的方式，将并发区减小到只交换一个指针。
12.CObjectPool：在内存池基础上，取得内存后，在该地址上使用 placement new主动调用构造，返回一个对象，归还时在内部调用其析构并归还内存。
13.CMemoryPool：一个内存池，在模板实例化后，构造中确定一个对象的大小，根据需要的数量申请足够的内存。需要时通过成员函数取得或者归还内存。
14.union CObj：程序中没有声明这个类型的对象，这个联合体存在的目的是为了在内存池初始化操作内存时的遍历。
15.CTimer：一个计时器。
16.Barrier：一个跨平台线程同步的屏障，用于同时等待多个线程。
17.struct CNetMsgHead：发送或接收消息时的自定义协议头部，存储包的长度和消息类型。
二 各个类详述及部分代码
2.1.main() & FairySunOfNetBaseStart() & FairySunOfNetBaseOver() & 两个辅助函数
在main()中，分别在开头调用FairySunOfNetBaseStart，在结束时调用FairySunOfNetBaseOver。
FairySunOfNetBaseStart()：实例化线程管理对象，实例化日志对象，并提交一条测试日志。接下来如果是windows平台，启动windowssocket环境，如果不是windows就是linux了，为SIGPIPE信号设置一个处理函数。为该信号设置处理函数的原因是，对端关闭的情况下，向其发送数据，偶尔会出现该信号，而该信号的默认行为是结束程序，但是这也不是错误，不需要结束程序，所以设置一个处理函数来避免程序结束。
FairySunOfNetBaseOver()：则是调用单例线程管理对象的结束函数，来正常等待结束所有线程。同时如果是windows，则关闭环境。
ErrorWaitExit()：这个函数是程序某些地方执行失败，无法继续执行时主动退出，调用该函数等待一段时间退出，以保证日志消息同步到磁盘，等待时间则是日志同步时间+1，单位秒。
int main()
{FairySunOfNetBaseStart();CTestSevice S1(6, 1000, 5);//S1.Mf_Epoll_Start(0, 4567);S1.Mf_NoBlock_Start(0, 4567);getchar();FairySunOfNetBaseOver();return 0;
}void FairySunOfNetBaseStart()
{CThreadManage::MfGetInstance(); // 这一行没有什么必要，毕竟每次调用该对象启动线程时，都会调用这个函数CLog::MfGetInstance(); // 实例化日志对象，并启动日志线程LogFormatMsgAndSubmit(std::this_thread::get_id() ,INFO_FairySun, "test"); // 提交一条日志保证启动#ifdef WIN32WORD sockVersion = MAKEWORD(2, 2);WSADATA wsdata;WSAStartup(sockVersion, &wsdata);
#else// 关于这里的修改见，第五条https://blog.csdn.net/qq_43082206/article/details/107374724sigset_t set;if (-1 == sigemptyset(&set)){LogFormatMsgAndSubmit(std::this_thread::get_id(), ERROR_FairySun, "sigemptyset error\n");ErrorWaitExit(0);}if (-1 == sigaddset(&set, SIGPIPE)){LogFormatMsgAndSubmit(std::this_thread::get_id(), ERROR_FairySun, "sigaddset error\n");ErrorWaitExit(0);}if (-1 == sigprocmask(SIG_BLOCK, &set, nullptr)){LogFormatMsgAndSubmit(std::this_thread::get_id(), ERROR_FairySun, "sigprocmask error\n");ErrorWaitExit(0);}//if (SIG_ERR == signal(SIGPIPE, SIGPIPE_Dispose))//{// printf("signal_fun set signal_dispose_fun fail\n");// LogFormatMsgAndSubmit(std::this_thread::get_id(), INFO_FairySun, "signal_fun set signal_dispose_fun fail\n");//}
#endif // !WIN32
}void FairySunOfNetBaseOver()
{CThreadManage::MfThreadManageOver(); // 最后清理线程管理类#ifdef WIN32WSACleanup();
#endif // !WIN32
}#ifndef WIN32
void SIGPIPE_Dispose(int no)
{printf("system produce SIGPIPE signal.\n");LogFormatMsgAndSubmit(std::this_thread::get_id(), WARN_FairySun, "system produce SIGPIPE signal.");
}
#endifextern const int LOG_SYN_TIME_SECOND;void ErrorWaitExit(int val)
{std::this_thread::sleep_for(std::chrono::seconds(LOG_SYN_TIME_SECOND + 1));exit(val);
}
2.2.CServiceNoBlock
本节有三个小部分。
第一部分(2.2.1)是类的定义。
第二部分(2.2.2)通过一张图，来描述各个线程以及其使用的结构。
第三部分(2.2.3)则是部分成员函数代码以及一些实现细节。
2.2.1类定义
class CServiceNoBlock
{protected:SOCKET MdListenSock; // 监听套接字int MdServiceMaxPeoples; // 该服务的上限人数，暂未被使用int MdDisposeThreadNums; // 消息处理线程数int MdHeartBeatTime; // 心跳时限，该时限内未收到消息默认断开，单位秒CTimer* MdHeartBeatTestInterval; // 心跳间隔检测用到的计时器CObjectPool<CSocketObj>* Md_CSocketObj_POOL; // 客户端对象的对象池std::map<SOCKET, CSocketObj*>* MdPClientJoinList; // 非正式客户端缓冲列表，等待加入正式列表，同正式客户列表一样，一个线程对应一个加入列表std::mutex* MdPClientJoinListMtx; // 非正式客户端列表的锁std::map<SOCKET, CSocketObj*>* MdPClientFormalList; // 正式的客户端列表，一个线程对应一个map[],存储当前线程的服务对象,map[threadid]找到对应map，map[threadid][socket]找出socket对应的数据std::shared_mutex* MdPClientFormalListMtx; // 为一MdPEachDisoposeThreadOfServiceObj[]添加sock时，应MdEachThreadOfMtx[].lockstd::map<SOCKET, CSocketObj*>* MdClientLeaveList; // 等待从正式列表中移除的缓冲列表std::mutex* MdClientLeaveListMtx; // 移除列表的锁
protected: // 用来在服务启动时，等待其他所有线程启动后，再启动Accept线程Barrier* MdBarrier1;
protected: // 这一组变量，用来在服务结束时，按accept recv dispose send的顺序来结束线程以保证不出错Barrier* MdBarrier2; // accept线程 和 recv线程 的同步变量 ！！！！！在epoll中未被使用！！！！！Barrier* MdBarrier3; // recv线程 和 所有dispos线程 的同步变量Barrier* MdBarrier4; // send线程 和 dispose线程 的同步和前面不同，用屏障的概念等待多个线程来继续执行
public:CServiceNoBlock(int HeartBeatTime = 300, int ServiceMaxPeoples = 1000, int DisposeThreadNums = 1);virtual ~CServiceNoBlock();void Mf_NoBlock_Start(const char* ip, unsigned short port); // 启动收发处理线程的非阻塞版本
protected:void Mf_Init_ListenSock(const char* ip, unsigned short port); // 初始化套接字
private:void Mf_NoBlock_AcceptThread(); // 等待客户端连接的线程void Mf_NoBlock_RecvThread(); // 收线程void Mf_NoBlock_DisposeThread(int SeqNumber); // 处理线程void Mf_NoBlock_SendThread(); // 发线程void Mf_NoBlock_ClientJoin(std::thread::id threadid, int SeqNumber); // 客户端加入正式列表void Mf_NoBlock_ClientLeave(std::thread::id threadid, int SeqNumber); // 客户端移除正式列表
protected:virtual void MfVNetMsgDisposeFun(SOCKET sock, CSocketObj* cli, CNetMsgHead* msg, std::thread::id & threadid) = 0; // 使用者复写该函数，来处理收到的消息
};
2.2.2各个线程及结构
这一节只是叙述线程和并发使用的结构，实现细节见2.3各个线程函数的具体描述。
在该类中，分别会启动AcceptThread RecvThread SendThread以及DisoposeThreadNums条DisposeThread(处理线程)。
每条线程会对应三个列表，分别是待加入列表、正式列表、待离开列表。(这里使用如此多的结构，则是为了尽可能降低锁的粒度，属于空间换时间)
AcceptThread中，则会每次找出当前负载人数最小的线程，把接收到的连接放入待加入列表。
RecvThread & SendThread中，则会把那些无效的、超时的、断开的描述符放入待离开列表。
各条DisposeThread中，则会每隔一段检查待加入和待离开列表，分别将他们加入正式列表，或从正式列表中移除。
对正式列表使用读锁的是RecvThread、DisposeThread、SendThread三条线程，使用写锁的则是MfClientJoin & MfClientLeave两个函数。
关于各个结构使用的结构，有些使用了独占锁，有些使用了读写锁，这些细节都在对应的中描述。
2.2.3部分成员函数及代码
void CServiceNoBlock::Mf_NoBlock_Start(const char * ip, unsigned short port)
①该函数首先调用另一个成员函数来初始化监听套接字。
②接下来初始化各个线程同步需要的线程屏障。
③最后按顺序启动各个线程。
关于线程启动的接口则在CThreadManage类中说明，主要是通过std::functional 加 std::bind，将线程函数统一包装成 void fun(void)的形式启动。
注：这里通过屏障操作，启动线程时，总是使accept线程最后启动，因为如果accept线程先启动了但是其他线程没有启动，可能会发生一些问题。结束线程时，总是按 AcceptThread RecvThread DisposeThread SendThread的顺序结束，行为是对所有的描述符分别进行 最后一次接收、处理、发送。
void CServiceNoBlock::Mf_NoBlock_Start(const char * ip, unsigned short port)
{std::thread::id threadid = std::this_thread::get_id();Mf_Init_ListenSock(ip, port);// 处理线程同步MdBarrier1->MfInit(MdDisposeThreadNums + 2 + 1); // 处理线程数量 + send和recv线程 + recv线程本身MdBarrier2->MfInit(2); // accept + recv两条线程MdBarrier3->MfInit(MdDisposeThreadNums + 1); // 处理线程数量 + recv线程MdBarrier4->MfInit(MdDisposeThreadNums + 1); // 处理线程数量 + send线程// 启动各个线程CThreadManage::MfCreateThreadAndDetach(std::bind(&CServiceNoBlock::Mf_NoBlock_SendThread, this)); // 启动发送线程for (int i = 0; i < MdDisposeThreadNums; ++i)CThreadManage::MfCreateThreadAndDetach(std::bind(&CServiceNoBlock::Mf_NoBlock_DisposeThread, this, i)); // 启动多条处理线程线程CThreadManage::MfCreateThreadAndDetach(std::bind(&CServiceNoBlock::Mf_NoBlock_RecvThread, this)); // 启动接收线程CThreadManage::MfCreateThreadAndDetach(std::bind(&CServiceNoBlock::Mf_NoBlock_AcceptThread, this)); // 启动等待连接线程
}
void CServiceNoBlock::Mf_Init_ListenSock(const char* ip, unsigned short port)
①首先把ip port信息填写到对应的结构里
②接着获取套接字，随后为其调用bind、listen，任何一个过程出错都会导致提交日志然后退出。
③最后没有出错则把监听套接字设置为非阻塞，至于为什么设置成阻塞模式，会在CServiceNoBlock::Mf_NoBlock_AcceptThread()中说明。
void CServiceNoBlock::Mf_Init_ListenSock(const char* ip, unsigned short port)
{std::thread::id threadid = std::this_thread::get_id();// 初始化监听套接字sockaddr_in addr{};addr.sin_family = AF_INET;addr.sin_port = htons(port);
#ifndef WIN32if (ip) addr.sin_addr.s_addr = inet_addr(ip);else addr.sin_addr.s_addr = INADDR_ANY;
#elseif (ip) addr.sin_addr.S_un.S_addr = inet_addr(ip);else addr.sin_addr.S_un.S_addr = INADDR_ANY;
#endifMdListenSock = socket(AF_INET, SOCK_STREAM, 0);if (SOCKET_ERROR == MdListenSock){LogFormatMsgAndSubmit(threadid, ERROR_FairySun, "server create listenSock failed!");ErrorWaitExit();}if (SOCKET_ERROR == bind(MdListenSock, (sockaddr*)&addr, sizeof(addr))){LogFormatMsgAndSubmit(threadid, ERROR_FairySun, "listenSock bind failed!");ErrorWaitExit();}if (SOCKET_ERROR == listen(MdListenSock, 100)){LogFormatMsgAndSubmit(threadid, ERROR_FairySun, "listen socket failed!");ErrorWaitExit();}// 将接收套接字设为非阻塞，这里为了解决两个问题// 1是阻塞套接字，在整个服务程序退出时，如果没有客户端连接到来，会导致accept线程阻塞迟迟无法退出// 2是在unp第十六章非阻塞accept里，不过这个问题只在单线程情况下出现，就不写了
#ifndef WIN32 // 对套接字设置非阻塞int flags = fcntl(MdListenSock, F_GETFL, 0);fcntl(MdListenSock, F_SETFL, flags | O_NONBLOCK);
#elseunsigned long ul = 1;ioctlsocket(MdListenSock, FIONBIO, &ul);
#endif
}
void CServiceNoBlock::Mf_NoBlock_AcceptThread()
①该线程函数中，首先声明用于接收返回的sock以及地址变量。
②随后使用MdBarrier1变量等待RecvThread DisposeThread SendThread等线程启动。
③接着就是主循环，主循环的执行条件则是在全局的线程管理对象中的一个flag，通过线程ID获取。
④说明为什么使用非阻塞的监听套接字，在这里如果使用非阻塞，则在程序结束(主循环flag标志被设为false)时，陷入accept调用后，如果迟迟没有新的连接到来，accept也就无法返回，那么AcceptThread也就不会退出，导致整个程序卡住。
⑤主循环中，accept成功后，先对返回的套接字设置非阻塞，并为其从对象池取得一个连接对象(该对象在MfClientLeave | SendThread结束时被归还给对象池)，然后寻找当前服务人数最少的那个线程，找到后将该套接字及对应的连接对象加入对应的待加入列表，在Dispose线程中被正式加入，关于各个线程间并发使用的结构，则在2.2.2中描述。
⑥当主循环结束后，通知recv线程可以继续执行，进行最后一次接收。
void CServiceNoBlock::Mf_NoBlock_AcceptThread()
{std::thread::id threadid = std::this_thread::get_id();// 接受连接时用到的数据SOCKET sock ; sockaddr_in addr; // 地址int sizeofsockaddr = sizeof(sockaddr);
#ifndef WIN32socklen_t len; // 地址长度
#elseint len; // 地址长度
#endifint minPeoples = INTMAX_MAX; // 存储一个最小人数int minId = 0; // 最小人数线程的Idint currPeoples = 0; // 当前已连接人数// 等待其他所有线程MdBarrier1->MfWait();printf("NoBlock Accept Thread already Start\n");LogFormatMsgAndSubmit(threadid, INFO_FairySun, "NoBlock AcceptThread already Start");// 主循环，每次循环接收一个连接while (CThreadManage::MfGetThreadFlag(threadid)) {currPeoples = 0;for (int i = 0; i < MdDisposeThreadNums; ++i) // 计算当前已连接人数currPeoples += MdPClientFormalList[i].size();if (currPeoples > MdServiceMaxPeoples) // 大于上限服务人数，就不accpet，等待然后开始下一轮循环{std::this_thread::sleep_for(std::chrono::seconds(1));continue;}sock = INVALID_SOCKET;addr = {};len = sizeofsockaddr;sock = accept(MdListenSock, (sockaddr*)&addr, &len);if (SOCKET_ERROR == sock){if (errno == 0 || errno == EWOULDBLOCK) // windows非阻塞sock没有到达时errno是0，linux则是EWOULDBLOCKstd::this_thread::sleep_for(std::chrono::milliseconds(1));else if (INVALID_SOCKET == sock)LogFormatMsgAndSubmit(threadid, WARN_FairySun, "accept return INVALID_SOCKET!");elseLogFormatMsgAndSubmit(threadid, WARN_FairySun, "accept return ERROR!");continue;}else{#ifndef WIN32 // 对收到套接字设置非阻塞int flags = fcntl(sock, F_GETFL, 0);fcntl(sock, F_SETFL, flags | O_NONBLOCK);
#elseunsigned long ul = 1;ioctlsocket(sock, FIONBIO, &ul);
#endifminPeoples = INTMAX_MAX;for (int i = 0; i < MdDisposeThreadNums; ++i) // 找出人数最少的{if (minPeoples > MdPClientFormalList[i].size()){minPeoples = MdPClientFormalList[i].size();minId = i;}}CSocketObj* TempSocketObj = Md_CSocketObj_POOL->MfApplyObject(sock);TempSocketObj->MfSetPeerAddr(&addr);TempSocketObj->MfSetThreadIndex(minId);std::lock_guard<std::mutex> lk(MdPClientJoinListMtx[minId]); // 对应的线程map上锁MdPClientJoinList[minId].insert(std::pair<SOCKET, CSocketObj*>(sock, TempSocketObj)); // 添加该连接到加入缓冲区LogFormatMsgAndSubmit(threadid, INFO_FairySun, "SOCKET <%5d> <%s:%5d>\tTo ClientJoinList[%d]", sock, TempSocketObj->MfGetPeerIP(), TempSocketObj->MfGetPeerPort(), minId);}}// 主循环结束后清理
#ifdef WIN32closesocket(MdListenSock);
#elseclose(MdListenSock);
#endif // WIN32printf("NoBlock Accept Thread already over\n");LogFormatMsgAndSubmit(threadid, INFO_FairySun, "NoBlock Accept Thread already over");MdBarrier2->MfWait();
}
void CServiceNoBlock::Mf_NoBlock_RecvThread()
主循环之前则是和AcceptThread线程同步的操作。
①主循环中，第一层循环是各条DisposeThread线程对应正式列表，每次循环会私用一个读锁来锁定对应的正式列表不被修改。
②第二层循环则是对其中一个正式列表中的所有客户端对象调用接收数据的函数。
③如果返回值≤0，则出错，应该把该列表放入待离开列表。注意这里待离开列表是尝试锁定，使用尝试锁定的原因是：待离开列表会被RecvThrea和SendThread并发使用，这里为了不被卡住，使用尝试锁定，即便尝试锁定失败，也不会有什么影响，因为下一次循环时，其仍然会被发现错误，一次次尝试锁定，总有一次能成功。
④如果 返回值≥0 && 返回值!=INT32_MAX，则说明正确接收了数据，接收数据量为返回值，成功接收到了值就更新该对象的心跳计时(心跳计时则会在DisposeThread中每隔一段时间检测)。判读返回值小于0的是为了判断是否出错。
⑤如果返回值是 INT32_MAX，该值代表recv调用没有出错，但是也没有成功那个接收到数据。返回INT32_MAX的目的则是为了在没有出错的情况下，区别是否成功接收到了数据，没有成功接收数据就不应该更新计时器。
⑥主循环最后，则是一个1毫秒的暂停，其存在是为了解决服务列表中没有连接对象或者连接对象很少时，过快的空循环导致CPU被大量无效占用。
⑦主循环之后，则是整个service对象结束时，按顺序结束线程的屏障，等待AccpetThread线程发来通知，对所有客户端进行最后一次接收数据。不需要判定错误，因为马上就要结束了。
void CServiceNoBlock::Mf_NoBlock_RecvThread()
{std::thread::id threadid = std::this_thread::get_id();printf("Recv Thread already Start\n");LogFormatMsgAndSubmit(threadid, INFO_FairySun, "Recv Thread already Start");MdBarrier1->MfWait(); int ret = 0;// 主循环while (CThreadManage::MfGetThreadFlag(threadid)){for (int i = 0; i < MdDisposeThreadNums; ++i){ std::shared_lock<std::shared_mutex> ReadLock(MdPClientFormalListMtx[i]); // 对应的线程map上锁，放在这里上锁的原因是为了防止插入行为导致迭代器失效for(auto it = MdPClientFormalList[i].begin(); it != MdPClientFormalList[i].end(); ++it){ret = it->second->MfRecv();if (0 >= ret) // 返回值小于等于0时表示socket出错或者对端关闭{std::unique_lock<std::mutex> lk(MdClientLeaveListMtx[i], std::defer_lock); // 这里尝试锁定，而不是阻塞锁定，因为这里允许被跳过if (lk.try_lock()) // 每一轮recv或者send发现返回值<=0,都会尝试锁定，该连接加入待移除缓冲区MdClientLeaveList[i].insert(std::pair<SOCKET, CSocketObj*>(it->first, it->second)); }else if(INT32_MAX != ret) // 没有出错时返回的值都是大于等于0的，但是返回值是INT32_MAX时，没有出错，但是也没有成功压入数据，只有成功压入数据时才更新计时器it->second->MfHeartBeatUpDate();}}std::this_thread::sleep_for(std::chrono::milliseconds(1));}// 主循环结束后，首先等待accept线程的结束以及发来的通知MdBarrier2->MfWait();// 然后对所有套接字最后一次接收for (int i = 0; i < MdDisposeThreadNums; ++i){std::shared_lock<std::shared_mutex> ReadLock(MdPClientFormalListMtx[i]); for (auto it = MdPClientFormalList[i].begin(); it != MdPClientFormalList[i].end(); ++it)it->second->MfRecv();}printf("Recv Thread already over\n");LogFormatMsgAndSubmit(threadid, INFO_FairySun, "Recv Thread already over");// 接下来通知所有dispose线程MdBarrier3->MfWait();
}
void CServiceNoBlock::Mf_NoBlock_DisposeThread(int SeqNumber)
同样，主循环之前是和AcceptThread线程同步的操作。
①主循环中，每隔一秒检查其对应的待加入和待离开队列，将他们移入正式列表，或者从正式列表中移出。
②接下来使用读锁锁定，检查 正式列表中 所有描述符 的 第二缓冲区，如果有消息，先检查是不是默认的心跳包，是就更新心跳计时，否则就调用用户覆写的虚函数MfVNetMsgDisposeFun来处理。
③主循环最后依旧是1毫秒的暂停，防止空循环占用CPU。
④主循环之后等待RecvThread的同步，RecvThread完成最后一次接收后，DisposeThread则开始最后一次处理。
⑤最后则是DisposeThread和SendThread的同步，通知其完成最后一次发送。
void CServiceNoBlock::Mf_NoBlock_DisposeThread(int SeqNumber)
{std::thread::id threadid = std::this_thread::get_id();CTimer time;printf("Dispose Thread <%d> already Start\n", SeqNumber);LogFormatMsgAndSubmit(threadid, INFO_FairySun, "Dispose Thread <%d> already Start", SeqNumber);MdBarrier1->MfWait();// 主循环while (CThreadManage::MfGetThreadFlag(threadid)){if (time.getElapsedSecond() > 1){Mf_NoBlock_ClientJoin(threadid, SeqNumber); // 将待加入队列的客户放入正式Mf_NoBlock_ClientLeave(threadid, SeqNumber); // 清理离开队列的客户端time.update();}// 遍历正式客户端列表处理所有消息{std::shared_lock<std::shared_mutex> ReadLock(MdPClientFormalListMtx[SeqNumber]);for (auto it : MdPClientFormalList[SeqNumber]){while (it.second->MfHasMsg()){if (-1 == ((CNetMsgHead*)it.second->MfGetRecvBufP())->MdCmd) // 当该包的该字段为-1时，代表心跳it.second->MfHeartBeatUpDate(); // 更新心跳计时elseMfVNetMsgDisposeFun(it.first, it.second, (CNetMsgHead*)it.second->MfGetRecvBufP(), threadid);it.second->MfPopFrontMsg();}}}std::this_thread::sleep_for(std::chrono::milliseconds(1));}// 主循环结束后，所有处理线程都等待recv线程发来的通知MdBarrier3->MfWait();// 对所有套接字进行最后一次处理for (auto it : MdPClientFormalList[SeqNumber]){while (it.second->MfHasMsg()){MfVNetMsgDisposeFun(it.first, it.second, (CNetMsgHead*)it.second->MfGetRecvBufP(), threadid);it.second->MfPopFrontMsg();}}printf("Dispose Thread <%d> already over\n", SeqNumber);LogFormatMsgAndSubmit(threadid, INFO_FairySun, "Dispose Thread <%d> already over", SeqNumber);MdBarrier4->MfWait();
}
void CServiceNoBlock::Mf_NoBlock_SendThread()
主循环之前也是和AcceptThread同步的操作。
①和RecvThread类似的形式，一个两层循环。
②第一层循环遍历各个DisposeThread的正式列表。并使用读锁锁定
③第二层循环遍历相应正式列表中的每一个连接对象，为其调用send，将第二缓冲区中存储的数据发送出去。
④这里的返回值只需要判断socket是否出错、无效、断开，send则不需要更新心跳。因为只有收到对方的数据才能代表对方存活。
⑤接下来就是等待所有Dispose线程结束后，对所有连接进行最后一次发送。
⑥最后则是关闭所有描述符，并将其归还到对象池。
void CServiceNoBlock::Mf_NoBlock_SendThread()
{std::thread::id threadid = std::this_thread::get_id();printf("Send Thread already Start\n");LogFormatMsgAndSubmit(threadid, INFO_FairySun, "Send Thread already Start");MdBarrier1->MfWait();int ret = 0;// 主循环while (CThreadManage::MfGetThreadFlag(threadid)){for (int i = 0; i < MdDisposeThreadNums; ++i){std::shared_lock<std::shared_mutex> ReadLock(MdPClientFormalListMtx[i]); // 对应的线程map上锁，放在这里上锁的原因是为了防止插入行为导致迭代器失效for (auto it = MdPClientFormalList[i].begin(); it != MdPClientFormalList[i].end(); ++it){ret = it->second->MfSend();if (0 >= ret){std::unique_lock<std::mutex> lk(MdClientLeaveListMtx[i], std::defer_lock); // 这里尝试锁定，而不是阻塞锁定，因为这里允许被跳过if (lk.try_lock()) // 每一轮recv或者send发现返回值<=0,都会尝试锁定,总有一次能锁定他，将该连接加入待移除缓冲区MdClientLeaveList[i].insert(std::pair<SOCKET, CSocketObj*>(it->first, it->second)); }//else if (INT32_MAX != ret) // 没有出错时返回的值都是大于等于0的，但是返回值是INT32_MAX时，没有出错// ;}}std::this_thread::sleep_for(std::chrono::milliseconds(1));}// 主循环结束后，等待所有处理线程到达屏障,发送完所有数据，然后关闭所有套接字MdBarrier4->MfWait();// 发送所有数据for (int i = 0; i < MdDisposeThreadNums; ++i){std::shared_lock<std::shared_mutex> ReadLock(MdPClientFormalListMtx[i]); for (auto it = MdPClientFormalList[i].begin(); it != MdPClientFormalList[i].end(); ++it)it->second->MfSend();}// 关闭所有套接字for (int i = 0; i < MdDisposeThreadNums; ++i){for (auto it = MdPClientFormalList[i].begin(); it != MdPClientFormalList[i].end(); ++it){it->second->MfClose();Md_CSocketObj_POOL->MfReturnObject(it->second);}}printf("Send Thread already over\n");LogFormatMsgAndSubmit(threadid, INFO_FairySun, "Send Thread already over");
}
void CServiceNoBlock::Mf_NoBlock_ClientJoin
该函数将待加入表中的对象加入到正式列表，因为需要修改正式列表，所以这里使用写锁独占，全部加入正式列表后后清空待加入列表。
void CServiceNoBlock::Mf_NoBlock_ClientJoin(std::thread::id threadid, int SeqNumber)
{// 这个函数是客户端加入的，需要操作的，阻塞也是必然的std::lock_guard<std::mutex> lk(MdPClientJoinListMtx[SeqNumber]);if (!MdPClientJoinList[SeqNumber].empty()){std::lock_guard<std::shared_mutex> WriteLock(MdPClientFormalListMtx[SeqNumber]);for (auto it = MdPClientJoinList[SeqNumber].begin(); it != MdPClientJoinList[SeqNumber].end(); ++it){MdPClientFormalList[SeqNumber].insert(std::pair<SOCKET, CSocketObj*>(it->first, it->second));LogFormatMsgAndSubmit(threadid, INFO_FairySun, "SOCKET <%5d> <%s:%5d>\tTo ClientFormalList[%d]", it->first, it->second->MfGetPeerIP(), it->second->MfGetPeerPort(), SeqNumber);}MdPClientJoinList[SeqNumber].clear();}
}
void CServiceNoBlock::Mf_NoBlock_ClientLeave
①该函数是检查定时器，这个定时器是每隔一段时间检查所有连接的心跳计时，如果超时就将其放入待离开列表。
②接下来遍历待离开列表，将其依次从正式列表中找到并移除，最后清空待离开列表。
void CServiceNoBlock::Mf_NoBlock_ClientLeave(std::thread::id threadid, int SeqNumber)
{static const int HeartIntervalTime = MdHeartBeatTime / 3; // 心跳检测间隔时限// 本函数中的两个也都使用尝试锁，因为这里不是重要的地方，也不是需要高效执行的地方std::unique_lock<std::mutex> lk(MdClientLeaveListMtx[SeqNumber], std::defer_lock);std::unique_lock<std::shared_mutex> WriteLock(MdPClientFormalListMtx[SeqNumber], std::defer_lock);if (lk.try_lock() && WriteLock.try_lock()){ // 第一个循环每隔 心跳超时时间的三分之一 检测一次心跳是否超时if (HeartIntervalTime < MdHeartBeatTestInterval->getElapsedSecond()){// 遍历当前列表的客户端心跳计时器，超时就放入待离开列表for (auto it : MdPClientFormalList[SeqNumber]){if (it.second->MfHeartIsTimeOut(MdHeartBeatTime)){MdClientLeaveList[SeqNumber].insert(std::pair<SOCKET, CSocketObj*>(it.first, it.second));LogFormatMsgAndSubmit(threadid, INFO_FairySun, "SOCKET <%5d> <%s:%5d> heart time out", it.first, it.second->MfGetPeerIP(), it.second->MfGetPeerPort());}}}// 第二个循环清除客户端if (!MdClientLeaveList[SeqNumber].empty()){for (auto it = MdClientLeaveList[SeqNumber].begin(); it != MdClientLeaveList[SeqNumber].end(); ++it){MdPClientFormalList[SeqNumber].erase(it->first);it->second->MfClose();LogFormatMsgAndSubmit(threadid, INFO_FairySun, "SOCKET <%5d> <%s:%5d>\tClose And Leave ClientFormalList[%d]", it->first, it->second->MfGetPeerIP(), it->second->MfGetPeerPort(), it->second->MfGetThreadIndex());Md_CSocketObj_POOL->MfReturnObject(it->second);}MdClientLeaveList[SeqNumber].clear();}}
}
2.3.CServiceEpoll
本节有三个小部分。
第一部分(2.3.1)是类的定义。
第二部分(2.3.2)简单介绍和思路描述。
第三部分(2.3.3)则是部分成员函数代码以及一些实现细节。
2.3.1类的定义
#ifndef WIN32class CServiceEpoll :public CServiceNoBlock
{private:std::vector<SOCKET> MdEpoll_In_Fd; // epoll句柄std::vector<epoll_event*> MdEpoll_In_Event; // 接收线程用到的结构集合int MdThreadAvgPeoples; // 每个线程的平均人数
public:CServiceEpoll(int HeartBeatTime = 300, int ServiceMaxPeoples = 1000, int DisposeThreadNums = 1);virtual ~CServiceEpoll();void Mf_Epoll_Start(const char* ip, unsigned short port);
private:void Mf_Epoll_AcceptThread(); // 等待客户端连接的线程void Mf_Epoll_RecvAndDisposeThread(int SeqNumber); // 处理线程void Mf_Epoll_SendThread(); // 发线程void Mf_Epoll_ClientJoin(std::thread::id threadid, int SeqNumber); // 客户端加入正式列表void Mf_Epoll_ClientLeave(std::thread::id threadid, int SeqNumber); // 客户端移除正式列表
};#endif // !WIN32
2.3.2思路描述
首先使用宏定义，因为epoll只存在于linux，所以只定义在linux下。
主要思路是在Recv和send时，不是再轮询所有的描述符了（直接recv或send），而是在为其调用recv或send前，先使用epoll监听，是否需要recv或send，然后只对其返回的列表进行操作，这样就避免了无用的recv或send轮询。
这里直接继承自非阻塞的版本，区别是重写了Start、AcceptThread、RecvThread、SendThread、MfClientJoin、MfClientLeave等成员函数。因为这些函数中涉及到了epool的操作。
另外，合并了RecvThread和DisposeThread两个函数，因为epoll在接收之后随后就要进行处理，如果分开到不同的线程，操作上会有一些不太方便。
这里的epoll描述符和待加入、正式、待离开列表一样，每个DisposeThread对应一个。
最后，这里只在recv时使用了epoll，在recv中只监听可读事件。send时则不需要，因为send只要缓冲区还有空间就返回可写事件，而在调用send发送前，总是会检查其第二缓冲区是否有数据，之后才决定是否调用send，而当send缓冲区数据满导致调用失败时，第二缓冲区的数据也不会被写出去。
2.3.3部分成员函数及代码
void CServiceEpoll::Mf_Epoll_Start(const char* ip, unsigned short port)
epoll版本的start函数和NoBlock版本的基本相同，区别一是为epoll建立描述符，区别二是启动线程时，把recv的操作合并到了DisposeThread中。
void CServiceEpoll::Mf_Epoll_Start(const char* ip, unsigned short port)
{for (int i = 0; i < MdDisposeThreadNums; ++i) // 为各个处理线程建立epoll的描述符{MdEpoll_In_Fd.push_back(epoll_create(MdThreadAvgPeoples));}for (int i = 0; i < MdDisposeThreadNums; ++i) // 检查各个描述符是否出错{if (-1 == MdEpoll_In_Fd[i]){printf("create epoll fd error!\n");LogFormatMsgAndSubmit(std::this_thread::get_id(), ERROR_FairySun, "create epoll fd error!");ErrorWaitExit();}}Mf_Init_ListenSock(ip, port);// 处理线程同步MdBarrier1->MfInit(MdDisposeThreadNums + 1 + 1); // 处理线程数量 + send和recv线程 + recv线程本身MdBarrier2->MfInit(2); // 在epoll中未被使用MdBarrier3->MfInit(MdDisposeThreadNums + 1); // accept + 处理线程数量MdBarrier4->MfInit(MdDisposeThreadNums + 1); // 处理线程数量 + send线程CThreadManage::MfCreateThreadAndDetach(std::bind(&CServiceEpoll::Mf_Epoll_SendThread, this)); // 启动发送线程for (int i = 0; i < MdDisposeThreadNums; ++i)CThreadManage::MfCreateThreadAndDetach(std::bind(&CServiceEpoll::Mf_Epoll_RecvAndDisposeThread, this, i)); // 启动多条处理线程,！！！！收线程和处理线程合并了！！！！CThreadManage::MfCreateThreadAndDetach(std::bind(&CServiceEpoll::Mf_Epoll_AcceptThread, this)); // 启动等待连接线程
}
void CServiceEpoll::Mf_Epoll_AcceptThread()
这一块代码和NoBlock版本只有一处不同，为了减少篇幅就不贴了
不同的地方是AcceptThread和DisposeThread中间的RecvThread线程被合并了，线程间的同步变量发生了调整，仅此而已。
void CServiceEpoll::Mf_Epoll_RecvAndDisposeThread(int SeqNumber)
主循环前是和AcceptThread的同步操作。
①主循环中，首先每隔1秒来检查待加入队列和待离开队列。
②接下来就是上读锁，立即返回的epoll_wait()调用，返回后对可读的描述符集合遍历，依次调用recv，接着就是处理消息。
③出错条件和更新计时器条件和NoBlock版本的RecvThread一样。
④主循环最后依旧有1毫秒的等待。
⑤主循环后面是等待AcceptThread发来通知，对所有描述符进行最后依次接收和处理。处理完毕之后通知SendThread进行最后一次的发送。
void CServiceEpoll::Mf_Epoll_RecvAndDisposeThread(int SeqNumber)
{std::thread::id threadid = std::this_thread::get_id();CTimer time;printf("Recv and Dispose Thread <%d> already Start\n", SeqNumber);LogFormatMsgAndSubmit(threadid, INFO_FairySun, "Recv and Dispose Thread <%d> already Start", SeqNumber);MdBarrier1->MfWait();int Epoll_N_Fds;int ret = 0;SOCKET tmp_fd;CSocketObj* tmp_obj;// 主循环while (CThreadManage::MfGetThreadFlag(threadid)){if (time.getElapsedSecond() > 1){Mf_Epoll_ClientJoin(threadid, SeqNumber); // 将待加入队列的客户放入正式Mf_Epoll_ClientLeave(threadid, SeqNumber); // 清理离开队列的客户端time.update();}// 遍历正式客户端列表处理所有消息{std::shared_lock<std::shared_mutex> ReadLock(MdPClientFormalListMtx[SeqNumber]);Epoll_N_Fds = epoll_wait(MdEpoll_In_Fd[SeqNumber], MdEpoll_In_Event[SeqNumber], MdThreadAvgPeoples, 0);// 第一个循环对epoll返回的集合接收数据for (int j = 0; j < Epoll_N_Fds; ++j){tmp_fd = MdEpoll_In_Event[SeqNumber][j].data.fd;tmp_obj = MdPClientFormalList[SeqNumber][tmp_fd];ret = tmp_obj->MfRecv();if (0 >= ret) // 返回值小于等于0时表示socket出错或者对端关闭{std::unique_lock<std::mutex> lk(MdClientLeaveListMtx[SeqNumber], std::defer_lock); // 这里尝试锁定，而不是阻塞锁定，因为这里允许被跳过if (lk.try_lock()) // 每一轮recv或者send发现返回值<=0,都会尝试锁定MdClientLeaveList[SeqNumber].insert(std::pair<SOCKET, CSocketObj*>(tmp_fd, tmp_obj)); }else if (INT32_MAX != ret) // 没有出错时返回的值都是大于等于0的，但是返回值是INT32_MAX时，没有出错，但是也没有成功压入数据，只有成功压入数据时才更新计时器MdPClientFormalList[SeqNumber][tmp_fd]->MfHeartBeatUpDate();}// 第二个循环处理那些收到的数据for (int j = 0; j < Epoll_N_Fds; ++j){tmp_fd = MdEpoll_In_Event[SeqNumber][j].data.fd;tmp_obj = MdPClientFormalList[SeqNumber][tmp_fd];while (tmp_obj->MfHasMsg()){if (-1 == ((CNetMsgHead*)tmp_obj->MfGetRecvBufP())->MdCmd) // 当该包的该字段为-1时，代表心跳包tmp_obj->MfHeartBeatUpDate(); // 更新心跳计时elseMfVNetMsgDisposeFun(tmp_fd, tmp_obj, (CNetMsgHead*)tmp_obj->MfGetRecvBufP(), threadid);tmp_obj->MfPopFrontMsg();}}}std::this_thread::sleep_for(std::chrono::milliseconds(1));}// 主循环结束后，所有处理线程都等待recv线程发来的通知MdBarrier3->MfWait();// 对所有套接字进行最后一次处理std::shared_lock<std::shared_mutex> ReadLock(MdPClientFormalListMtx[SeqNumber]);Epoll_N_Fds = epoll_wait(MdEpoll_In_Fd[SeqNumber], MdEpoll_In_Event[SeqNumber], MdThreadAvgPeoples, 0);for (int j = 0; j < Epoll_N_Fds; ++j){tmp_fd = MdEpoll_In_Event[SeqNumber][j].data.fd;tmp_obj = MdPClientFormalList[SeqNumber][tmp_fd];ret = tmp_obj->MfRecv();while (tmp_obj->MfHasMsg()){MfVNetMsgDisposeFun(tmp_fd, tmp_obj, (CNetMsgHead*)tmp_obj->MfGetRecvBufP(), threadid);tmp_obj->MfPopFrontMsg();}}printf("Recv and Dispose Thread <%d> already over\n", SeqNumber);LogFormatMsgAndSubmit(threadid, INFO_FairySun, "Recv and Dispose Thread <%d> already over", SeqNumber);MdBarrier4->MfWait();
}
void CServiceEpoll::Mf_Epoll_SendThread()
在一开始，调用send之前，也会通过epoll去监听可写事件，但是一般情况下，只要缓冲区中还有空间，就会有可写事件，这里被我觉得是不必要的，所以去掉了，改成了现在的样子。在it.second->MfSend();这个调用内部会检查第二缓冲区是否有数据，有数据就调用send，即便send调用失败，也会有安全保证。
所以这个函数看起来整个和NoBlock的版本基本没有区别，不过我不想回去在删掉这个函数了。
void CServiceEpoll::Mf_Epoll_SendThread()
{std::thread::id threadid = std::this_thread::get_id();printf("Send Thread already Start\n");LogFormatMsgAndSubmit(threadid, INFO_FairySun, "Send Thread already Start");MdBarrier1->MfWait();int ret = 0;// 主循环while (CThreadManage::MfGetThreadFlag(threadid)){for (int i = 0; i < MdDisposeThreadNums; ++i){std::shared_lock<std::shared_mutex> ReadLock(MdPClientFormalListMtx[i]); // 对应的线程map上锁，放在这里上锁的原因是为了防止插入行为导致迭代器失效for (auto it: MdPClientFormalList[i]){ret = it.second->MfSend();if (0 >= ret) // 返回值小于等于0时表示socket出错或者对端关闭{std::unique_lock<std::mutex> lk(MdClientLeaveListMtx[i], std::defer_lock); // 这里尝试锁定，而不是阻塞锁定，因为这里允许被跳过if (lk.try_lock()) // 每一轮recv或者send发现返回值<=0,都会尝试锁定MdClientLeaveList[i].insert(std::pair<SOCKET, CSocketObj*>(it.first, it.second));}//else if (INT32_MAX != ret) // 没有出错时返回的值都是大于等于0的，但是返回值是INT32_MAX时，没有出错，但是也没有成功压入数据，只有成功压入数据时才更新计时器// ;}}std::this_thread::sleep_for(std::chrono::milliseconds(1));}// 主循环结束后，等待所有处理线程到达屏障,发送完所有数据，然后关闭所有套接字MdBarrier4->MfWait();// 发送所有数据for (int i = 0; i < MdDisposeThreadNums; ++i){std::shared_lock<std::shared_mutex> ReadLock(MdPClientFormalListMtx[i]); for (auto it: MdPClientFormalList[i]){it.second->MfSend();}}for (int i = 0; i < MdDisposeThreadNums; ++i){for (auto it = MdPClientFormalList[i].begin(); it != MdPClientFormalList[i].end(); ++it){it->second->MfClose();Md_CSocketObj_POOL->MfReturnObject(it->second);}}printf("Send Thread already over\n");LogFormatMsgAndSubmit(threadid, INFO_FairySun, "Send Thread already over");
}
void CServiceEpoll::Mf_Epoll_ClientJoin(std::thread::id threadid, int SeqNumber)
这个函数同非阻塞版本的区别把socket描述符的读事件加入epoll的描述符集合。
void CServiceEpoll::Mf_Epoll_ClientJoin(std::thread::id threadid, int SeqNumber)
{epoll_event EvIn{};EvIn.events = EPOLLIN; std::lock_guard<std::mutex> lk(MdPClientJoinListMtx[SeqNumber]);if (!MdPClientJoinList[SeqNumber].empty()){std::lock_guard<std::shared_mutex> WriteLock(MdPClientFormalListMtx[SeqNumber]);for (auto it = MdPClientJoinList[SeqNumber].begin(); it != MdPClientJoinList[SeqNumber].end(); ++it){MdPClientFormalList[SeqNumber].insert(std::pair<SOCKET, CSocketObj*>(it->first, it->second));EvIn.data.fd = it->first;epoll_ctl(MdEpoll_In_Fd[SeqNumber], EPOLL_CTL_ADD, it->first, &EvIn); // 设置套接字到收线程的EpollLogFormatMsgAndSubmit(threadid, INFO_FairySun, "SOCKET <%5d> <%s:%5d>\tTo ClientFormalList[%d]", it->first, it->second->MfGetPeerIP(), it->second->MfGetPeerPort(), SeqNumber);}MdPClientJoinList[SeqNumber].clear();}
}
void CServiceEpoll::Mf_Epoll_ClientLeave(std::thread::id threadid, int SeqNumber)
这里的逻辑依旧是每隔一段时间检查心跳，然后移除客户端，多出来的操作是将描述符从epoll集合从移除。
void CServiceEpoll::Mf_Epoll_ClientLeave(std::thread::id threadid, int SeqNumber)
{static const int HeartIntervalTime = MdHeartBeatTime / 3; // 心跳检测间隔时限// 本函数中的两个也都使用尝试锁，因为这里不是重要的地方，也不是需要高效执行的地方std::unique_lock<std::mutex> lk(MdClientLeaveListMtx[SeqNumber], std::defer_lock);std::unique_lock<std::shared_mutex> WriteLock(MdPClientFormalListMtx[SeqNumber], std::defer_lock);if (lk.try_lock() && WriteLock.try_lock()){// 第一个循环每隔 心跳超时时间的三分之一 检测一次心跳是否超时if (HeartIntervalTime < MdHeartBeatTestInterval->getElapsedSecond()){// 遍历当前列表的客户端心跳计时器，超时就放入待离开列表for (auto it : MdPClientFormalList[SeqNumber]){if (it.second->MfHeartIsTimeOut(MdHeartBeatTime)){MdClientLeaveList[SeqNumber].insert(std::pair<SOCKET, CSocketObj*>(it.first, it.second));LogFormatMsgAndSubmit(threadid, INFO_FairySun, "SOCKET <%5d> <%s:%5d> heart time out", it.first, it.second->MfGetPeerIP(), it.second->MfGetPeerPort());}}}// 第二个循环清除客户端if (!MdClientLeaveList[SeqNumber].empty()){for (auto it = MdClientLeaveList[SeqNumber].begin(); it != MdClientLeaveList[SeqNumber].end(); ++it){it->second->MfClose();epoll_ctl(MdEpoll_In_Fd[SeqNumber], EPOLL_CTL_DEL, it->first, nullptr);MdPClientFormalList[SeqNumber].erase(it->first);LogFormatMsgAndSubmit(threadid, INFO_FairySun, "SOCKET <%5d> <%s:%5d>\tClose And Leave ClientFormalList[%d]", it->first, it->second->MfGetPeerIP(), it->second->MfGetPeerPort(), it->second->MfGetThreadIndex());Md_CSocketObj_POOL->MfReturnObject(it->second);}MdClientLeaveList[SeqNumber].clear();}}
}
2.4.CClientLinkManage
2.4.1类定义
class CClientLinkManage
{private:std::map<std::string, CClientLink*> MdClientLinkList;std::shared_mutex MdClientLinkListMtx;std::atomic<int> MdIsStart = 0; // 收发线程是否启动，不启动时，不能添加连接，因为如果放在构造中启动线程是危险的Barrier* MdBarrier; // 用于创建连接前的收发线程启动int MdHeartSendInterval; // 心跳发送时间间隔，单位秒CNetMsgHead* MdDefautHeartPacket; // 默认的心跳包对象CTimer* MdHeartTime; // 心跳计时器对象
public:CClientLinkManage(int HeartSendInterval = 3);~CClientLinkManage();void MfStart(); // 启动收发线程int MfCreateAddLink(std::string Linkname, const char* ip, unsigned short port); // 如果需要建立新的连接，就new一个ClientLink，同时记录连接的目标，然后加入列表是，设置client可用void MfCloseLink(std::string Linkname); // 关闭某个连接bool MfSendData(std::string str, const char* data, int len); // 发送数据，插入缓冲区const char* MfGetRecvBufferP(std::string name); // 返回接收缓冲区的指针，可以直接读这里的数据bool MfHasMsg(std::string name); // 判断缓冲区是否有数据void MfPopFrontMsg(std::string name); // 缓冲区数据按包弹出bool MfLinkIsSurvive(std::string name); // 某个服务是否活着
private:void MfSendThread(); // 发送线程，循环调用send，收发线程根据是否可用标志确定行为void MfRecvThread(); // 接收线程，循环调用recv，收发线程根据是否可用标志确定行为
};
2.4.2思路描述
①该类直译过来大概就是客户端到服务端连接的管理，其行为上类似于CService_XXX类。
②在该类中，连接列表，即socket列表只有一个。
③该类中也没有提供可覆写的虚函数，供用户自定义逻辑，仅提供某个连接的 发送 和 接受 两个行为，来对一个描述符读写，另外在内部封装了自动发送心跳包的机制。
④每个连接通过一个字符串来在map中索引，该字符串可以用来描述某个连接的行为。
⑤MfCreateAddLink & MfCloseLink函数提供了建立连接和关闭的接口。
⑥MfGetRecvBufferP & MfHasMsg & MfPopFrontMsg & MfLinkIsSurvive四个函数就是对连接收发数据的四个操作。
2.4.3部分成员函数及代码
void CClientLinkManage::MfStart()
该函数仅仅是简单启动收发两条线程。
MdIsStart这个变量是用来防止在收发线程启动前创建连接的，详见类定义。线程不放在构造中启动。
注：在深度探索C++对象模型中提到，进入对象的构造之前，首先会为成员对象调用构造函数（这里的 成员对象的构造函数 指的是那些用户提供构造的构造函数，而不是编译器生成的那些无效的构造）。那么之后进入构造函数后，在启动线程之前把所有成员值都有效化，接着启动线程这样的行为理论上是安全的，但是谁知道不同的编译器到底会不会有这些行为呢。
void CClientLinkManage::MfStart()
{MdBarrier->MfInit(3);// 启动收发线程,需要做的是在启动收发线程前不应该建立连接CThreadManage::MfCreateThreadAndDetach(std::bind(&CClientLinkManage::MfSendThread, this));CThreadManage::MfCreateThreadAndDetach(std::bind(&CClientLinkManage::MfRecvThread, this));MdBarrier->MfWait();MdIsStart = 1;
}
int CClientLinkManage::MfCreateAddLink(std::string Linkname, const char* ip, unsigned short port)
该函数通过 连接名 目标ip:port对，这三个参数来创建一个连接。
①首先判断收发线程是否启动
②然后判断该连接是否不存在
③条件都成立后申请一个代表客户端连接的对象，调用其连接函数。成功则加入队列并返回，失败则释放对象返回错误。
int CClientLinkManage::MfCreateAddLink(std::string Linkname, const char* ip, unsigned short port)
{if (!MdIsStart.load()){printf("启动收发线程后再创建连接。\n");return SOCKET_ERROR;}{std::shared_lock<std::shared_mutex> lk(MdClientLinkListMtx);if (MdClientLinkList.find(Linkname) != MdClientLinkList.end()){printf("service <%s> already existed!\n", Linkname.c_str());LogFormatMsgAndSubmit(std::this_thread::get_id(), ERROR_FairySun, "service <%s> already existed ", Linkname.c_str());return SOCKET_ERROR;}}CClientLink* temp = new CClientLink(Linkname);int ret = temp->MfConnect(ip, port);if (SOCKET_ERROR != ret) // 成功连接后就加入正式队列{std::unique_lock<std::shared_mutex> lk(MdClientLinkListMtx);MdClientLinkList.insert(std::pair<std::string, CClientLink*>(Linkname, temp));return ret;}delete temp;return SOCKET_ERROR;
}
void CClientLinkManage::MfCloseLink(std::string Linkname)
该函数尝试关闭一个连接，如果该连接在连接列表中，就关闭连接，随后移除列表。
void CClientLinkManage::MfCloseLink(std::string Linkname)
{std::unique_lock<std::shared_mutex> lk(MdClientLinkListMtx);auto it = MdClientLinkList.find(Linkname);if (it != MdClientLinkList.end()){it->second->MfClose();MdClientLinkList.erase(it);}
}
这四个函数提供了收发数据的接口，只是简单封装ClientLink的成员函数。都通过创建连接时，使用的连接名字来索引。
MfSendData函数添加一块数据到发送缓冲区。
MfGetRecvBufferP则直接返回接收缓冲区的地址以供读取。
MfHasMsg和MfPopFrontMsg是判断缓冲区中是否有数据和移除数据，操作的单位是包。关于包的描述见后。
bool CClientLinkManage::MfSendData(std::string name, const char* data, int len)
{return MdClientLinkList[name]->MfDataToBuffer(data, len);
}const char* CClientLinkManage::MfGetRecvBufferP(std::string name)
{return MdClientLinkList[name]->MfGetRecvBufferP();
}bool CClientLinkManage::MfHasMsg(std::string name)
{return MdClientLinkList[name]->MfHasMsg();
}void CClientLinkManage::MfPopFrontMsg(std::string name)
{return MdClientLinkList[name]->MfPopFrontMsg();
}
bool CClientLinkManage::MfLinkIsSurvive(std::string name)
该函数判断一个连接是否可用，条件是其在队列中能找到，同时通过一个成员函数判断。需要该成员函数判断的原因是：连接失效后没有及时被移除，如果只判断是否在列表中可能出现连接有效却发送失败的情况，第二个条件解决了这个问题。
bool CClientLinkManage::MfLinkIsSurvive(std::string name)
{std::shared_lock<std::shared_mutex> lk(MdClientLinkListMtx);auto it = MdClientLinkList.find(name);return it != MdClientLinkList.end() && it->second->MfGetIsConnect();
}
void CClientLinkManage::MfSendThread()
这个是被启动的发发送线程。
①主循环中首先是每个一段时间发送主动发送心跳包的操作。遍历客户端列表，对每个连接都发送一个心跳包出去。
②第二个循环是遍历所有客户端，为其情况第二缓冲区。
③在SendThread中不判断客户端是否失效，因为客户端可能很久都不送数据，在其内部也没有真正的去调用那个系统级别的send，也就没法判断socket是否失效，所以这个操作放在recv线程中。
void CClientLinkManage::MfSendThread()
{std::thread::id threadid = std::this_thread::get_id();printf("client send thread already start.\n");LogFormatMsgAndSubmit(threadid, INFO_FairySun, "client send thread already start.");MdBarrier->MfWait();// 主循环while (CThreadManage::MfGetThreadFlag(threadid)){// 为每个连接发送心跳包,这里是把心跳包加入第二缓冲区，之后一起由循环整个一起发送if (MdHeartTime->getElapsedSecond() > MdHeartSendInterval){std::shared_lock<std::shared_mutex> lk(MdClientLinkListMtx);for (auto it = MdClientLinkList.begin(); it != MdClientLinkList.end(); ++it)it->second->MfDataToBuffer((const char*)MdDefautHeartPacket, sizeof(CNetMsgHead));MdHeartTime->update();}// 加括号使锁提前释放{std::shared_lock<std::shared_mutex> lk(MdClientLinkListMtx);for (auto it = MdClientLinkList.begin(); it != MdClientLinkList.end(); ++it){it->second->MfSend(); // 出错移除操作放在recv进程操作}}std::this_thread::sleep_for(std::chrono::milliseconds(1));}printf("client send thread already over.\n");LogFormatMsgAndSubmit(threadid, INFO_FairySun, "client send thread already over.");
}
这个是被启动的接收线程。
①主循环就是单纯的为一个连接尝试接收数据，并判断连接是否失效，失效就将其移除。
②这里有一个将共享读锁提升为独占写锁的操作，如果发现了失效连接，就先声明一个独占锁，使用std::adopt_lock标志和先前的锁共享所有权，随后调用读锁的release函数，其行为是接触锁和互斥元的关联，然后将失效连接从列表中移除。
void CClientLinkManage::MfRecvThread()
{std::thread::id threadid = std::this_thread::get_id();printf("client recv thread already start.\n");LogFormatMsgAndSubmit(threadid, INFO_FairySun, "client recv thread already start.");std::map<std::string, CClientLink*>::iterator TmpIt;MdBarrier->MfWait();// 主循环while (CThreadManage::MfGetThreadFlag(threadid)){{std::shared_lock<std::shared_mutex> lk(MdClientLinkListMtx);for (auto it = MdClientLinkList.begin(); it != MdClientLinkList.end();){TmpIt = it++;if (0 >= TmpIt->second->MfRecv()){std::unique_lock<std::shared_mutex> uk(MdClientLinkListMtx, std::adopt_lock);lk.release(); // 解除lk和互斥元的关联，这样lk析构时就不会第二次对互斥元解锁了TmpIt->second->MfClose();delete TmpIt->second;MdClientLinkList.erase(TmpIt->first);}}}std::this_thread::sleep_for(std::chrono::milliseconds(1));}printf("client recv thread already over.\n");LogFormatMsgAndSubmit(threadid, INFO_FairySun, "client recv thread already over.");
}
2.5.CClientLink
2.5.1类定义
class CClientLink
{private:std::string MdLinkName; // 服务名称，内容CSocketObj* MdClientSock; // 客户连接对象std::atomic<int> MdIsConnect = 0; // 表示是否连接成功
private:
public:CClientLink(std::string s = "未命名");~CClientLink();int MfConnect(const char* ip, unsigned short port); // 发起一个连接int MfClose(); // 关闭一个连接SOCKET MfGetSocket(); // 返回描述符std::string MfGetSerivceNmae(); // 返回当前服务的名称int MfGetIsConnect(); // 当前服务是否连接int MfRecv(); // 供服务管理者调用接收数据 只是简单调用第二缓冲区的成员函数int MfSend(); // 供服务管理者调用发送数据 只是简单调用第二缓冲区的成员函数bool MfDataToBuffer(const char* data, int len); // 供调用者插入数据，插入数据到发送缓冲区const char* MfGetRecvBufferP(); // 供使用者处理数据，取得接收缓冲区原始指针 只是简单调用第二缓冲区的成员函数bool MfHasMsg(); // 供使用者处理数据，缓冲区是否有消息 只是简单调用第二缓冲区的成员函数void MfPopFrontMsg(); // 供使用者处理数据，第一条信息移出缓冲区 只是简单调用第二缓冲区的成员函数
};
2.5.2思路描述
该类只是简单的封装class CSocketObj;对象，后者代表一个连接，前者在其基础上增加了服务名称和一个标志,方便在class CClientLinkManage中操作。
2.6.3部分成员函数及代码
这部分只描述两个成员函数，其余成员函数基本都只是封装class CSocketObj;的成员函数 或者 只是简单返回成员变量。
int CClientLink::MfConnect(const char* ip, unsigned short port)
①该函数在CClientLinkManage::MfCreateAddLink被调用，用于主动发起到某个ip:port对的连接。
②首先是防止socket函数调用失败
③接着是最多尝试连接3次，失败后返回错误，成功后为其设置非阻塞模式。
int CClientLink::MfConnect(const char* ip, unsigned short port)
{auto Id = std::this_thread::get_id();int ret = -1;sockaddr_in addr{};addr.sin_family = AF_INET;addr.sin_port = htons(port);
#ifndef WIN32if (ip) addr.sin_addr.s_addr = inet_addr(ip);else addr.sin_addr.s_addr = INADDR_ANY;
#elseif (ip) addr.sin_addr.S_un.S_addr = inet_addr(ip);else addr.sin_addr.S_un.S_addr = INADDR_ANY;
#endifSOCKET CliSock = socket(AF_INET, SOCK_STREAM, 0);if (SOCKET_ERROR == CliSock){LogFormatMsgAndSubmit(Id, ERROR_FairySun, "client create listenSock failed!");std::this_thread::sleep_for(std::chrono::seconds(LOG_SYN_TIME_SECOND + 1)); // 比日志同步时间多一秒，保证提交的日志被写到硬盘return SOCKET_ERROR;}for (int i = 0; i < 3; ++i){printf("service <%s> %d st linking...\n", MdLinkName.c_str(), i + 1);ret = connect(CliSock, (sockaddr*)&addr, sizeof(sockaddr_in));if (SOCKET_ERROR == ret){printf("service <%s> %d st link fail，errno:<%d>. errnomsg:<%s>\n", MdLinkName.c_str(), i + 1, errno, strerror(errno));LogFormatMsgAndSubmit(Id, ERROR_FairySun, "service <%s> %d st link fail，errno:<%d>. errnomsg:<%s>", MdLinkName.c_str(), i + 1, errno, strerror(errno));continue;}else {printf("service <%s> %d st link success\n", MdLinkName.c_str(), i + 1);LogFormatMsgAndSubmit(Id, ERROR_FairySun, "service <%s> %d st link success", MdLinkName.c_str());#ifdef WIN32 unsigned long ul = 1;ioctlsocket(CliSock, FIONBIO, &ul);
#elseint flags = fcntl(CliSock, F_GETFL, 0);fcntl(CliSock, F_SETFL, flags | O_NONBLOCK);
#endifMdClientSock = new CSocketObj(CliSock);MdIsConnect = 1;break;}}return ret;
}
int CClientLink::MfClose()
这个函数在class CClientLinkManage中，主动关闭连接或者被动发现连接失效被移出队列时调用，主要是先设置MdIsConnect标志，然后关闭描述符，清理结构
int CClientLink::MfClose()
{if (MdIsConnect){MdIsConnect = 0;MdClientSock->MfClose();}if (nullptr != MdClientSock)delete MdClientSock;MdClientSock = nullptr;return 0;
}
2.6.CSocketObj
2.6.1类定义与简单描述
该类几乎全部成员函数都是从成员对象封装而来或者是对成员变量的操作。在此不列任何成员函数的代码，其作用见注释。
这里主要是两个第二缓冲区，一个用于接收数据，一个用于发送数据。
成员MdThreadIndex在服务端这边被使用，服务端会启动多个处理线程，该变量表示当前对象表示的连接位于哪个线程。
成员MdIP和MdPort则在服务端用来存储客户端的ip和端口。
class CSocketObj
{private:SOCKET MdSock; // socketint MdThreadIndex; // 当前socket对象位于线程的线程索引，哪个dispose线程，service用CSecondBuffer* MdPSendBuffer; // 发送缓冲区CSecondBuffer* MdPRecvBuffer; // 接收缓冲区CTimer* MdHeartBeatTimer; // 心跳计时器char MdIP[20]; // sock对端的地址int MdPort; // sock对端的地址
public:CSocketObj(SOCKET sock, int SendBuffLen = DEFAULT_BUFF_LEN, int RecvBuffLen = DEFAULT_BUFF_LEN);~CSocketObj();
public:SOCKET MfGetSock() { return MdSock; }char* MfGetRecvBufP(); // 返回接收缓冲区原始指针int MfRecv(); // 为该对象接收数据int MfSend(); // 为该对象发送数据void MfSetPeerAddr(sockaddr_in* addr); // 设置对端IP和端口char* MfGetPeerIP(); // 获取对端IPint MfGetPeerPort(); // 获取对端端口void MfSetThreadIndex(int index); // 设置线程索引int MfGetThreadIndex(); // 获取线程索引，哪个dispose线程，service用bool MfDataToBuffer(const char* data, int len); // 压数据到发送缓冲区bool MfHasMsg(); // 接收缓冲区是否有消息void MfPopFrontMsg(); // 第一条信息移出接收缓冲区void MfHeartBeatUpDate(); // 更新心跳计时bool MfHeartIsTimeOut(int seconds); // 传入一个秒数，返回是否超过该值设定的时间int MfClose(); // 关闭套接字
};
2.7.CSecondBuffer
2.7.1类定义
class CSecondBuffer
{private:char* MdPBuffer = nullptr; // 缓冲区指针int MdBufferLen = -1; // 缓冲区长度int Mdtail = 0;std::mutex MdMtx; // 操作数据指针时需要的互斥元CSecondBuffer();
public:CSecondBuffer(int bufferlen = DEFAULTBUFFERLEN);~CSecondBuffer();char* MfGetBufferP(); // 返回缓冲区原始指针以供操作bool MfDataToBuffer(const char* data, int len); // 写数据到缓冲区int MfBufferToSocket(SOCKET sock); // 数据从 缓冲区 写到 套接字，返回值主要用于判断socket是否出错，出错条件是send或recv返回值<=0int MfSocketToBuffer(SOCKET sock); // 数据从 套接字 写到 缓冲区，返回值主要用于判断socket是否出错，出错条件是send或recv返回值<=0bool MfHasMsg(); // 缓冲区中是否够一条消息的长度void MfPopFrontMsg(); // 弹出缓冲区中的第一条消息void MfBufferPopData(int len); // 缓冲区中数据弹出private:bool MfSend(SOCKET sock, const char* buf, int len, int* ret); // 封装SEND和RECV调用其非阻塞模式，并处理两个问题bool MfRecv(SOCKET sock, void* buf, int len, int* ret); // 返回值用来区分ret值结果参数是否可用，如果为1表示ret的值应该被用于更新tail，反之则不应该更新
};
2.7.2简单描述
①这一个第二缓冲区类既被用作发送缓冲区，又被用作接收缓冲区，所以同时提供了两者操作的接口。
②MfDataToBuffer & MfBufferToSocket两个函数提供了用户写数据到缓冲区 和 把缓冲区中的数据写到socket的缓冲区的操作。
③MfSocketToBuffer & MfBufferPopData两个函数提供了从socket缓冲区提取数据到第二缓冲区 和 从第二缓冲区数据移出数据的操作。
④MfSend 和 MfRecv封装了send和recv这两个系统调用，处理了非阻塞套接字一些特殊情况，以及被中断的系统调用的情况。
⑤另外考虑到可能会多线程操作，声明了一个互斥元变量。
2.7.3部分成员函数及代码
MfDataToBuffer & MfBufferToSocket
这两个函数一个让用户把数据塞到第二缓冲区，一个把第二缓冲区中的数据塞到对应的socket缓冲区中。
MfDataToBuffer中只是简单上锁，复制数据，然后修改数据的索引。
MfBufferToSocket比较复杂一些，
①首先使用了尝试锁定，锁定该缓冲区，这里使用尝试锁定的原因当然是为了防止被卡住，按照脑子里的设想，如果第一轮没有成功锁定，第二轮几乎就能锁定了。
②打个比方吧，比如说接收线程、处理线程两个同时访问一个列表(该列表中是CSocketObj对象)。收线程和处理线程会同时遍历该列表，因为他们都是读锁(该锁锁定的是对象列表，保证列表不被修改，而不是其中的缓冲区)，会出现两条线程同时遍历到同一个CSocketObj对象，然后两条线程又在内部尝试对其缓冲区对象上锁，那么一个线程锁住该缓冲区，一个线程因为锁不上而跳过，去遍历其他对象，两条线程的下一轮能换虽然还有可能遇上同时锁定一个对象的情况，但是两轮循环恰好都是一个对象的几率就有些小了。
注：明明是发送缓冲区的代码，却举了个接收缓冲区的例子，不过能说明意思就行了。
③至于CSecondBuffer::MfSend这个函数，暂时只需要知道其返回值被用来判断 作为第四参数的另一个返回值是否可用 即可。
bool CSecondBuffer::MfDataToBuffer(const char* data, int len)
{std::lock_guard<std::mutex> lk(MdMtx);if (Mdtail + len <= MdBufferLen){memcpy(MdPBuffer + Mdtail, data, len);Mdtail += len;return true;}return false;
}int CSecondBuffer::MfBufferToSocket(SOCKET sock)
{int ret = 0;std::unique_lock<std::mutex> lk(MdMtx, std::defer_lock);if (lk.try_lock()) // 锁定缓冲区{if (Mdtail > 0) // 有数据时{if (MfSend(sock, MdPBuffer, Mdtail, &ret)) // ret值可用时{if (ret <= 0) // 如果该值小于等于0，代表socket出错 或者 对端关闭，原样返回ret值return ret;Mdtail -= ret; // 否则就是成功写入socket缓冲区，更新值memcpy(MdPBuffer, MdPBuffer + ret, Mdtail); // 缓冲区中未被发送的数据移动到缓冲区起始位置}}}// 只有在成功锁定后 && 缓冲区有数据 && send返回值可用 同时成立的情况下// 才返回真实的ret，否则就是返回INT32_MAX，来表示没有出错，但是不代表接受了数据，只被调用者用于判断是否出错return INT32_MAX;
}
以下四个函数提供了从socket缓冲区到第二缓冲区、判断第二缓冲区中是否有玩获赠的消息包、数据包从缓冲区弹出的操作。
MfSocketToBuffer和前面一样使用尝试锁定，如果在MfRecv就更新缓冲区索引值。
MfHasMsg中首先判断是否够一个消息头的长度，如够就直接将第二缓冲区转型为消息头类型，取得该消息包长度，随后判断该消息包在第二缓冲区是否完整。
MfBufferPopData:消息包从第二缓冲区中被移出时，其中剩余的所有消息又被全部迁移到缓冲区的头部。(这里一开始想着用循环数组来避免一次次复制移动的，但是后来可能没写好，过程一度十分折磨，退回到复制移动的版本后少了150行，就这样一直没动了。)
int CSecondBuffer::MfSocketToBuffer(SOCKET sock)
{// 没有出错时返回的值都是大于等于0的，但是返回值是INT32_MAX时，没有出错，但是也没有成功压入数据int ret = 0;std::unique_lock<std::mutex> lk(MdMtx, std::defer_lock);if (lk.try_lock()) // 锁定缓冲区{if (MdBufferLen - Mdtail > 0) // 缓冲区又空间{if (MfRecv(sock, MdPBuffer + Mdtail, MdBufferLen - Mdtail, &ret)) // ret值可用时{if (ret <= 0) // 如果该值小于等于0，代表socket出错 或者 对端关闭，原样返回ret值return ret;Mdtail += ret; // 否则就是成功写入socket缓冲区，更新值return ret;}}}// 只有在成功锁定后 && 剩余空间大于0 && recv返回值可用 同时成立的情况下// 才返回真实的ret，否则就是返回INT32_MAX，来表示没有出错，但是不代表接受了数据，只被调用者用于判断是否出错return INT32_MAX;
}bool CSecondBuffer::MfHasMsg()
{static int MSG_HEAD_LEN = sizeof(CNetMsgHead);if (Mdtail >= MSG_HEAD_LEN)return Mdtail >= ((CNetMsgHead*)MdPBuffer)->MdLen;return false;
}void CSecondBuffer::MfPopFrontMsg()
{if (MfHasMsg())MfBufferPopData( ((CNetMsgHead*)MdPBuffer)->MdLen );
}void CSecondBuffer::MfBufferPopData(int len)
{std::lock_guard<std::mutex> lk(MdMtx);int n = Mdtail - len;//printf("tail:%d, len:%d\n", Mdtail, len);if (n >= 0){memcpy(MdPBuffer, MdPBuffer + len, n);Mdtail = n;}
}
这一块虽然内容不多，但是确实最重要的关于send和recv的调用。两个函数处理的内容基本一样，这里只说一个。
①调用send后，如果其返回值大于等于0，那么send没有出错且成功接收或者对端关闭，返回true，表示第四参数的ret值可以被使用。
②如果返回值小于0，就需要分别判断几种情况，其中前两种情况是非阻塞套接字需要处理的情况，这里直接贴出结论，详细描述见第四条。
linux平台：非阻塞套接字，send在发送缓冲区满 和 recv在接收缓冲区空时，返回-1，errno被设置为EWOULBLOCK(11)
win平台：非阻塞套接字，send在发送缓冲区满 和 recv在接收缓冲区空时，返回-1，errno值为0
③当返回值小于0，且errno值为EINTR时，代表 被中断的系统调用，，这个也不是错误，提交下日志随后返回。具体说就是send、recv、write、write等陷入内核态的函数调用，可能永远无法返回，当linux系统产生信号，转而去调用信号处理函数，然后这些函数调用被迫返回，返回值是一个赋值，errno则被设置为EINTR。另见第五章第一条。
bool CSecondBuffer::MfSend(SOCKET sock, const char* buf, int len, int * ret)
{*ret = send(sock, buf, len, 0);if (0 <= *ret) // 大于等于0时，要么对端正确关闭，要么发送成功，都使返回值可用，返回truereturn true;else // 否则处理errno返回0，发送缓冲区满发送失败，被系统调用打断{if (0 == errno) return false;if (EWOULDBLOCK == errno) // 非阻塞模式，socket发送缓冲区已满的情况，这种情况下，不是错误，返回false，ret值不应该被用于更新tailreturn false;if (EINTR == errno) // 同样，EINTR不是错误，返回false，表示返回值不可用，ret值不应该被用于更新tail{LogFormatMsgAndSubmit(std::this_thread::get_id(), WARN_FairySun, "SOCKET <%5d> SEND be interrupted, errno is EINTR", sock);return false;}// 这种情况下，返回值小于0，错误，ret值是send的返回值，返回false，ret值不可用，表示未解析的错误LogFormatMsgAndSubmit(std::this_thread::get_id(), ERROR_FairySun, "SOCKET <%5d> SEND RETURN VAL:<%d>, errno(%d) : %s", sock, *ret, errno, strerror(errno));return false;}
}bool CSecondBuffer::MfRecv(SOCKET sock, void* buf, int len,int* ret)
{#ifndef WIN32*ret = recv(sock, buf, len, 0);
#else*ret = recv(sock, (char *)buf, len, 0);
#endif // !WIN32if (0 <= *ret) // 大于等于0时，要么错误正确接收，要么对端关闭，返回值都可用，返回truereturn true;else // 否则处理recv被系统调用打断，非阻塞接收缓冲区空，以及errno为0的情况{if (0 == errno) return false;if (EWOULDBLOCK == errno) // 非阻塞模式，socket接收缓冲区空的情况，这种情况下，不是错误，返回false，ret值不应该被用于更新tail或headreturn false;if (EINTR == errno) // 同样，EINTR不是错误，返回false，返回值不可用，ret值不应该被用于更新tail或head{LogFormatMsgAndSubmit(std::this_thread::get_id(), WARN_FairySun, "SOCKET <%5d> RECV be interrupted, errno is EINTR", sock);return false;}// 这种情况下，返回值小于0，错误，ret值是send的返回值，返回false，ret值不可用，表示未解析的错误LogFormatMsgAndSubmit(std::this_thread::get_id(), ERROR_FairySun, "SOCKET <%5d> RECV RETURN VAL:<%5d>, errno(%d) : %s", sock, *ret, errno, strerror(errno));return false;}
}
2.8.CThreadManage
2.8.1类定义
class CThreadManage
{private:static CThreadManage* MdPInstance; // 单例对象指针static std::map<std::thread::id, bool> MdAllIdAndStopFlag; // 每个线程的ID对应一个是否运行的bool值static std::thread::id MdLogThreadId; // 日志线程ID，特殊对待static std::atomic<bool> MdLogThreadStopFlag; // 日志线程标志static Barrier MdBarrier1; // 日志线程结束前用于等待其他线程的屏障
public: static Barrier MdBarrier2; // 等待日志线程结束private:CThreadManage() {};CThreadManage(const CThreadManage&) = delete;CThreadManage& operator=(const CThreadManage&) = delete;
public:~CThreadManage() {};static CThreadManage* MfGetInstance(); // 返回对象指针static void MfCreateLogThreadAndDetach(void(*p)()); // 因为日志线程的特殊性，对于日志线程特别启动static bool MfGetLogThreadFlag(); // 取得日志线程的flagstatic void MfCreateThreadAndDetach(std::function<void(void)> p); // 使用线程管理对象启动一个普通线程static void MfPackerFun(std::function<void(void)> p); // 启动线程时用这个函数包裹一下，原因是需要这个函数来实现一个类似屏障的东西static bool MfGetThreadFlag(std::thread::id threadid); // 取得某个线程的flagstatic void MfThreadManageOver(); // 最终在程序结束时，由库控制的那个函数调用，行为是停止所有线程，清空日志线程等
};
2.8.2思路描述
该类使用了单例模式，在函数FairySunOfNetBaseStart()中被实例化，其构造设为私有，复构和赋值被禁用。
成员和成员函数全部使用了static方式以方便使用。
该类和日志类都是单例对象，并且因为日志线程的特殊性，总是要最后才结束日志线程，所以这里和日志类的耦合度有些高。
每启动一个线程，该类中就会为对应线程产生一个标志。被启动线程主循环使用该标志来判断其是否继续循环。
当程序结束时，会依次把所有线程的标志都设为false，以此方式来控制所有线程结束，在等待所有线程结束后，最后结束日志线程。
比如说被启动线程内部是一个如下的格式：
std::thread::id threadid = std::this_thread::get_id();
while (CThreadManage::MfGetThreadFlag(threadid))
{}
2.8.3
这一块是定义类中的静态成员，需要将其定义在.cpp文件。
CThreadManage* CThreadManage::MdPInstance = nullptr;
std::map<std::thread::id, bool> CThreadManage::MdAllIdAndStopFlag{};
std::thread::id CThreadManage::MdLogThreadId;
std::atomic<bool> CThreadManage::MdLogThreadStopFlag{};
Barrier CThreadManage::MdBarrier1;
Barrier CThreadManage::MdBarrier2;
以下两个函数专门用于启动日志线程以及获取其结束标志。
void CThreadManage::MfCreateLogThreadAndDetach(void(*p)())
{MfGetInstance(); // 调用一次保证对象被构造std::thread t(p);MdLogThreadId = t.get_id();MdLogThreadStopFlag = true;t.detach();
}bool CThreadManage::MfGetLogThreadFlag()
{return MdLogThreadStopFlag;
}
以下三个函数，提供了启动普通线程的操作。
MfCreateThreadAndDetach参数是一个std::function<void(void)>的函数签名，在外部使用时，不管被启动的线程是否有参数，都使用std::bind()将其包装成void(void)的形式。另外在启动线程时，又使用了成员函数MfPackerFun将其包裹一层，其存在的意义是为了 日志线程 结束之前 等待其他所有线程结束。
MfGetThreadFlag则是获取对应线程的标志。
MfPackerFun，在该函数，被启动的线程函数p终于开始真正执行，当其内部的主循环因为标志变为false结束，执行到MdBarrier1.MfWait();，在此等待其他所有线程结束。
void CThreadManage::MfCreateThreadAndDetach(std::function<void(void)> p)
{MfGetInstance(); // 调用一次保证对象被构造std::thread t(CThreadManage::MfPackerFun, p);MdAllIdAndStopFlag[t.get_id()] = true;t.detach();
}bool CThreadManage::MfGetThreadFlag(std::thread::id threadid) // 取得某个线程的flag
{if (MdPInstance != nullptr)return MdAllIdAndStopFlag[threadid];return false;
}void CThreadManage::MfPackerFun(std::function<void(void)> p)
{p();MdBarrier1.MfWait();
}
MfThreadManageOver()该函数在程序结束时被FairySunOfNetBaseOver()调用，正确的等待所有线程，最后结束日志线程。
①Barrier MdBarrier1；这个类是根据linuxpthread_barrier_xxx等模仿的一个win和linux都可用的线程屏障，用于等待多个线程。详见后对应章节。
②遍历被启动的线程id列表，将其标志全部设为false，使其主循环结束，退出，虽然等待所有线程借宿。
③之后设置日志线程的标志使退出主循环，接着等待日志线程正常结束。
④最后，释放线程管理的单例对象。
void CThreadManage::MfThreadManageOver()
{printf("ThreadManage OverIng Program!\n");LogFormatMsgAndSubmit(std::this_thread::get_id(), INFO_FairySun, "ThreadManage OverIng Program!");// 这一块代码用来在结束日志线程前等待并结束其他线程!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!MdBarrier1.MfInit(MdAllIdAndStopFlag.size() + 1);for (auto it = MdAllIdAndStopFlag.begin(); it != MdAllIdAndStopFlag.end(); ++it) // 停止日志线程之外的其他线程it->second = false;MdBarrier1.MfWait();printf("all thread arrived barried.\n");// 这一块代码用来等待和结束日志线程!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!MdBarrier2.MfInit(2);MdLogThreadStopFlag = false; // 停止日志线程MdBarrier2.MfWait();if (!MdPInstance){delete MdPInstance;MdPInstance = nullptr;}
}
2.9.CLog & enum EMsgLevel & struct CLogMsg & CDoubleBufLogQueue
2.9.1定义
const int LOG_SYN_TIME_SECOND = 3; // 日志同步时间
const int LOG_MSG_LEN = 256; // 日志消息长度struct CLogMsg
{char MdMsg[LOG_MSG_LEN];
};enum EMsgLevel
{DEBUG_FairySun,INFO_FairySun,WARN_FairySun,ERROR_FairySun,FATAL_FairySun
};class CDoubleBufLogQueue
{private:std::mutex* MdMtx; // 交换两个队列时使用的互斥元std::queue<CLogMsg*>* MdBuf1; // 日志读线程使用buf1std::queue<CLogMsg*>* MdBuf2; // 另外的写进程使用buf2public:CDoubleBufLogQueue();~CDoubleBufLogQueue();inline std::mutex* MfGetMtx();inline std::queue<CLogMsg*>* MfGetBuf1();inline std::queue<CLogMsg*>* MfGetBuf2();void MfSwapBuf();
};class CLog
{private:static CLog* MdPInstance; // 单例对象指针
public:static std::thread::id MdLogThreadId; // 当前日志线程IDstatic CMemoryPool<CLogMsg> MdLogMsgObjectPool; // 日志消息对象池static std::map<std::thread::id, CDoubleBufLogQueue> MdThreadIdAndLogQueue; // 每个线程ID对应双缓冲的消息队列static std::atomic<long long int> MdMsgId; // 消息Idprivate:CLog() {};
public:~CLog();CLog(const CLog&) = delete;CLog& operator=(const CLog&) = delete;public:static CLog* MfGetInstance(); // 返回对象指针static void MfThreadRun(); // 遍历所有双缓冲的消息队列，并写入文件，同时负责生成日志目录和日志文件static void MfSubmitLog(std::thread::id ThreadId, CLogMsg* Log); // 真正提交日志的函数static CLogMsg* MfApplyLogObject(); // 该函数存在的意义，因为LogFormatMsgAndSubmit需要模板可变参，但是MdLogMsgObjectPool是静态成员，定义在头文件会报重定义错，模板函数写在cpp文件又会无法解析，所以写个函数包一层static void MfLogSort(); // 日志排序
};
2.9.2简介
CLogMsg就是简单将char数组封装来存储日志一条消息，目的是方便操作。
EMsgLevel是日志消息等级的定义。
CDoubleBufLogQueue类就是两个简单的队列，主要是为了降低提交日志消息时并发区的大小。在这里简单描述以下该类，随后不再描述：两个成员指针分别指向一个日志队列，在日志线程中总是使用指针MdBuf1，在其他线程提交日志时，总是使用MdBuf2，日志线程中，会根据一定条件交换两个指针所指的队列，这样就将并发区的大小从整个日志队列减小到两个指针的交换。
以上三个类是为了支撑CLog类，CLog就是真正的日志处理的那一块地方。
在CLog的中，每个线程都会有一个对应的双缓冲队列，这么做的原因当然也是为了减少并发区使用的范围，也算空间换时间把，因为这么做的话每个线程都只是和日志线程来并发访问，而不需要和其他线程竞争。
2.9.3部分成员函数及代码
同样，这里是日志类中静态成员的定义。其中有一个内存池，见后。
CLog* CLog::MdPInstance = nullptr;
CMemoryPool<CLogMsg> CLog::MdLogMsgObjectPool(1000);
std::thread::id CLog::MdLogThreadId{};
std::map<std::thread::id, CDoubleBufLogQueue> CLog::MdThreadIdAndLogQueue{};
std::atomic<long long int> CLog::MdMsgId{};
以下代码块是函数void CLog::MfThreadRun()的核心部分，因为该函数过长，所以省略了其他地方，这里是将日志消息从队列中写到文件的部分，其他线程提交日志消息队列到线程见后。
①主循环中第一个if是没过一个小时重新创建一个日志文件。
②主循环中接下来的for就是遍历每个线程对应的双缓冲列表。
③如果两个指针所指的日志消息队列都空就稍微停一下再跳到下一个。
④如果不是两个都空，就交换两个指针所指的队列，随后将队列中所有消息写入文件。
⑤写入文件的日志可能暂时还在缓存中，每个一段时间主动将缓存同步到磁盘。
⑥最后，因为是多线程写入同一个文件，日志消息顺序再文件中是乱的，调用另外一个函数使其有序。
// 每隔一段时间日志重新写一个文件printf("LogThread will start.\n");while (CThreadManage::MfGetLogThreadFlag()){// 这块代码用来每隔一个小时重新建立一个日志文件if (3600 < Time.getElapsedSecond()) {fclose(LogFileStream); // 关闭原来的文件流memset(FinalLogFileName, 0, 256); // 清空文件名重新拼接strcat(FinalLogFileName, LogFileNameFront);strcat(FinalLogFileName, YearMonthDayHourMinuteSecondStr().c_str());strcat(FinalLogFileName, LogFileNameRear);//printf("LogNewFlieName:%s\n", FinalLogFileName); // 调试时打印文件名LogFileStream = fopen(FinalLogFileName, "wb"); // 打开文件流if (nullptr == LogFileStream){//printf("!!!!!Fatal error : create NewLogFile fail222\n");LogFormatMsgAndSubmit(std::this_thread::get_id(), FATAL_FairySun, "!!!!!Fatal error : create NewLogFile fail222");ErrorWaitExit();}Time.update();}// 接下来就是每次循环把所有缓冲区先交换，紧随其后通过流把日志消息写入文件for (auto it1 = MdThreadIdAndLogQueue.begin(); it1 != MdThreadIdAndLogQueue.end(); ++it1){CLogMsg* Temp;if (it1->second.MfGetBuf1()->empty() && it1->second.MfGetBuf2()->empty()){std::this_thread::sleep_for(std::chrono::milliseconds(1));continue;}it1->second.MfSwapBuf();while (!it1->second.MfGetBuf1()->empty()){Temp = it1->second.MfGetBuf1()->front();fputs(Temp->MdMsg, LogFileStream);//printf("%p ：%s", it1->second.MfGetBuf1(), Temp->MdMsg);it1->second.MfGetBuf1()->pop();MdLogMsgObjectPool.MfReturnMemory(Temp);//delete Temp;}if (LOG_SYN_TIME_SECOND < TimeSyn.getElapsedSecond()){fflush(LogFileStream);TimeSyn.update();}}}// 日志线程主循环退出后，应该清空所有缓冲区for (auto it1 = MdThreadIdAndLogQueue.begin(); it1 != MdThreadIdAndLogQueue.end(); ++it1){CLogMsg* Temp;while (!it1->second.MfGetBuf1()->empty()){Temp = it1->second.MfGetBuf1()->front();fputs(Temp->MdMsg, LogFileStream);MdLogMsgObjectPool.MfReturnMemory(Temp);it1->second.MfGetBuf1()->pop();}it1->second.MfSwapBuf();while (!it1->second.MfGetBuf1()->empty()){Temp = it1->second.MfGetBuf1()->front();fputs(Temp->MdMsg, LogFileStream);MdLogMsgObjectPool.MfReturnMemory(Temp);it1->second.MfGetBuf1()->pop();}}fclose(LogFileStream);// 最后文件关闭后，调用日志排序MfLogSort();CThreadManage::MdBarrier2.MfWait();
以下两个函数线程提交日志消息到队列的两个函数，其中后者调用前者，用户只使用后者，后者LogFormatMsgAndSubmit则是为用户提供了一个变参接口的提交方式。
在MfSubmitLog中，提交日志前先上锁，防止在提交过程中指针指向的队列被交换。另外再说明一边，在把日志消息从队列写到文件那个函数里总是使用指针MdBuf1，而这里提交日志的地方总是使用MdBuf2.
在LogFormatMsgAndSubmit中，只是简单的把日志消息的id、等级、时间、和用户提供的字串以及可变参拼接到一个CLogMsg对象中。这里使用了内存池的方式申请，提交日志时从中申请一块内存存储CLogMsg对象，在日志消息写到文件后这一块内存被归还给内存池。
void CLog::MfSubmitLog(std::thread::id ThreadId, CLogMsg* Log)
{std::lock_guard<std::mutex> lk(*(CLog::MdThreadIdAndLogQueue[ThreadId].MfGetMtx()));
// CLog::MdThreadIdAndLogQueue[ThreadId]; // 这一行用来保证map中的第二参数被初始化CLog::MdThreadIdAndLogQueue[ThreadId].MfGetBuf2()->push(Log);
}template <typename ...Arg>
void LogFormatMsgAndSubmit(std::thread::id Id, EMsgLevel Level, const char* format, Arg... args) // 暴露给外部按格式生成日志并提交的函数
{const time_t t = time(NULL);struct tm Time;
#ifndef WIN32localtime_r(&t, &Time);
#elselocaltime_s(&Time, &t);
#endifCLogMsg* Temp = CLog::MfApplyLogObject();//CLogMsg_FairySun* Temp = new CLogMsg_FairySun;switch (Level){case DEBUG_FairySun:snprintf(Temp->MdMsg, LOG_MSG_LEN, "MsgId:%20lld\t%6s\t%4d-%02d-%02d_%02d-%02d-%02d\tTid:%11x\t", ++CLog::MdMsgId, "DEBUG", Time.tm_year + 1900, Time.tm_mon + 1, Time.tm_mday, Time.tm_hour, Time.tm_min, Time.tm_sec, std::hash<std::thread::id>{}(std::this_thread::get_id()));break;case INFO_FairySun:snprintf(Temp->MdMsg, LOG_MSG_LEN, "MsgId:%20lld\t%6s\t%4d-%02d-%02d_%02d-%02d-%02d\tTid:%11x\t", ++CLog::MdMsgId, "INFO", Time.tm_year + 1900, Time.tm_mon + 1, Time.tm_mday, Time.tm_hour, Time.tm_min, Time.tm_sec, std::hash<std::thread::id>{}(std::this_thread::get_id()));break;case WARN_FairySun:snprintf(Temp->MdMsg, LOG_MSG_LEN, "MsgId:%20lld\t%6s\t%4d-%02d-%02d_%02d-%02d-%02d\tTid:%11x\t", ++CLog::MdMsgId, "WARN", Time.tm_year + 1900, Time.tm_mon + 1, Time.tm_mday, Time.tm_hour, Time.tm_min, Time.tm_sec, std::hash<std::thread::id>{}(std::this_thread::get_id()));break;case ERROR_FairySun:snprintf(Temp->MdMsg, LOG_MSG_LEN, "MsgId:%20lld\t%6s\t%4d-%02d-%02d_%02d-%02d-%02d\tTid:%11x\t", ++CLog::MdMsgId, "ERROR", Time.tm_year + 1900, Time.tm_mon + 1, Time.tm_mday, Time.tm_hour, Time.tm_min, Time.tm_sec, std::hash<std::thread::id>{}(std::this_thread::get_id()));break;case FATAL_FairySun:snprintf(Temp->MdMsg, LOG_MSG_LEN, "MsgId:%20lld\t%6s\t%4d-%02d-%02d_%02d-%02d-%02d\tTid:%11x\t", ++CLog::MdMsgId, "FATAL", Time.tm_year + 1900, Time.tm_mon + 1, Time.tm_mday, Time.tm_hour, Time.tm_min, Time.tm_sec, std::hash<std::thread::id>{}(std::this_thread::get_id()));break;}auto TempInt = strlen(Temp->MdMsg);
#ifdef WIN32_snprintf_s(Temp->MdMsg + TempInt, LOG_MSG_LEN - TempInt - 1, 255, format, args...);
#elsesnprintf(Temp->MdMsg + TempInt, LOG_MSG_LEN - TempInt - 1, format, args...);
#endif // !WIN32strcat(Temp->MdMsg, "\n");CLog::MfSubmitLog(Id, Temp);
}
2.10.union CObj & CObjectPool CMemoryPool
2.10.1定义
union CObj
{CObj* MdNext;char* MdData;
};template <class T>
class CMemoryPool
{private:int MdObjectLen; // 一个T类型对象的长度int MdNumbers; // 能分配多少个对象char* MdPPool; // 整个内存池所占用的空间地址起点CObj* MdMemoryLinkHead; // 内存池被转化为链表后的链表头std::mutex MdMtx; // 使用该内存池时的锁int MdLackCount; // 内存池使用不足时计数
public:CMemoryPool(int numbers); ~CMemoryPool(); CMemoryPool(const CMemoryPool&) = delete;CMemoryPool& operator=(const CMemoryPool&) = delete;T* MfApplyMemory(); // 申请一块内存void MfReturnMemory(T* obj); // 归还一块内存
};template <class T>
class CObjectPool
{private:CMemoryPool<T> MdMemoryPool;
public:CObjectPool(int numbers):MdMemoryPool(numbers){};~CObjectPool() {};CObjectPool(const CObjectPool&) = delete;CObjectPool& operator=(const CObjectPool&) = delete;template <typename ...Arg>T* MfApplyObject(Arg... a); // 申请一快内存并构造对象void MfReturnObject(T* obj); // 析构对象并归还内存
};
2.10.2简单描述
CMemoryPool申请一块超大内存将其初始化成每块大小确定的内存链表，CObj是初始化链表时方便操作内存的结构(这里的代码疯狂用指针转型，一度搞蒙自己)。
CObjectPool是在CMemoryPool申请的内存块的基础上，为其调用构造和析构函数，以完成对象的建立，所以一个叫内存池一个叫对象池。
思路来自《STL源码剖析》第二章的自由链表，不过那里的自由链表是为了解决内存碎片，而这里的则是为了避开频繁的 new 和 delete 调用，将内存申请和归还操作变更为两三个指针的改变。。
2.10.3部分成员函数及代码
①首先是内存池的构造函数，这里将如果内存池需要的可分配内存块对象小于等于0就出错。
②然后是如果对象长度小于8时，就将其强制等于8字节，这里是为了兼容64位，因为64位的指针是8字节，当长度小于8时，其作为链表时就无法存储指针了。
③这里来一张图，结合里边的注释食用吧(至于代码细节还是不建议细看了，主要说明思路，毕竟我自己现在也不想看它，嗯，现在代码运行很稳定，不要随便改动，手动滑稽)：
④举个栗子：比如说一个16字节的内存池，首先MdPPool指向整块内存开始的位置(整个方框是new来的空间)，假设第一行地址为0，第二行为17，第三行34，每行代表一块可用内存，被初始化时，第一块内存的前1 ~ 9字节作为指针指向第二行地址17，第二行的前1 ~ 9字节作为指针指向第三行地址34，依此类推，整块内存变成了一条链表。
当分配一块内存出去时，地址1被返回，链表的头指针MdMemoryLinkHead被设置为指向地址18，此时返回的 地址1这个16字节被用户用来随意填充数据，当该块内存被归还时，首先把归还的地址1减一，得到地址0，查看该字节是0还是1，是0就代表是从内存池分配，然后将1 ~ 9字节设置为链表头指针指向的内存，链表头指针则被设地址1。
当内存池中无内存可分配，使用new申请16+1字节，第一个字节被设为1，表示是new申请的，然后返回new到的地址+1，归还时查看这一个1字节，是1，直接将其delete。
CMemoryPool<T>::CMemoryPool(int numbers):MdObjectLen(0), MdNumbers(numbers), MdPPool(nullptr), MdLackCount(0)
{if (0 >= MdNumbers) // 数量 没有被声明为正确的值话，就退出{printf("CObjectPool init fail!\tmsg: MDObjectNumber uninitialized\n11");ErrorWaitExit();}MdObjectLen = sizeof(T); // 确定一个对象的长度if (8 > MdObjectLen) // 兼容64位{printf("WARRING!\tmsg: MDObjectLen < 8，already set MDObjectLen is 8\n11");MdObjectLen = 8；}MdPPool = new char[(MdObjectLen + 1) * MdNumbers]{ 0 }; // 申请对象池的空间 对象长度*数量// 这里要解释一下为了多申请1字节的空间// 在调用掉该对象从对象池中返回对象池时，可能会有内存池中没有内存可分配的情况// 此时的行为应该是用new来申请空间，然后返回内存// 为了区分返回的内存是由内存池还是new分配，在分配的内存前加1字节的标志// 因为在对象池在申请时所有字节都已经设0，所以这1字节的标志为0就代表是内存池分配// new时，主动把该标志设为1，这样归还对象时，根据该标志来执行不同的操作// 接下来把申请的对象池初始化成一个链表MdObjectLinkHead = (CObj*)(MdPPool + 1);CObj* temp = MdObjectLinkHead;for (int i = 0; i < MdNumbers; ++i){temp->MdNext = (CObj*)((char*)MdObjectLinkHead + (i * (MdObjectLen + 1)));temp = (CObj*)((char*)MdObjectLinkHead + (i * (MdObjectLen + 1)));}((CObj*)(MdPPool + (MdNumbers - 1) * (MdObjectLen + 1)))->MdNext = nullptr;
}template <class T>
T* CMemoryPool<T>::MfApplyMemory()
{char* ret = nullptr;std::unique_lock<std::mutex> lk(MdMtx);if (nullptr != MdObjectLinkHead){ret = (char*)MdObjectLinkHead;MdObjectLinkHead = MdObjectLinkHead->MdNext;return (T*)ret;}else{lk.unlock();++MdLackCount;ret = new char[MdObjectLen + 1];ret[0] = 1;return (T*)(ret + 1);}return nullptr;
}template <class T>
void CMemoryPool<T>::MfReturnMemory(T* obj)
{if (nullptr == obj)return;std::unique_lock<std::mutex> lk(MdMtx);if (0 == *((char*)obj - 1)){if (nullptr == MdObjectLinkHead){((CObj*)obj)->MdNext = nullptr;MdObjectLinkHead = (CObj*)obj;}else{((CObj*)obj)->MdNext = MdObjectLinkHead->MdNext;MdObjectLinkHead = (CObj*)obj;}return;}else{lk.unlock();delete[]((char*)obj - 1);}
}
接下来就是用内存池封装的对象池了，真的是很简洁的两个函数，得意.jpg。
这里主要是placement new的使用，关于其三种不同的new和delete见More Effective C++ 35，第8条。
MfApplyObject函数申请一块内存，在其上使用placement new施以构造函数，并将其模板可变参作为构造函数的参数。
MfReturnObject则是简单为其调用析构函数，然后归还内存。
试图说明构造和析构的必要，因为构造析构函数中可能调用了new或delete，或者需要为其成员对象调用构造或者析构等等，如果不这么做可能会内存泄露或者发生其他不可预知的事情。
template <class T>
template <typename ...Arg>
T* CObjectPool<T>::MfApplyObject(Arg... arg)
{T* tmp = MdMemoryPool.MfApplyMemory();return new(tmp)T(arg...);
}template <class T>
void CObjectPool<T>::MfReturnObject(T* obj)
{obj->~T();MdMemoryPool.MfReturnMemory(obj);
}
2.11.CTime
2.11.1定义
class CTimer
{private:std::chrono::time_point<std::chrono::high_resolution_clock> MdBegin;public:CTimer(){ update(); }void update(); // 更新为当前的时间点double getElapsedSecond(); // 获取当前秒double getElapsedTimeInMilliSec(); // 获取毫秒long long getElapsedTimeInMicroSec(); // 获取微妙
};
2.11.2描述
这里是一个使用C++标志库高精度时钟的一个计时器，嗯，从某个地方搬过来的。
其主要思路就是记录下某个时间点，然后在需要时，取得当前时间点，当前时间点-记录下的时间点，就是过去的时间了，至于实现细节那就是标志库的事情了。
2.11.3成员函数代码
void CTimer::update()
{MdBegin = std::chrono::high_resolution_clock::now();
}double CTimer::getElapsedSecond()
{return (double)getElapsedTimeInMicroSec() * 0.000001;
}double CTimer::getElapsedTimeInMilliSec()
{return (double)this->getElapsedTimeInMicroSec() * 0.001;
}long long CTimer::getElapsedTimeInMicroSec()
{return std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::high_resolution_clock::now() - MdBegin).count();
}
2.12.Barrier
2.12.1定义
class Barrier
{private:int MdTarget = 0;
#ifdef WIN32std::atomic<unsigned int> MdBarrier;
#elsepthread_barrier_t MdBarrier;
#endif // WIN32
public:~Barrier();void MfInit(int target);void MfWait();
};
2.12.2描述
这里线程屏障的概念来自，第十一章第五条。
其原型来自于三个linux函数pthread_barrier_init、pthread_barrier_wait、pthread_barrier_destory。
一开始是只用C++的原子操作来模仿这个行为，但是，同样的代码在windows上运行正常，在linux里就不正常了，这个原子操作经常不能正确的等待线程，线程乱序执行，这里不用怀疑，我在这个问题上想了好几天，代码翻了无数次测了无数次。后来我不得不用宏定义的方式来区分不同的平台，在linux中使用对应的API，这样就正常。
最近几天在哪看到了大概这样一句话，有些明了了，至于具体细节我不清楚：linux的线程就是进程，只不过linux用某些方式把这些进程组织起来让他们看起来像线程的行为。可能这就是C++原子操作无法同步的原因把，因为他们本质上还是进程。
2.12.3代码
Barrier::~Barrier()
{#ifndef WIN32pthread_barrier_destroy(&MdBarrier);
#endif // WIN32
}void Barrier::MfInit(int target)
{MdTarget = target;
#ifdef WIN32MdBarrier = 0;
#elsepthread_barrier_init(&MdBarrier, nullptr, MdTarget);
#endif // WIN}void Barrier::MfWait()
{#ifdef WIN32++MdBarrier;while (MdBarrier.load() < MdTarget){std::this_thread::sleep_for(std::chrono::seconds(1));}
#elsepthread_barrier_wait(&MdBarrier);
#endif // WIN32
}
2.13.struct CNetMsgHead
2.13.1定义
struct CNetMsgHead
{int MdLen; // 该包总长度int MdCmd; // 该包执行的操作CNetMsgHead(){MdLen = sizeof(CNetMsgHead);MdCmd = -1; // 该值为-1时默认为心跳包}
};
2.13.2描述
这里就是自定义的协议头部，将消息体定义为继承该类的对象，那么消息体前就会有该消息头，方便网络收发时处理。
结束
总算是干完一遍了，写博客的过程中又发现了一些不足的地方，修修改改什么的，可以松口气了。
2020年12月9日22:24:27
从零造的一个网络轮子相关推荐
- python数据校验_最近抽空造了一个数据校验的轮子 Python -validator
最近抽空造了一个数据校验的轮子 python-validator. 在开发 web 应用时,经常需要校验前端传入的数据.如果使用 Django,那么可以使用自带的 forms 进行数据校验. pyth ...
- 这些年造过的一些轮子
这些年造过的一些轮子记录 by - theajack [2023/06/16 晚] 一晃眼,毕业从事前端也有4年时间了,如果从大三实习开始算起,那就是8年时间了.也算是比较久了,不过惭愧自认为没有什么 ...
- 真正的全栈工程师！B站硬核UP主自己造了一个激光雷达
机器之心报道 机器之心编辑部 转载自:机器之心 从零造一个激光雷达,需要多久? 激光雷达(LiDAR)是激光探测及测距系统的简称,目前广泛应用在无人驾驶和扫地机器人等领域.这种广泛的应用一方面得益于激 ...
- [Python从零到壹] 九.网络爬虫之Selenium基础技术万字详解（定位元素、常用方法、键盘鼠标操作）
欢迎大家来到"Python从零到壹",在这里我将分享约200篇Python系列文章,带大家一起去学习和玩耍,看看Python这个有趣的世界.所有文章都将结合案例.代码和作者的经验讲 ...
- [Python从零到壹] 五.网络爬虫之BeautifulSoup基础语法万字详解
欢迎大家来到"Python从零到壹",在这里我将分享约200篇Python系列文章,带大家一起去学习和玩耍,看看Python这个有趣的世界.所有文章都将结合案例.代码和作者的经验讲 ...
- Linux就这个范儿 第12章 一个网络一个世界
Linux就这个范儿 第12章 一个网络一个世界 与Linux有缘相识还得从一项开发任务说起.十八年前,我在Nucleus OS上开发无线网桥AP,需要加入STP生成树协议(SpanningTree ...
- [Python从零到壹] 四.网络爬虫之入门基础及正则表达式抓取博客案例
欢迎大家来到"Python从零到壹",在这里我将分享约200篇Python系列文章,带大家一起去学习和玩耍,看看Python这个有趣的世界.所有文章都将结合案例.代码和作者的经验讲 ...
- 巴法云 使用 esp01s wifi模块 做一个网络智能开关
巴法云 使用 esp01s wifi模块 做一个网络智能开关 前言 一:配置arduino IDE,使其支持ESP8266编程开发 二:esp01s接线问题 1 下载模式 2 运行模式 三:开发环境测 ...
- c#使用正则表达式获取TR中的多个TD_[Python从零到壹] 四.网络爬虫之入门基础及正则表达式抓取博客案例...
首先祝大家中秋节和国庆节快乐,欢迎大家来到"Python从零到壹",在这里我将分享约200篇Python系列文章,带大家一起去学习和玩耍,看看Python这个有趣的世界.所有文章都 ...
最新文章
- 扎心！互联网行业的神回复
- FastGithub让Github畅通无阻
- 二分二题-P1678,P3902【二分,LIS】
- javascript学习系列(2):数组中的filter方法
- 大数据量分页存储过程效率测试附代码
- F. 张胖胖玩多米诺骨牌 （南阳理工oj—21新生第二场招新赛）
- 思科交换机配置试题_【干货】思科交换机路由器怎么配置密码？
- scanf和getch函数的区别
- iShot--Mac最强截图工具
- .net reactor使用教程(4)——.net reactor针对De4Dot脱壳工具的应对
- 鸡小德手机小常识 如何鉴定手机是否为行货
- Dagger2实际应用篇
- MySQL并列排名，不占位（并中文排序）
- 谷歌浏览器打开调试模式后刷新页面，页面一直加载问题
- 制作kali live usb启动盘-win10下
- 基于XDOC云服务的标准公文模板【信函】
- 关于Nvidia卡在注册界面的问题
- UG二次开发GRIP给尺寸增加*
- Linux 命令大全 (新手必备收藏)
- 2.Open3D教程——文件读取和保存
热门文章
- 金字塔的思维-结构化逻辑思维
- 错误：You have an error in your SQL syntax； check the manual that corresponds to your MySQL server vers
- 为了在上海租房，我用python连夜爬了20000多条房源信息
- 外贸企业邮箱登录入口，怎么登陆邮箱app？
- (转)超详细iOS证书申请流程
- 玩过这些经典单机游戏_就说明你已经老了
- 玩过这些经典单机游戏，就说明你已经老了
- UNIAPP day_01(8.30) uin-app概述
- shell stop|start|restart
- 包子笔记 - 投资的真谛是钱生钱的不断复利前进