• 【Linux】线程池

    一、概念

    1.线程池是一种利用池化技术思想来实现的线程管理技术，主要是为了复用线程、便利地管理线程和任务、并将线程的创建和任务的执行解耦开来。我们可以创建线程池来复用已经创建的线程来降低频繁创建和销毁线程所带来的资源消耗。
    2.线程池是一种线程使用模式。线程过多会带来调度开销，进而影响缓存局部性和整体性能。而线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利用，还能防止过分调度。可用线程数量应该取决于可用的并发处理器、处理器内核、内存、网络sockets等的数量。

    二、线程池的应用场景

    1. 需要大量的线程来完成任务，且完成任务的时间比较短。 WEB 服务器完成网页请求这样的任务，使用线程池技术是非常合适的。因为单个任务小，而任务数量巨大，你可以想象一个热门网站的点击次数。 但对于长时间的任务，比如一个Telnet连接请求，线程池的优点就不明显了。因为 Telnet 会话时间比线程的创建时间大多了。
    2. 对性能要求苛刻的应用，比如要求服务器迅速响应客户请求。
    3. 接受突发性的大量请求，但不至于使服务器因此产生大量线程的应用。突发性大量客户请求，在没有线程池情况下，将产生大量线程，虽然理论上大部分操作系统线程数目最大值不是问题，短时间内产生大量线程可能使内存到达极限，出现错误。

    三、线程池的优点
     

    1.降低资源消耗，复用已创建的线程来降低创建和销毁线程的消耗。
    2.提高响应速度，任务到达时，可以不需要等待线程的创建立即执行。
    3.提高线程的可管理性，使用线程池能够统一的分配、调优和监控。

    四、代码

    这里封装了Task，lockGuard，Thread，以及ThreadPool,具体请从ThreadPool看起，前面的封装只是为了方便ThreadPool使用.

    简化流程：ThreadPool也就是线程池，把任务数据结构（这里是队列）和多个线程封装在一起，当任务被生产出来并被线程池检测到，就会唤醒一个线程去处理，处理方法是把任务拷贝一份到线程独立栈上，并删除该任务。这里使用到了单例模式，把构造函数设为私有，只能调用类的静态成员函数来构造唯一一个线程池对象。

    ThreadPool.hpp

    1. #pragma once 
    2. #include
    3. #include
    4. #include
    5. #include
    6. #include
    7. #include
    8.  
    9.  
    10. #include"Task.hpp"
    11. #include"Thread.hpp"
    12. #include"lockGuard.hpp"
    13.  
    14. //                                单例模式
    15. const int N=6;
    16.  
    17. template<class T>
    18. class ThreadPool
    19. {
    20. private: 
    21.     ThreadPool(int num=N)//构造函数私有化
    22.     :_num(num)
    23.  
    24.     {
    25.         pthread_mutex_init(&_mutex,nullptr);
    26.         pthread_cond_init(&_cond,nullptr);
    27.  
    28.     }
    29.     ThreadPool(const ThreadPool &tp) = delete;//删除默认拷贝构造
    30.     void operator=(const ThreadPool &tp) = delete;//删除赋值函数
    31.     pthread_mutex_t* getlock()
    32.     {
    33.         return &_mutex;
    34.     }
    35.    
    36.  
    37.     void threadwait()
    38.     {
    39.         //挂起一个线程
    40.         pthread_cond_wait(&_cond,&_mutex);
    41.     }
    42.  
    43.     void threadwakeup()
    44.     {
    45.         //唤醒一个线程
    46.         pthread_cond_signal(&_cond);
    47.     }
    48.  
    49. public:
    50.     //创建单例模式对象；
    51.     static ThreadPool* getinstance()
    52.     {  
    53.         if(_instance==nullptr) //当一个对象已经被创建以后，就不进入申请锁并且判断的环节了；
    54.         {
    55.             lockGuard lock(&_mutex_instance);
    56.             if(_instance==nullptr)
    57.             {
    58.                 _instance=new ThreadPool();
    59.                 _instance->init();
    60.                 _instance->start();
    61.             }
    62.         }
    63.         return _instance; 
    64.     }
    65.    
    66.     ~ThreadPool()
    67.     {
    68.         for(auto& e:_threads)
    69.         {
    70.             e.join();
    71.         }
    72.         pthread_mutex_destroy(&_mutex);
    73.         pthread_cond_destroy(&_cond);
    74.     }
    75.  
    76.     bool isEmpty()
    77.     {
    78.         return _tasks.empty();
    79.     }
    80.  
    81.     void init()
    82.     {
    83.         //创建线程
    84.         for(int i=1;i<=_num;++i) 
    85.             //pthread_create(&_threads[i],nullptr,ThreadRoutine,this);
    86.             _threads.push_back(Thread(i,ThreadRoutine,this));
    87.     }
    88.     
    89.  
    90.     void start()
    91.     {   
    92.         //线程启动
    93.         for(auto& e:_threads)
    94.         {
    95.             e.run();
    96.         }
    97.     }
    98.  
    99.     void check()
    100.     {
    101.         for(auto& e:_threads)
    102.         {
    103.            std::cout<<"线程ID"<threadid()<<" , "<threadname()<<"is running··· "<
    104.         }
    105.  
    106.     }
    107.     //放入任务
    108.     void pushtask(const T& task)
    109.     {
    110.         lockGuard lock(&_mutex);
    111.         _tasks.push(task);
    112.         
    113.         threadwakeup();//有新任务进来，唤醒线程去处理
    114.     }
    115.  
    116.     
    117.     T poptask()
    118.     {
    119.         T t=_tasks.front();
    120.         _tasks.pop();
    121.         return t;
    122.     }
    123. private:
    124.     static void* ThreadRoutine(void* args)
    125.     {   
    126.         pthread_detach(pthread_self());
    127.         //指针强转成线程池对象类型；
    128.         ThreadPool* tp=static_cast*>(args);
    129.         while(true)
    130.         {
    131.             
    132.             T t;  
    133.             {    
    134.  
    135.                 lockGuard lock(tp->getlock());
    136.  
    137.                 //1.判断是否有任务
    138.                 //有->处理
    139.                 //无->等待
    140.                 //如果任务队列为空，则等待
    141.                 while(tp->isEmpty())
    142.                 {
    143.                     tp->threadwait();
    144.                 }
    145.                 t=tp->poptask();//从共有区域拿到线程独立栈上；
    146.             }
    147.             
    148.                 t();//调用task类里面的仿函数处理任务
    149.                 std::cout << "thread handler done, result: " << t.formatRes() << std::endl;
    150.             }
    151.     }
    152. private:
    153.     
    154.     std::vector _threads;
    155.     int _num;//线程池里有几个线程；
    156.  
    157.     std::queue _tasks; // 使用STL的自动扩容的特性
    158.  
    159.     pthread_mutex_t _mutex;
    160.     pthread_cond_t _cond;
    161.  
    162.     static ThreadPool *_instance;//懒汉方式实现单例模式
    163.     static pthread_mutex_t _mutex_instance;//单例对象有自己的锁
    164. };
    165.  
    166.  
    167. //静态变量初始化
    168. template<class T>
    169. ThreadPool* ThreadPool::_instance=nullptr;
    170.  
    171. template<class T>
    172. pthread_mutex_t ThreadPool::_mutex_instance=PTHREAD_MUTEX_INITIALIZER;

    Task.hpp

    1. #pragma once
    2. #include
    3. #include
    4.  
    5.  
    6. class Task
    7. {
    8. public:
    9.     Task(){}
    10.     
    11.     Task(int x,int y,char op):_x(x),_y(y),_op(op),_result(0),_exitCode(0)  {}
    12.  
    13.     int operator()()
    14.     {
    15.         switch(_op)
    16.         {   
    17.             case '+':
    18.                 _result=_x+_y;
    19.                 break;
    20.             case '-':
    21.                 _result=_x-_y;
    22.                 break;
    23.             case '*':
    24.                 _result=_x*_y;
    25.                 break;
    26.             case '/':
    27.                 {
    28.                     if(_y==0) 
    29.                         _exitCode=-1;
    30.                     else   
    31.                         _result=_x/_y;    
    32.                 }
    33.                 break;
    34.             case '%':
    35.                 {
    36.                     if(_y==0) 
    37.                         _exitCode=-2;
    38.                     else   
    39.                         _result=_x%_y;    
    40.                 }
    41.                 break;
    42.             default:
    43.                 break;
    44.         }
    45.     }
    46.  
    47.     //任务
    48.     std::string formatArg()
    49.     {
    50.         return std::to_string(_x) + _op + std::to_string(_y) + "=?";
    51.     }
    52.  
    53.     //任务处理结果
    54.      std::string formatRes()
    55.     {
    56.         return std::to_string(_result) + "(" + std::to_string(_exitCode)+ ")";
    57.     }
    58.  
    59.     ~Task() {}
    60.    
    61.  
    62.  
    63. private:
    64.     int _x;
    65.     int _y;
    66.     char _op;
    67.     int _result;
    68.     int _exitCode;
    69. };

    lockGuard.hpp

    1. #pragma once
    2.  
    3. #include 
    4. #include 
    5.  
    6. class Mutex//成员：加锁函数和解锁函数
    7. {
    8. public:
    9.     Mutex(pthread_mutex_t* pmutex):_pmutex(pmutex)   {}
    10.     
    11.     void lock()
    12.     {
    13.         pthread_mutex_lock(_pmutex);
    14.     }
    15.    
    16.        void unlock()
    17.     {
    18.         pthread_mutex_unlock(_pmutex);
    19.     }
    20.    
    21.  
    22.     ~Mutex(){}
    23.  
    24. private:
    25.     pthread_mutex_t* _pmutex;//需要传入一个互斥量（锁）的指针；
    26. };
    27.  
    28.  
    29. //对Mutex进行二次封装；
    30. //创建该对象时自动加锁，析构时自动解锁；
    31. class lockGuard
    32. {   
    33. public:
    34.     lockGuard(pthread_mutex_t* pmutex):_mutex(pmutex)//利用锁的指针构建Mutex对象
    35.     {
    36.         _mutex.lock();
    37.     }
    38.  
    39.    ~lockGuard()
    40.     {
    41.         _mutex.unlock();
    42.     }
    43.  
    44. private:
    45.     Mutex _mutex;//类内创建对象
    46. };

    Thread.hpp

    1. #pragma once
    2.  
    3. #include 
    4. #include 
    5. #include 
    6.  
    7. class Thread
    8. {
    9. public:
    10.     typedef void* (*func_t) (void*);
    11.     typedef enum
    12.     {
    13.         NEW=0,
    14.         RUNNING,
    15.         EXITED
    16.     }ThreadStatus;
    17.  
    18. public:
    19.       //状态：new，running，exited
    20.         int status()
    21.         {
    22.             return _status;
    23.         }
    24.  
    25.         //线程名
    26.         std::string threadname()
    27.         {
    28.             return _name;
    29.         }
    30.  
    31.         //线程ID（共享库中的进程地址空间的虚拟地址）
    32.         pthread_t threadid()
    33.         {
    34.             if(_status==RUNNING)//线程已经被创建，线程id已经输入到成员变量_tid中；
    35.                 return _tid;
    36.             else 
    37.             {  
    38.                 return 0;
    39.             }
    40.         }
    41.  
    42. public:
    43.  
    44.         //构造函数；
    45.         Thread(int num,func_t func,void* args)//num代表第几个线程
    46.         :_tid(0),
    47.         _status(NEW),
    48.         _func(func),
    49.         _args(args)
    50.  
    51.         {
    52.             char name[128];
    53.             snprintf(name,sizeof(name),"thread-%d",num);
    54.             _name=name;
    55.         }
    56.         
    57.         //析构函数
    58.         ~Thread(){}
    59.  
    60.       
    61.  
    62.         //静态成员函数不能访问类内所有成员，因为没有this指针；
    63.         static void* runHelper(void *args)
    64.         {
    65.             Thread* td=(Thread*)args;
    66.             (*td)();//调用仿函数执行线程的回调函数；
    67.             return nullptr; 
    68.         }
    69.  
    70.         void operator()()//仿函数
    71.         {
    72.             //如果函数指针不为空，则执行该函数指针指向的回调函数；
    73.             if(_func!=nullptr)  _func(_args);
    74.         }
    75.  
    76.         //创建线程
    77.         void run()
    78.         {
    79.             //因为runHelper函数必须只能有一个void*参数，所以runHelper函数在类内必须定义为static，这样才没有this指针；
    80.             int n=pthread_create(&_tid,nullptr,runHelper,this);
    81.             if(n!=0) return exit(1);//线程创建失败，那么直接退出进程；
    82.             _status=RUNNING;
    83.         }
    84.  
    85.         //等待线程结束
    86.         void join()
    87.         {
    88.             int n=pthread_join(_tid,nullptr);
    89.             if(n!=0) 
    90.             {
    91.                 std::cerr<<"main thread join thread "<<_name<<" error "<
    92.                 return;
    93.             }
    94.             _status=EXITED;//线程退出；
    95.         }
    96.        
    97. private:
    98.     pthread_t _tid;//线程ID（原生线程库中为该线程所创建的TCB起始虚拟地址）
    99.     std::string _name;//线程名
    100.     func_t _func;//线程要执行的回调
    101.     void* _args;//线程回调函数参数
    102.     ThreadStatus _status;//枚举类型：状态
    103. };
    104.

    main.cc

    1. #include
    2. #include
    3. #include"ThreadPool.hpp"
    4.  
    5.  
    6.  
    7. using namespace std;
    8.  
    9. int main()
    10. {   
    11.     // unique_ptr < ThreadPool > tp(new ThreadPool(20));
    12.     // tp->init();
    13.     // tp->start();
    14.     // tp->check();
    15.  
    16.     //实例化类，并且多次调用getinstance()函数；
    17.     printf("0X%x\n", ThreadPool::getinstance());
    18.     printf("0X%x\n", ThreadPool::getinstance());
    19.     printf("0X%x\n", ThreadPool::getinstance());
    20.     printf("0X%x\n", ThreadPool::getinstance());
    21.     printf("0X%x\n", ThreadPool::getinstance());
    22.     printf("0X%x\n", ThreadPool::getinstance());
    23.     while(true)
    24.     {
    25.         // 充当生产者, 从网络中读取数据，构建成为任务，推送给线程池
    26.         int x, y;
    27.         char op;
    28.         std::cout << "please Enter x> ";
    29.         std::cin >> x;
    30.         std::cout << "please Enter y> ";
    31.         std::cin >> y;
    32.         std::cout << "please Enter op(+-*/%)> ";
    33.         std::cin >> op;
    34.  
    35.         Task t(x,y,op);
    36.         ThreadPool::getinstance()->pushtask(t); //单例对象也有可能在多线程场景中使用！
    37.     }
    38.  
    39.  
    40.     return 0;
    41. }

    makefile

    1. ThreadPool:main.cc
    2. 	g++ $^ -o $@ -std=c++11 -lpthread
    3. .PHONY:clean
    4. clean:
    5. 	rm -f ThreadPool

  • 相关阅读:
    周鸿祎自称3次破解特斯拉云端系统：安全隐患巨大
    大厂标配 | 百亿级并发系统设计 | 学完薪资框框涨
    Hadoop3教程（二十）：MapReduce的工作机制总结
    【Leetcode】70. 爬楼梯
    23.1 微服务理论基础
    Python中，如何读取和写入文件？
    Simulink模型加密共享
    java计算机毕业设计学生网上请假系统源码+系统+数据库+lw文档
    2696. Minimum String Length Afte
    构造与析构
  • 原文地址：https://blog.csdn.net/zzxz8/article/details/132638557