首页 > 代码库 > 实现UDP高效接收/响应
实现UDP高效接收/响应
环境Linux g++6.3.0
问题一：一个ip地址如何接收高并发请求
问题二：如何高并发响应消息
发送请求端只能通过ip地址+端口号向服务器发送请求码，所以服务器只能用一个UDP去绑定此ip以及端口号。而如何完成高并发发送响应消息，
谁去发送这个响应消息，接收请求信息的UDP？这就造成其中一个任务必须等待另一个任务执行完毕，sendto是非阻塞，而recvfrom是阻塞，若
执行recvfrom碰巧没有下个请求信息或者网络阻塞造成UDP丢失，那么sendto岂是不能执行(一直等待recvfrom结束)，这时就会想用超时解决，或
者也可以开启线程来执行这个sendto函数。对于线程，先不考虑消息传递共享问题(IPC)，假若有多个响应需要同一时间一起发送，那么此特定
UDP需要与哪个匹配，它可以同时发送这些任务？不可能吧，所以需要多个UDP来发送这些响应。现在可以接收以及并发响应。
问题三：线程只能执行函数一次，则结束。所以每次都得重新构建新线程，如何避免
问题四：当线程构建速率大于线程接收速率时，如何处理
把线程扔进队列中，进行统一管理，然后如何取任务？这个问题的解决办法与问题四解决办法相同，也就是在线程池内创建一个队列用来存放
任务，而后线程从任务队列中取出任务执行。问题四的解决是可以限制任务队列的容量，当满的时候进行等待。
问题五：线程池对外开放接口，参数应该是哪种类型
问题六：接收请求信息UDP和发送响应信息UDP有何共同点，如何创建这两个类
希望被执行的对象是函数类/函数指针，所以线程池对外的接口参数也就是函数类/函数指针及其参数，而决定这些特性取决发送响应信息UDP类，
而把发送消息信息UDP类添加进线程池中的类是接收请求信息UDP类(响应依赖请求)，接收请求信息UDP类与发送响应信息UDP类共同点是对
socket构建，所以先创建一个基类UDP类
问题七：发送响应信息的载体UDP socket如何分配，发送响应信息类所需要的信息从何处获取
首先意识到决定这一切，也是对线程池添加任务速率的瓶颈，影响并发响应速率的源头：接收请求信息UDP。所以在此类中添加一个UDP socket
队列，而发送响应信息类所需信息有sockaddr_in(ipv4)、响应内容以及一个发送载体UDP socket，因此在UDP中添加这些属性
问题八：如何添加任务进线程池
怎么办？C++11 std::function<void(P)> 然后呢？参数有多个，所以
template <typename...Args> class ThreadPool { . . . void AddTask(std::function<void(Args...)>); }
先不说如何实现在线程池中存放可变参数，这样后此线程池需要被特例化且智能被一种参数所用。可能有点表达不明白，既然是存放，那么我们
的目的是为了取出，所以如何取？必须提前知道所取的类型using Task= std::function<void(Args...)>; 而这必须在构建时就确定好。
如果能把参数类型擦除(去）掉多好，那么就成了有多个无类型的参数，可在存放参数数量又是一个麻烦，要是只有一个那么就好了。所以我们要
把多个属性包装在一个类中再把类型擦除掉，perfect！
实现过程：
//基础udp类，创建了初始化sock，关闭sock，获取sock方法 //其一，AdvanceUdp类继承此类 //其二，在并发发送请求消息时需要多个udp发送消息，需要 //一个循环队列，而循环队列的成员就是此类 class CommonUdp { protected: int sock; //sock public: //初始化，AF_INET ipv4族，SOCK_DGRAM 数据流 //IPPROTO_UDP UDP协议 void Init() { sock= socket(AF_INET,SOCK_DGRAM,IPPROTO_UDP); if (sock< 0) { throw std::out_of_range("socket() failed\n"); } } //关闭sock void Close() const { close(sock); } //获取sock int GetSock() const noexcept{return sock;} //析构 调用Close virtual ~CommonUdp() { #if defined(Csz_test) std::cout<<"CommonUdp::destructor\n"; #endif Close(); } };
UDP载体队列，我采用环形链表，添加任务时从表中取出sock，接着链表指向下一个，不断循环使用。没使用Container其一是觉得没有必要，
其二若要进行循环使用则还需要一个标记位来记录上一次的位置。而自构环形链表类可以方便使用，省空间。
1 //Upd环形列表 2 //禁止了拷贝构造/移动构造/拷贝赋值/移动赋值为了避免内存泄露 3 class UdpAnnulus 4 { 5 public: 6 CommonUdp udp; 7 UdpAnnulus *next; 8 UdpAnnulus() noexcept 9 { 10 #if defined(Csz_test) 11 std::cout<<"UdpAnnulus constructor\n"; 12 #endif 13 } 14 ~UdpAnnulus() noexcept 15 { 16 #if defined(Csz_test) 17 std::cout<<"UdpAnnulus destructor\n"; 18 #endif 19 } 20 UdpAnnulus(const UdpAnnulus &)=delete; 21 UdpAnnulus(UdpAnnulus &&)=delete; 22 UdpAnnulus& operator=(const UdpAnnulus &)=delete; 23 UdpAnnulus& operator=(UdpAnnulus &&)=delete; 24 };
接着是发送响应消息的方法类所需要的参数类型 ：
data是消息存储内存起始地址，data[3] 数组类型，后面会看见其实请求消息只有一字节，也就是八位。前三位用来表示8个请求类型，后五位用来表
示请求哪个位置信息。sock 发送响应消息载体。client_addr存放ipv4地址信息，包括ip地址、端口号、协议类型以及套接字类型等。sentry顾名思义
是一个哨兵，用来映射请求消息后五位所在具体位置。
现在来详细分析为什么请求码明明只有一个字节却需要用到三个char。理由其一，若是恶意信息则只需检查0== data[1] 是否为false，true则是正常用
户请求，false则忽略且记录mac地址，当被记录次数达到一定数量可以delete。理由其二，recvfrom参数中表示接收容量的参数必须比实际少一位，
用来填‘\0‘，所以如果用2个char则每次0== data[1]都为true，也就起不到筛选效果。
1 //广播方法类的参数，保存在Any类中 2 //data 接收请求码 3 //sock 发送UDP sock 4 //sentry 响应信息在文件中的位置标记(ifstream.seekg(sentry)) 5 //client_addr 接收响应消息的对象 6 class Message 7 { 8 public: 9 char data[3]; 10 int sock; 11 uint32_t sentry; 12 sockaddr_in client_addr; 13 #if defined(Csz_test) 14 ~Message(){std::cout<<"Message::destructor\n";} 15 #endif 16 };
下面是接收请求消息的UDP类，删除了部分功能实现代码
因为继承了CommonUdp所以只需进行bind()即可使用。设置了两个bool来记录结束以及保证Free()函数只执行一次，当!end_flag为true则ServiceUdpGo
继续循环执行，而once_flag在执行了Free()函数后被置为true，调用析构函数时检查once_flag是否为true，若不是则调用Free()，Free包含了关闭接收请
求消息的载体socket以及销毁释放环形UDP链表。guard是环形Udp当前位置，每当获取到当前sock后，guard自动指向自己的下一个。Mess是发送响应
消息Udp方法类的参数，在ServiceUdpGo函数中配置好信息，接着加入线程池任务队列中，等待被执行。RecordPtr以key-value形式映射请求消息后五
位与响应消息所在具体位置的关系，0~2^5对应每条Json头位置。
InitJsonFileRecord()用来初始化RecordPtr对相应文件进行检索找到对应位置。
CreateUdpAnnulus()创建环形UDP链表。有两种办法，其一不断把新的节点插在头节点的下一个，而原本头节点的下一个变成新节点的下一个，也就是
说头节点以及最后一个节点保持着不变，在最后把最后一个节点的下一个节点指向头节点。其二每次把新节点放到尾，最后用尾的下一个指向头节点。
注意的一点是，节点里面的数据也就是UDP socket 在构建好节点后需要初始化，而初始化可能会扔出异常，这时候就需要一个异常处理机制，避免程序
崩掉而环形链表所申请的空间未释放，造成内存泄露。
DestroyUdpAnnulus()销毁环形UDP链表，我的做法是先把环形拆掉，变成单向链表，然后进行内存资源回收。
ExceptionUdpAnnulus()创建环形UDP链表时，处理因初始化UDP socket失败扔出的异常。
GetBroadCastSock() 获取环形UDP链表当前socket，并将哨兵移动到下一个。
Check()检查data[1]是否等于0，也就是说请求消息是否合法
RetNum() 通过映射将data[0]后五位转换成响应消息真实所在文件位置。
AdvanceUdp()构造，进行绑定ip地址、端口号，创建环形链表，初始化RecordPtr。
～AdvanceUdp()析构，若在实例变量释放时还未调用Free()进行资源回收，则析构会自主调用，若已调用则不做任何something
Free() 关闭socket，销毁环形UDP链表
Bind() 将socket与ip及端口进行绑定
End() 结束ServiceUdpGo()函数循环
ServiceUdpGo() 这里有一个值得注意的地方，当调用End() 而recvfrom处于阻塞状态，那么就不能立即结束函数，必须等待recvfrom错误或者下一个请求
消息到来。 select多路复用，挂起等待I/O发送信号量唤醒，若有消息到来则会将到来描述符置为，也就是说每次都要重新设置所要监视的描述符。超时
时间也需要每次重置，否则下一次就是0,0也就是说轮询效果。
//继承了普通Udp类，有个sock，以及初始化/关闭/获取sock方法 //通过AdvanceUdp 实质是一个udp反复接收其他udp发送过来的请求信息 //然后进行分类处理并派送给其他udp去进行发送 //end_flag 结束标记，线程不断跑接收请求函数，是否结束则跟着次标记 //once_flag 保证关闭socket和释放环形链表只执行一次 //mess 将任务扔进线程池需要方法类以及其参数，而mess就是其参数 //RecordPtr 记录了json文件每条记录在文件中的位置，因为UDP是无状态 //所以需要认为记录‘状态‘ //调用End() 去设置end_flag为true，而ServiceUdpGo() 是while (!end_flag) //调用 recvfrom 阻塞/等待,也就是说在没接收到任何信息时，是不会结束此线程 //循环。解决办法 1.设置超时 2.设置锁 condition_variable class AdvanceUdp : public CommonUdp { private: using RecordPtr= std::shared_ptr<std::vector<uint32_t>>; private: bool end_flag; //结束标记 bool once_flag; //运行一次标记 UdpAnnulus *guard; //udp环形链表 哨兵 Message mess; //response消息 RecordPtr mv_record; //智能指针shared_ptr RecordPtr os_record; RecordPtr sw_record; RecordPtr tv_record;private: //初始化 RecordPtr void InitJsonFileRecord()；//创建环形链表 //有两种方法 一种是不断的把新指针插在头的下一个(注释)，另一种是按顺下申请下去 //每申请一个变量就为成员变量udp调用初始化函数 void CreateUdpAnnulus(uint8_t &&T_num) { #if defined(Csz_test) std::cout<<"AdvanceUdp::CreateUdpAnnulus()\n"; #endif guard= new UdpAnnulus(); guard->next= nullptr; //异常 try { guard->udp.Init(); } catch (const std::out_of_range &T_mess) { std::cerr<<T_mess.what(); delete guard; exit(1); } //UdpAnnulus *flag; UdpAnnulus *temp= guard; while (T_num> 1) { //UdpAnnulus *temp= new UdpAnnulus(); //temp->next= guard->next; //guard->next= temp; temp->next= new UdpAnnulus(); temp= temp->next; try { //guard->next->udp.Init(); temp->udp.Init(); } catch (const std::out_of_range &T_mess) { temp->next= nullptr; std::cerr<<T_mess.what(); ExceptionUdpAnnulus(); exit(1); } //if (nullptr== temp->next) // flag= temp; --T_num; } //flag->next= guard; temp->next= guard; //尾的下一个是头 } //销毁环形链表 //先把头的下一个赋值给临时变量，再把头下一个置空 //这时候就是个单向链表了，从临时变量开始进行删除 void DestroyUdpAnnulus() { #if defined(Csz_test) std::cout<<"AdvanceUdp::DestroyUdpAnnulus()\n"; #endif if (nullptr== guard) return ; UdpAnnulus *temp= guard->next; guard->next= nullptr; while (nullptr!= temp) { UdpAnnulus *dele= temp; temp= temp->next; delete dele; } guard= nullptr; } //环形链表初始化udp时抛出异常处理机制 //在初始化环形链表时头是不变的 void ExceptionUdpAnnulus() { #if defined(Csz_test) std::cout<<"AdvanceUdp::ExceptionUdpAnnulus()\n"; #endif while (nullptr!=guard) { UdpAnnulus *dele= guard; guard= guard->next; delete dele; } } //获取环形链表中udp socket //然后guard移动到下一个 int GetBroadCastSock() { int value= http://www.mamicode.com/guard->udp.GetSock(); guard= guard->next; #if defined(Csz_test) std::cout<<"AdvanceUdp::GetBroadCastSock() sock:"<<value<<"\n"; #endif return value; } //判断是否是合法用户 //第二个字符是否为空‘\0‘ bool Check() {if (mess.data[1]== 0) return true; else return false; } //返回所请求页数所在的位置(文件中的位置) //×××1 1111 后五位表示页数 2^5=64页 //111× ×××× 前三位为类型 2^3=8种 //000× :0 001× :32 010× :64 011× 96 //100× :128 101× :160 110× :192 111× :224 //32:MV 64:TV 96:SW 128:OS uint32_t RetNum(char &T_seek) { #if defined(Csz_test) std::cout<<"AdvanceUdp::RetNum()\n"; #endif switch (SelectType()(T_seek)) { case 32: if ((T_seek& 0x1f)>= mv_record->size()) return -1; return (*mv_record)[T_seek& 0x1f]; case 64: if ((T_seek& 0x1f)>= tv_record->size()) return -1; return (*tv_record)[T_seek& 0x1f]; case 96: if ((T_seek& 0x1f)>= sw_record->size()) return -1; return (*sw_record)[T_seek& 0x1f]; case 128: if ((T_seek& 0x1f)>= os_record->size()) return -1; return (*os_record)[T_seek& 0x1f]; default: //0,224,160,192 不合法请求码 return -2; } } public: //构造函数，参数为udp环形链表的个数 //Bind方法 扔出异常std::out_of_range //对T_num进行检测，不能是负数 AdvanceUdp(uint16_t &&T_port,uint8_t &&T_num= UdpNum) : end_flag(false),once_flag(false),guard(nullptr) { try { Bind(std::forward<uint16_t>(T_port)); } catch (const std::out_of_range &T_mess) { std::cerr<<T_mess.what(); exit(1); } //T_num= T_num> 0? :T_num : 1; CreateUdpAnnulus(std::forward<uint8_t>(T_num)); InitJsonFileRecord(); } //析构函数，若已经手动释放则不做任何操作 ~AdvanceUdp() { #if defined(Csz_test) std::cout<<"AdvanceUdp::destructor\n"; #endif if (!once_flag) { Free(); } } AdvanceUdp(const AdvanceUdp &)=delete; AdvanceUdp(AdvanceUdp &&)=delete; AdvanceUdp& operator=(const AdvanceUdp &)=delete; AdvanceUdp& operator=(AdvanceUdp &&)=delete; //关闭udp socket(用于接收消息),释放环形链表 //设置操作标记符为true void Free()；//将sock与地址/端口进行绑定 //调用基类进行sock初始化 //struct sockaddr_in ipv4地址结构体 void Bind(uint16_t &&T_port)；//停止ServiceUdpGo void End()；//线程调用函数 //参数为线程池，用来存放请求任务 //recvfrom 接收请求消息,消息大小需要比实际大小少1，因为要用来存放最后一个null //因为udp是有界，所以在接收消息时，当接收消息大小< 缓冲大小时，多余的数据将被 //悄悄的丢弃，利用这个我们调用Check()检查mess.data[1]是否为空，若不时则是非法用户 void ServiceUdpGo(std::shared_ptr<Csz_ThreadPool::ThreadPool> T_pool) { //多路复用 fd_set sock_set; //创建描述符 int max_desc= sock+ 1; //设置最大描述符 int code; //记录select返回值 -1为错误，0为超时 #if defined(Csz_test) std::cout<<"AdvanceUdp::ServiceUdpGo() start\n"; #endif while (!end_flag) { FD_ZERO(&sock_set); //清零 FD_SET(sock,&sock_set); //设置描述符 //时间结构必须每次都重置，否则下次清零 struct timeval time_out; time_out.tv_sec= TIMEOUT; //秒 time_out.tv_usec= 0; //微秒 memset(&mess,0,sizeof(mess)); socklen_t addr_len= sizeof(mess.client_addr); if ((code= select(max_desc,&sock_set,NULL,NULL,&time_out))<= 0) { if (-1== code) { std::cerr<<"AdvanceUdp::ServiceUdpGo() select return -1\n"; exit(0); } #if defined(Csz_test) //std::cout<<"AdvanceUdp::ServiceUdpGo() time out\n"; #endif //continue; //超时后若直接进行下一次操作，recvfrom则不会运行， //下次select也是等待超时，而没收到上次信号量 } if (FD_ISSET(sock,&sock_set)) { if (recvfrom(sock,mess.data,sizeof(mess.data)-1,0,(struct sockaddr*)&mess.client_addr,&addr_len)< 0) { #if !defined(Csz_O1) std::cerr<<"recvfrom< 0\n"; #endif continue; } } else { continue; } if (Check()) { mess.sentry= RetNum(mess.data[0]); mess.sock= GetBroadCastSock(); T_pool->AddTask(UdpBroadCast(),std::move(mess)); } else { //AOP //记录错误请求ip //std::cerr<<"error request from ip" #if !defined(Csz_O1) char client_addr[16]={0}; inet_ntop(AF_INET,&mess.client_addr.sin_addr.s_addr,client_addr,sizeof(client_addr)); std::cerr<<"error request from ip "<<client_addr<<" port "<<ntohs(mess.client_addr.sin_port)<<"\n"; #endif } } #if defined(Csz_test) std::cout<<"AdvanceUdp::ServiceUdpGo() stop\n"; #endif } };
c++17添加了一个类Any
An object of class any stores an instance of any type that satisfies the constructor requirements or is empty, and this is referred to as the state of the
class any object. The stored instance is called the contained object. Two states are equivalent if they are either both empty or if both are not empty and
if the contained objects are equivalent.The class any describes a type-safe container for single values of any type.
而这里要用到的是个类似Any类的类，它能擦除掉任何类型并把值保存在内部，在需要的时候可以取出来。其实就是把任何想保存的类型变成一个统一
的类型，用模板来提取所要保存类型的类型信息(std::type_index)，而Value(值)则用内部类(其实跟定义变量一样，只不过这个变量是我们自己构建)的
派生类来记录下，为什么不能直接写在Any类中呢？在特例化的时候有不同属性的Any的std::type_index是不等价，而如果直接用内部类来记录值，则
内部类特例化的时候有不同属性的内部类的 std::type_index是不等价，从而导致两个Any的std::type_index不等价。说到底就是std::type_index要等价，
除了类名相同，属性名也得相同。所以在Any类中定义一个指向内部类的unique_ptr指针，而指针实际是指向内部类的派生类。这有点类似桥连接，只
知道Any类有个智能指针，而不在意这个指针真实是指向哪里。
//擦除掉类型，利用在构造函数时，传进所要擦除的类型以及值，用指针 //进行记录值，用type_index记录类型，而在需要返回类型的值时，用模板 //继续传如类型，与type_index比较若相等则进行强制类型转换,dynamic_cast struct Any { private: struct Base; //接口 using BasePtr= std::unique_ptr<Base>; //智能指针保证资源释放 std::type_index index; //记录当前保存的类型 BasePtr ptr; //记录当前类型的智能指针，get()返回指向的类型 private: struct Base { //用虚函数保证继承关系的派生类能够调用析构函数 //在实现多态时，当用基类操作派生类，在析构时防止只析构基类而不析构派生类 virtual ~Base(){} //必须virtual virtual BasePtr Clone()const =0; //必须virtual }; //保存值的类 //模板特例化传进的是保持值的类型，而构造函数模板则是保证值传递进来是原始类型值 //这样做的好处是可能是右值，也能剩下构造值的时间 template <typename T> class Derived : public Base { public: T value; public: template <typename U> Derived(U &&T_value) : value(std::forward<U>(T_value)) { #if defined(Csz_test) std::cout<<"Any::Derived::constructor\n"; #endif } #if defined(Csz_test) ~Derived(){std::cout<<"Any::Derived::destructor\n";} #endif BasePtr Clone()const { #if defined(Csz_test) std::cout<<"Any::Derived::Clone()\n"; #endif return BasePtr(new Derived<T>(value)); } }; //辅助方法,调用Derived::Clone()变简单 BasePtr Clone() const { #if defined(Csz_test) std::cout<<"Any::Close()\n"; #endif if (nullptr!= ptr) return ptr->Clone(); return nullptr; } public: //默认构造 Any() : index(typeid(void)),ptr(nullptr){} //右值构造 Any(Any &&T_other) : index(std::move(T_other.index)),ptr(std::move(T_other.ptr)){} //左值构造 Any(const Any &T_other) : index(T_other.index),ptr(T_other.Clone()){} //值构造 //用模板，限定模板类型为非Any类 //std::decay 先移除T引用得到U，U为 remove_reference<T>::type //若是is_array<U>::value 为true,则修改类型type为 remove_extent<U>::type* 移除数组的一个维度，若是一维数组则是变量指针 //若是is_function<U>::value为true,则修改类型type为 add_pointer<U>::type 也就是函数指针 //剩下的修改type类型为 remove_cv<U>::type template <typename T,typename= typename std::enable_if<!std::is_same<typename std::decay<T>::type,Any>::value,T>::type> Any(T &&T_value) : index(typeid(typename std::decay<T>::type)),ptr(new Derived<typename std::decay<T>::type>(std::forward<T>(T_value))){} #if defined(Csz_test) ~Any(){std::cout<<"Any::destructor\n";} #endif bool IsNull() const noexcept { #if defined(Csz_test) std::cout<<"Any::IsNull()\n"; #endif return nullptr== ptr; } //查询当前保存的类型是否为函数特例化的类型 template <typename T> bool Is() const noexcept { #if defined(Csz_test) std::cout<<"Any::Is()\n"; #endif return std::type_index(typeid(T))== index; } template <typename T> T AnyCast() { #if defined(Csz_test) std::cout<<"Any::AnyCast()\n"; #endif if (!Is<T>()) { #if !defined(Csz_O1) std::cerr<<"can not cast "<<index.name()<<" to "<<typeid(T).name()<<"\n"; #endif throw std::bad_cast(); } //std::unique_ptr<T>.get() 返回智能指针指向T的指针,若没有则返回nullptr //std::unique<Base> ptr auto derived= dynamic_cast<Derived<T>*>(ptr.get()); //若是值为类则必须在类里面写好移动构造函数，避免了资源释放两次 //若值是标准类型，其实也是无所谓但养成好习惯，标准类型的move and copy效果是一样的 return std::move(derived->value); } //copy赋值操作符 Any& operator=(const Any &T_other) { #if defined(Csz_test) std::cout<<"Any::operator= copy()\n"; #endif if (T_other.ptr== ptr) return *this; ptr= T_other.Clone(); index= T_other.index; return *this; } //move赋值操作符 Any& operator=(Any &&T_other) { #if defined(Csz_test) std::cout<<"Any::operator= move\n"; #endif if (T_other.ptr== ptr) return *this; ptr=std::move(T_other.ptr); index= std::move(T_other.index); return *this; } };
Any类是在哪里使用呢？除了是作为传入线程池的参数还有就是作为另一个关键方法类的参数，那就是发送响应消息UDP类。
在方法类中指定从Any类中取出所需要的类型，若Any中并没有保存这个类型则会扔出异常。当类型取出后，对消息字节的前三位做 相与 得出type。
为什么不是函数，而是方法类呢？在调用的时候函数一定是通过指针去调用，而方法类可能是内联代码。
//广播方法类 //参数为Any，在操作时进行转换 //对消息进行类型分类 struct UdpBroadCast : public std::unary_function<Any,void> { void operator()(Any &&T_any)const noexcept { //broadcast //Any::AnyCast throw std::bad_cast Message value; try { value= T_any.AnyCast<Message>(); } catch (std::bad_cast &T_mess) { #if !defined(Csz_O1) std::cerr<<"UdpBroadCast::operator(): "<<T_mess.what()<<"\n"; #endif return ; } #if defined(Csz_test) std::cout<<"UdpBroadCast::operator() request:"<<short(value.data[0])<<"\n"; #endif switch (SelectType()(value.data[0])) { case 32: SendTo(JSON_MV,value.sentry,value.sock,value.client_addr); break; case 64: SendTo(JSON_TV,value.sentry,value.sock,value.client_addr); break; case 96: SendTo(JSON_SW,value.sentry,value.sock,value.client_addr); break; case 128: SendTo(JSON_OS,value.sentry,value.sock,value.client_addr); break; default: //0,224,160,192 return ; } } //实际广播方法 //若T_sentry== -1 uint32_t -1对应最大值(请求页数大于最大值) //若T_sentry== -2 不合法请求码 //则不做任何处理，请求页数超过了数据库所拥有的页数 //文件打不开可能是文件名不对，或者句柄数量已到上限 //std::string 也是不断动态申请空间，当当前空间不足时 //将再申请一块更大的空间(通常是以前空间×2)，再释放 //之前的空间，所以为了避免返回整合空间一开始就将空间 //设置位较大，使用temp.reserve(); //若后面JSON是动态改变则可以在过后使用temp.shrink_to_fit()释放多余空间 void SendTo(const char * T_file_name,const uint32_t &T_sentry,const int &T_sock,const sockaddr_in &T_client_addr) const noexcept { if (uint32_t(-2)== T_sentry) { #if !defined(Csz_O1) char client_addr[16]={0}; inet_ntop(AF_INET,&T_client_addr.sin_addr.s_addr,client_addr,sizeof(client_addr)); std::cerr<<"ip "<<client_addr<<" port "<<ntohs(T_client_addr.sin_port)<<" RequestCode out of range\n"; #endif return ; } std::string temp; temp.reserve(32); std::ifstream handle(T_file_name); if (!handle.is_open()) { #if !defined(Csz_O1) std::cerr<<"file_name: "<<T_file_name<<" open failed\n"; #endif temp.assign("服务器数据错误"); } else if (uint32_t(-1)== T_sentry) { #if !defined(Csz_O1) char client_addr[16]={0}; inet_ntop(AF_INET,&T_client_addr.sin_addr.s_addr,client_addr,sizeof(client_addr)); std::cerr<<"ip "<<client_addr<<" port "<<ntohs(T_client_addr.sin_port)<<" request page> Max\n"; #endif temp.assign("请求页码大于最大值"); } else { temp.reserve(1152); handle.seekg(T_sentry); std::getline(handle,temp); } //sendto 失败返回-1 if (sendto(T_sock,temp.c_str(),temp.size(),0,(struct sockaddr*)&T_client_addr,sizeof(T_client_addr))< 0) { #if !defined(Csz_O1) char client_addr[16]={0}; inet_ntop(AF_INET,&T_client_addr.sin_addr.s_addr,client_addr,sizeof(client_addr)); std::cerr<<T_sock<<": sendto "<<client_addr<<" port "<<ntohs(T_client_addr.sin_port)<<" failed\n"; #endif //记录 } } };
现在只差线程池了，而线程池中关键的还是任务的添与取。任务队列采用双向链表而线程队列采用单向链表，因为任务队列有分先后，而线程队列不关心
先后问题且单向链表比双向链表在空间利用上更为有效。
stop_flag 停止任务添加与取出标记。size_max用来线程链表最大容量，避免内存爆炸。is_full与is_empty是多线程之间的通信变量，可以使一些线程等待
其他线程的通知，一般总是绑定一个std::unique_lock。task_list 任务队列，存放一个std::pair<std::function,Any> ，std::function用来存放方法类或着函数，
而Any则是这些方法类或者函数的参数，std::function用于擦除方法类、函数以及lambda的区别，而Any用于擦除参数类型的不同，使整个线程池耦合性降低。
Add()是个模板函数，为std::forward的使用提供环境(引用未定义)，使std::forward正确推出左值和右值，当是右值时可以节省内存的开销。作用是将任务
添加经链表中。这里对停止做了一个处理，是为了不造成任务数据丢失，为线程池暂停执行与重新开始提供了条件。
Put()重载了两个，左值参数与右值参数，共同点都是调用了Add()。使用std::forward保持参数类型与传入参数类型是一致的，比如右值参数，传进Add()也
一样是一个右值参数，若没使用则是依左值参数传入。
Take() 取出任务，这里并不是将取到的变量返回，而是通过参数传入进行取值，很明显的一个优势就是不必构建一个变量进行取值返回。
Stop() 停止任务添加与取出，这里需要注意的是在获取锁以及改变stop_flag标记位是在局部范围内完成，若没有的话，在唤醒其他线程时，由于锁还未
释放，所以其他线程则需等待获取锁。
IsFull()检查是否为满，为条件锁提供判断。
IsEmpty()检查是否为空，为条件锁提供判断。这两个函数都不可以添加锁，因为在Add()或者Take()都已经获取了mutex，而现在若是要再获取mutex则发生
死锁。
Size() 获得任务链表当前大小。
Clear() 清除任务链表。
#define SIZE 30 //默认30个容量任务 //#define Csz_test //Func 方法类 template<typename Func> class TaskList { private: bool stop_flag; //标记是否停止 uint8_t size_max; //最大任务数 std::mutex task_mutex; //锁 std::condition_variable is_full; //为满时等待 std::condition_variable is_empty; //为空时等待 std::list<std::pair<Func,Any>> task_list; //以Any类擦除了函数参数的不同 private: //这里用到引用未定义，若传进来是右值则为右值引用，其他则为左值 //const Task && T_task 加了const 则不是引用未定义，只是个右值引用参数 template <typename Task,typename Parameter> void Add(Task && T_task,Parameter &&T_any) { std::unique_lock<std::mutex> guard(task_mutex); //上锁 //若不能执行则释放锁进入等待，加了第二参数相当 //while (!predicate()) // is_full.wait(); is_full.wait(guard,[this]{return stop_flag || !IsFull();}); if (true== stop_flag) { //任务是被停止了而不是结束，有可能继续，为了保证任务不丢失 //再次检查是否为满，不为满则加入任务 if (!IsFull()) task_list.emplace_back(std::forward<Task>(T_task),std::forward<Parameter>(T_any)); return ; } #if defined(Csz_test) std::cout<<"TaskList::Add::emplace_back\n"; #endif //就地构造，避免了一次没必要的移动拷贝或者赋值拷贝 task_list.emplace_back(std::forward<Task>(T_task),std::forward<Parameter>(T_any)); is_empty.notify_one(); //此时不为空，所以唤起因为空而等待的线程 } public: TaskList(const uint8_t &T_size= SIZE) noexcept : stop_flag(false),size_max(T_size){} ~TaskList()noexcept { #if defined(Csz_test) std::cout<<"TaskList::destructor\n"; #endif task_list.clear(); } TaskList(const TaskList &T_other)=delete; TaskList(TaskList &&T_other)=delete; TaskList& operator=(const TaskList &T_other)=delete; TaskList& operator=(TaskList &&T_other)=delete; //任务添加为左值 void Put(const Func &T_task,const Any &T_any) { #if defined(Csz_test) std::cout<<"TaskList::Put()copy\n"; #endif Add(T_task,T_any); } //任务添加为右值 //利用完美转发保持参数以右值形式传进 void Put(Func &&T_task,Any &&T_any) { #if defined(Csz_test) std::cout<<"TaskList::Put()move\n"; #endif Add(std::forward<Func>(T_task),std::forward<Any>(T_any)); //forward<int>(value) 为右值 } //不返回值，利用参数加引用，假设在没有返回值优化的情况下，那么将比有返回值 //的少了一个constructor以及一个copy void Take(Func &T_task,Any &T_any) { std::unique_lock<std::mutex> guard(task_mutex); //若不能执行则释放锁进入等待，加了第二个参数相当于 //while(!predicate()) // is_empty.wait(); is_empty.wait(guard,[this]{return stop_flag || !IsEmpty();}); if (true== stop_flag) { //T_task=[]{}; //先判断后运行 return ; } #if defined(Csz_test) std::cout<<"TaskList::Take()\n"; #endif auto task= std::move(task_list.front()); task_list.pop_front(); T_task= std::move(task.first); T_any= std::move(task.second); is_full.notify_one(); //此时不为满，所以唤醒因为满而进行等待的线程 } void Stop() noexcept { #if defined(Csz_test) std::cout<<"TaskList::Stop()\n"; #endif { //设为局部锁，若在加锁的情况下唤醒其他等待线程， //则其他线程必须等待mutex的释放 std::unique_lock<std::mutex> guard(task_mutex); stop_flag= true; } is_full.notify_all(); is_empty.notify_all(); }private: //在Add函数中调用到，而Add函数本身是在有锁的情况下进行 //若在此函数再进行加锁着就会出现饿死的情况也就是死锁 bool IsFull() const noexcept { #if defined(Csz_test) std::cout<<"TaskList::IsFull()\n"; #endif return task_list.size()== size_max; } //同上 在take bool IsEmpty() const noexcept { #if defined(Csz_test) std::cout<<"TaskList::IsEmpty()\n"; #endif return task_list.empty(); } public: //若是在外调用，保证返回值是当前的准确值 //加锁的情况下 size_t Size() const noexcept { #if defined(Csz_test) std::cout<<"TaskList::Size()\n"; #endif std::lock_guard<std::mutex> guard(task_mutex); return task_list.size(); } void Clear() { #if defined(Csz_test) std::cout<<"TaskList::Clear()\n"; #endif std::lock_guard<std::mutex> guard(task_mutex); task_list.clear(); } };
线程池，可以被多个线程操作(添加任务，结束线程)，暂停与重新开始(不建议多线程调用，因为可能会发生死锁或者抛出异常)。
stop_flag 原子操作变量，用于标记线程池暂停状态。
stop_mutex 协助线程池暂停线程运行，调用Stop()时对stop_mutex进行锁定，而线程在检查到stop_flag为true时对stop_mutex锁进行获取，着就造成线程
停止等待。其实这个暂停与重新开始功能很鸡肋，因为是直接对mutex进行操作，而不是用std::unique_lock或者std::lock_guard进行对锁的管理(也不能
使用这两个)，这就出现了问题，如果多个线程同时对线程池进行暂停或者重新开始，先说暂停把，那么就会造成调用者不断等待获取锁，从而饿死。而对
一个没有锁的锁进行unlock则会抛出异常(行为未定义)。所以若要使用这两个函数则必须只能是单线程操作，在析构时检查stop_flag是否为true，若是则
调用Restart()对 stop_mutex 进行解锁，避免线程池中的线程还处于获取stop_mutex锁的状态从而发生了死锁。
promise_one c++11新特性，保证收回资源函数只被运行一次，为多线程同时操作单个线程池提供了安全保障。
end_flag 原子操作变量，用于标记线程池是否结束。
thread_list 是单向链表std::forward_ist，用于存放线程。相比双向链表空间利用率更好。
task_list 任务链表类，用于存放任务以及其参数(Any)，内部是用双向链表std::list对数据进行保存。因为任务有请求的先后顺序，而单向链表对这个支持
不太友好。
Start() 初始化线程链表，链表内存放着是std::unique_ptr<std::thread> 用智能指针管理内存资源。这里需要注意的一点是需要传入this指针，每个类方法
都有一个隐藏的this指针，而这个参数在参数列表的第一个。
ThreadRun() 线程池中线程跑的函数，用end_flag判断是否结束，stop_flag判断是否获取stop_mutex锁。获取stop_mutex锁用std::lock_guard进行管理
出来局部范围自动释放。
ThreadPoolEnd() 结束线程池，先停止任务链表的添加与取出，再等待线程结束，释放内存。
ThreadPool()构造，对线程池中线程数量、任务链表的容量、stop_flag及end_flag赋值，对线程链表和任务链表进行初始化。
～ThreadPool析构，若没释放stop_mutex锁则对stop_mutex 进行解锁。若没运行End()函数对线程链表及任务链表资源回收则调用End()，这里std::call_once
对此提供保障。
AddTask() 提供左值参数与右值参数两个版本，当是右值参数时将节省内存的开销。
End()结束线程池，调用了std::call_once(std::once_flag, Callable&& f, Args&&... args)。
Stop() 暂停线程池中线程，对stop_mutex锁进行锁定。
Restart() 重新恢复线程池中线程的运行，对stop_mutex锁进行释放。
namespace Csz_ThreadPool { class ThreadPool { private: //stop_flag停止线程运行，Stop method and Restart method //停止后还是能继续往task_list中加入任务，而线程都处于waiting状态 //因为Stop()锁住了stop_mutex,而线程们正不断‘努力‘获取这个锁调用 //std::lock_guard<std::mutex> guard(stop_mutex) RAII机制， //而在Restart()释放了stop_mutex,线程依次获取此锁后,出了局部范围 //又释放了此锁，以便其他线程获取此锁。 //暂停与重新开始(不建议多线程调用，因为可能会发生死锁或者抛出异常) std::atomic_bool stop_flag; std::mutex stop_mutex; //std::once_flag 调用法std::call_once(promise_one,[]{})若是在调用 //函数里发生了异常或者扔出了异常则下次调用是会再次运行 //保证线程池的释放只进行一次(在多线程同时操作线程池ing) //std::atomic_bool 提供了原子操作 std::once_flag promise_one; std::atomic_bool end_flag; //若直接把线程放进container里则进行拷贝，若放进线程指针则需要 //注意在join时候必须释放内存，若使用智能指针则可以同时避免着两种情况 //使用forward_list原因:只需要把线程指针从一端放进去而不需要从另一端取出来 //而forward_list比list在空间利用率上更高 using ThreadPtr= std::unique_ptr<std::thread>; std::forward_list<ThreadPtr> thread_list; //线程队列 using Task= std::function<void(Any)>; TaskList<Task> task_list; //任务队列 private: //初始化，创建线程并把线程move进去线程队列，方便结束线程池时 //join线程 void Start(const int & T_thread_num) noexcept { #if defined(Csz_test) std::cout<<"Thread::Start()\n"; #endif for (int i= 0; i< T_thread_num; i++) { thread_list.push_front(std::move(ThreadPtr(new std::thread(&ThreadPool::ThreadRun,this)))); //thread_list.emplace_front(&ThreadPool::ThreadRun,this); //重载make_unique } } //线程执行的函数 void ThreadRun() noexcept { Task task; Any parameter; while (!end_flag) { task_list.Take(task,parameter); if (end_flag) return ; #if defined(Csz_test) std::cout<<"ThreadPool::ThreadRun task_begin\n"; #endif task(std::move(parameter)); #if defined(Csz_test) std::cout<<"ThreadPool::ThreadRun task_end\n"; //std::this_thread::sleep_for(std::chrono::seconds(1)); #endif if (stop_flag) { std::lock_guard<std::mutex> guard(stop_mutex); } } } //结束线程，等待线程执行完当前task,这里有个debug,也就是说 //你的结束并不是立即结束而是必须等待线程把所领取的task执行完 //最后清空线程队列以及任务队列 void ThreadPoolEnd() noexcept { #if defined(Csz_test) std::cout<<"ThreadPool::ThreadPoolEnd()\n"; #endif end_flag=true; task_list.Stop(); for (auto &thread : thread_list) if (thread) thread->join(); thread_list.clear(); task_list.Clear(); } public: //构造函数，在赋值参数的时候要严格按照申明顺序！！！ //parameter1 线程队列最大容量，默认为系统核数 //parameter2 任务队列最大容量，默认为 #define SIZE 30 ThreadPool(int T_thread_num= std::thread::hardware_concurrency(),uint8_t T_size= SIZE)noexcept : stop_flag(false),end_flag(false),task_list(T_size) { Start(T_thread_num); } //当忘了结束线程池则有析构函数来执行 ~ThreadPool() { #if defined(Csz_test) std::cout<<"ThreadPool::destructor\n"; #endif if (true== stop_flag) Restart(); End(); } //添加任务，右值 //使用完美转发保证了参数类型原样 void AddTask(Task &&T_task,Any &&T_parameter) noexcept { #if defined(Csz_test) std::cout<<"ThreadPool::AddTask()move\n"; #endif task_list.Put(std::forward<Task>(T_task),std::forward<Any>(T_parameter)); } //添加任务，除了右值 void AddTask(const Task &T_task,const Any &&T_parameter) noexcept { #if defined(Csz_test) std::cout<<"ThreadPool::AddTask()copy\n"; #endif task_list.Put(T_task,T_parameter); } //结束线程池 //call_once保证只正常执行一次 void End() noexcept { #if defined(Csz_test) std::cout<<"ThreadPool::End()\n"; #endif std::call_once(promise_one,[this]{ThreadPoolEnd();}); } //停止线程，但不停止任务添加 void Stop() noexcept { #if defined(Csz_test) std::cout<<"ThreadPool::Stop()\n"; #endif //多线程控制线程池 if (true== stop_flag) return ; stop_mutex.lock(); stop_flag= true; } //恢复线程，线程继续从任务队列中取任务执行 void Restart() noexcept { #if defined(Csz_test) std::cout<<"ThreadPool::Restart()\n"; #endif //Start(T_thread_num); //多线程控制线程池 if (false== stop_flag) return ; stop_flag= false; stop_mutex.unlock(); } }; }
期待下一段时间，嫌弃自己写的代码，批判自己的疏忽。
参考文献 深入应用C++11 代码优化与工程级应用
实现UDP高效接收/响应