3.3 BlockingQueue.hpp 生产者消费者模型头文件

#pragma once  #include  #include  #include  #include "LockGuard.hpp"  namespace My_Queue{       const int DEF_SIZE = 10;          template     class BlockingQueue     {     private:        // 任务队列         std::queue _queue;         size_t _cap; // 阻塞队列的容量         pthread_mutex_t _mtx; // 互斥锁         pthread_cond_t _pro_cond; // 生产者条件变量         pthread_cond_t _con_cond; // 消费者条件变量     public:         BlockingQueue(size_t cap = DEF_SIZE)           :_cap(cap)         {             // 初始化锁与条件变量             pthread_mutex_init(&_mtx,nullptr);             pthread_cond_init(&_pro_cond,nullptr);             pthread_cond_init(&_con_cond,nullptr);         }          ~BlockingQueue(){             //销毁锁与条件变量             pthread_mutex_destroy(&_mtx);             pthread_cond_destroy(&_pro_cond);             pthread_cond_destroy(&_con_cond);         }        // 生产数据（入队）         void Push(const T& inData){             // 加锁（RAII风格）             LockGuard lock(&_mtx);             // 循环判断条件是否满足             while(IsFull()){                 pthread_cond_wait(&_pro_cond, &_mtx);             }             _queue.push(inData);             // 可以加策略唤醒，比如生产一半才唤醒消费者             pthread_cond_signal(&_con_cond);             // 自动解锁         }          // 消费数据（出队）         void Pop(T* outData){             // 加锁（RAII 风格）             LockGuard lock(&_mtx);             // 循环判读条件是否满足             while(IsEmpty()) {                 pthread_cond_wait(&_con_cond, &_mtx);             }             *outData = _queue.front();             _queue.pop();             // 可以加策略唤醒，比如消费完后才唤醒生产者             pthread_cond_signal(&_pro_cond);             // 自动解锁         }         private:                  //判断是否为满         bool IsFull(){             return _queue.size() == _cap;         }         //判断是否为空         bool IsEmpty(){             return _queue.empty();         }     };  }

3.4 LockGuard.hpp 自动化锁头文件

#pragma once  #include  class LockGuard { private:     pthread_mutex_t* _pmtx; public:     LockGuard(pthread_mutex_t *pmtx)     :_pmtx(pmtx)     {        //加锁        pthread_mutex_lock(_pmtx);     }     ~LockGuard(){         //解锁         pthread_mutex_unlock(_pmtx);     } }; 

 3.5 Task.hpp 任务类

  现在需要修改 Task.hpp 任务头文件中的 Task 任务类，将其修改为一个服务于 网络通信中业务处理 的任务类（也就是 Service() 业务处理函数）

   在 Service() 业务处理函数中，需要包含 socket 套接字、客户端 IP、客户端端口号 等必备信息，除此之外，我们还可以将 可调用对象（Service() 业务处理函数） 作为参数传递给 Task 对象.

#pragma once  #include  #include   namespace My_task{      // Service() 业务处理函数的类型     using cb_t = std::function;      class Task{     public:         // 可以再提供一个默认构造（防止部分场景中构建对象失败）         Task()         {}          Task(int sock, const std::string& ip, const uint16_t& port, const cb_t& cb)             :_sock(sock)             ,_ip(ip)             ,_port(port)             ,_cb(cb)         {}          // 重载运算操作，用于回调 [业务处理函数]         void operator()(){             // 直接回调 cb [业务处理函数] 即可             _cb(_sock, _ip, _port);         }      private:         int _sock;         std::string _ip;         uint16_t _port;         cb_t _cb; // 回调函数     }; }

3.6 server.hpp 头文件

准备工作完成后，接下来就是往 server.hpp 服务器头文件中添加组件了

注意：

• 在构建 Task 对象时，需要使用 bind 绑定类内函数，避免参数不匹配
• 当前的线程池是单例模式，在 Task 任务对象构建后，通过线程池操作句柄 push 对象即可

// server.hpp  #pragma once  #include #include #include #include #include #include #include #include"err.hpp" #include #include #include #include"ThreadPool.hpp" #include"Task.hpp"  namespace My_server{      // 默认端口号     const uint16_t default_port = 1111;     //全连接队列的最大长度     const int backlog = 32;     using func_t = std::function;          //前置声明     class server;     //包含我们所需参数的类型     class ThreadData{       public:          ThreadData(int sock,const std::string&ip,const uint16_t&port,server*ptr)           :_sock(sock)           ,_clientip(ip)           ,_clientport(port)           ,_current(ptr)          {}       public:         int _sock;         std::string _clientip;         uint16_t _clientport;         server* _current;     };      class server     {     private:         /* data */         //套接字         int _listensock;         //端口号         uint16_t _port;         // 判断服务器是否结束运行         bool _quit;         // 外部传入的回调函数         func_t _func;     public:          server(const func_t &func,const uint16_t &port = default_port)          :_func(func)          ,_port(port)          ,_quit(false)         {}          ~server(){}          //初始化服务器         void InitServer(){             //1 创建套接字             _listensock = socket(AF_INET,SOCK_STREAM,0);             if(_listensock == -1){                 //绑定失败                 std::cerr<<"Create Socket Fail:"<::getInstance()->pushTask(t);             }         }               void Service(int sock,const std::string &clientip,const uint16_t &clientport){              char buff[1024];             std::string who = clientip + "-" + std::to_string(clientport);             while(true){                 // 以字符串格式读取，预留\0的位置                 ssize_t n = read(sock,buff,sizeof(buff)-1);                 if(n>0){                     //读取成功                     buff[n]='\0';                     std::cout<<"Server get: "<< buff <<" from "<

         void Service(int sock,const std::string &clientip,const uint16_t &clientport){              char buff[1024];             std::string who = clientip + "-" + std::to_string(clientport);                             // 以字符串格式读取，预留\0的位置                 ssize_t n = read(sock,buff,sizeof(buff)-1);                 if(n>0){                     //读取成功                     buff[n]='\0';                     std::cout<<"Server get: "<< buff <<" from "<