[多线程]基于单例懒汉模式的线程池的实现

标题：[多线程]基于单例懒汉模式的线程池的实现
@水墨不写bug

---

  

---

   本文我们将要实现一个线程池，其中用到的一些模块如LockGuard，Log等在之前的文章中有所报告，以是本文就不再赘述。
  

---

一、锁的RAII计划

1. #pragma once
2. #include <pthread.h>
4. // 使用锁需要频繁调用系统调用，十分麻烦
5. // 于是实现锁的RAII设计
6. class LockGuard
7. {
8. private:
9.         pthread_mutex_t *GetMutex() { return _mutex; }
11. public:
12.         LockGuard(pthread_mutex_t *mutex)
13.                 : _mutex(mutex)
14.         {
15.                 pthread_mutex_lock(_mutex);
16.         }
17.         ~LockGuard()
18.         {
19.                 pthread_mutex_unlock(_mutex);
20.         }
22. private:
23.         pthread_mutex_t *_mutex;
24. };
26. /*
27. *之前的理解错误了：不需要init和destroy，因为锁本身就是存在的！
28. *锁在一般情况下会内置在类的内部，需要使用（加锁）的时候，把锁的地址传进来就行了
29. *在构造函数内加锁，析构函数内解锁。
30. *e.g.
31. while(ture)
32. {
33.         LockGuard lockguard(td->_mutex);
34.         if(tickets > 0)
35.         {
36.                 //抢票
37.         }
38.         else
39.         {
40.                 break;
41.         }
42. }
43. *while内每一次进行循环，都需要创建一个新的锁，创建即加锁，if代码块结束即为解锁
44. */
46. /*
47.         void UnLock() { pthread_mutex_unlock(&_mutex); }
48.         void Lock() { pthread_mutex_lock(&_mutex); }
49. */

复制代码

以上代码是基于RAII（Resource Acquisition Is Initialization）计划模式的锁管理类LockGuard。
该类用于简化对pthread库的互斥锁（pthread_mutex_t）的利用，确保在加锁息争锁过程中的安全性和轻便性。

---

二、任务类举例

1. #pragma once
2. #include <functional>
3. #include <iostream>
5. using std::cout;
6. using std::endl;
8. // void DownLoad()
9. // {
10. //     cout<<"I am a DownTask"<<endl;
11. // }
13. // 任务类，包含各种任务
14. class Task
15. {
16. private:
17.     using task_t = std::function<void()>;
19.     void Excute()
20.     {
21.         _result = _x + _y;
22.     }
24. public:
25.     Task()
26.         : _x(0), _y(0)
27.     {
28.     }
29.     Task(int a, int b)
30.         : _x(a), _y(b)
31.     {
32.     }
34.     void operator()()
35.     {
36.         Excute();
37.     }
39.     void Debug()
40.     {
41.         cout<<"Debug : "<<_x<<" + "<<_y<<" = "<<_result<<endl;
42.     }
44. private:
45.     int _x;
46.     int _y;
47.     int _result;
48. };

复制代码

这个类封装了一个具体的任务：两数相加。
尽管这是一项简单的任务（这样计划是为了便于理解线程池的部分），但是其实任务类可以根据用户的需要来自行计划，可以是下载资源，请求网页html等。
   任务类的计划目标就是一个演示作用。
  

---

三、日记体系的计划

1. #pragma once
2. #include <string>
3. #include <iostream>
4. #include <unistd.h>
5. #include <sys/types.h>
6. #include <ctime>
7. #include <cstdarg>
8. #include <fstream>
9. #include <cstring>
10. #include "LockGuard.hpp"
12. namespace log_ddsm
13. {
14.     using std::cout;
15.     using std::endl;
17.     // 日志信息管理
19.     enum LEVEL
20.     {
21.         DEBUG = 1,   // 调试信息
22.         INFO = 2,    // 提示信息
23.         WARNING = 3, // 警告
24.         ERROR = 4,   // 错误，但是不影响服务正常运行
25.         FATAL = 5,   // 致命错误，服务无法正常运行
26.     };
27.     enum SIZE
28.     {
29.         TIMESIZE = 128,
30.         LOGSIZE = 1024,
32.         FILE_TYPE_LOG_SIZE = 2048
33.     };
34.     enum TYPE
35.     {
36.         SCREEN_TYPE = 8,
37.         FILE_TYPE = 16
38.     };
40.     // 默认日志文件名称
41.     const char *DEFAULT_LOG_NAME = "./log.txt";
42.     // 全局锁，保护打印日志
43.     pthread_mutex_t gmutex = PTHREAD_MUTEX_INITIALIZER;
45.     struct logMessage
46.     {
47.         std::string _level; // 日志等级
48.         int _level_num;     // 日志等级的int格式
50.         pid_t _pid;             // 这条日志的进程id
51.         std::string _file_name; // 文件名
52.         int _file_number;       // 行号
53.         std::string _cur_time;  // 当时的时间
54.         std::string _log_info;  // 日志正文
55.     };
57.     // 通过int获取日志等级
58.     std::string getLevel(int level)
59.     {
60.         switch (level)
61.         {
62.         case 1:
63.             return "DEBUG";
64.         case 2:
65.             return "INFO";
66.         case 3:
67.             return "WARNING";
68.         case 4:
69.             return "ERROR";
70.         case 5:
71.             return "FATAL";
72.         default:
73.             return "UNKNOWN";
74.         }
75.     }
77.     std::string getCurTime()
78.     {
79.         time_t cur = time(nullptr);
80.         struct tm *curtime = localtime(&cur);
81.         char buf[TIMESIZE] = {0};
82.         snprintf(buf, sizeof(buf), "%d-%d-%d %d:%d:%d",
83.                  curtime->tm_year + 1900,
84.                  curtime->tm_mon + 1,
85.                  curtime->tm_mday,
86.                  curtime->tm_hour,
87.                  curtime->tm_min,
88.                  curtime->tm_sec);
89.         return buf;
90.     }
92.     class Log
93.     {
94.     private:
95.         void FlushMessage(const logMessage &lg)
96.         {
97.             // 互斥锁，保护Print过程
98.             LockGuard lockguard(&gmutex);
100.             // 过滤逻辑
101.             // 致命错误到文件逻辑
102.             if (lg._level_num >= _ignore_level)
103.             {
104.                 if (_print_type == SCREEN_TYPE)
105.                     PrintToScreen(lg);
106.                 else if (_print_type == FILE_TYPE)
107.                     PrintToFile(lg);
108.                 else
109.                     std::cerr << __FILE__ << " " << __LINE__ << ":" << " UNKNOWN_TYPE " << std::endl;
110.             }
111.         }
112.         void PrintToScreen(const logMessage &lg)
113.         {
114.             printf("[%s][%d][%s][%d][%s] %s", lg._level.c_str(),
115.                    lg._pid,
116.                    lg._file_name.c_str(),
117.                    lg._file_number,
118.                    lg._cur_time.c_str(),
119.                    lg._log_info.c_str());
120.         }
122.         void PrintToFile(const logMessage &lg)
123.         {
124.             std::ofstream out(_log_file_name, std::ios::app); // 追加打印
125.             if (!out.is_open())
126.             {
127.                 std::cerr << __FILE__ << " " << __LINE__ << ":" << " LOG_FILE_OPEN fail " << std::endl;
128.                 return;
129.             }
130.             char log_txt[FILE_TYPE_LOG_SIZE] = {0}; // 缓冲区
131.             snprintf(log_txt, sizeof(log_txt), "[%s][%d][%s][%d][%s] %s", lg._level.c_str(),
132.                      lg._pid,
133.                      lg._file_name.c_str(),
134.                      lg._file_number,
135.                      lg._cur_time.c_str(),
136.                      lg._log_info.c_str());
138.             out.write(log_txt, strlen(log_txt)); // 写入文件
139.         }
141.     public:
142.         // 打印方式——默认向显示器打印
143.         // 如果选择文件类型打印，默认的文件名称是当前目录的log.txt
144.         //
145.         Log(int print_type = SCREEN_TYPE)
146.             : _print_type(print_type), _log_file_name(DEFAULT_LOG_NAME), _ignore_level(DEBUG)
147.         {
148.         }
150.         // 只有一个全局对象，多个构造函数多余,不如设计一个构造，添加一个设置log文件名称的接口
151.         //  Log(int print_type, const char *logname = DEFAULT_LOG_NAME)
152.         //      : _print_type(print_type), _log_file_name(logname), _ignore_level(1)
153.         //  {
154.         //  }
156.         void Enable(int type)
157.         {
158.             _print_type = type;
159.         }
161.         /// @brief 加载日志信息
162.         /// @param format 格式化输出
163.         /// @param  可变参数
164.         /*区分了本层和商城之后，就容易设置参数了*/
165.         void load_message(int level, std::string filename, int filenumber, const char *format, ...)
166.         {
167.             logMessage lg;
168.             lg._level = getLevel(level);
169.             lg._level_num = level;
171.             lg._pid = getpid();
172.             lg._file_name = filename;
173.             lg._file_number = filenumber;
174.             lg._cur_time = getCurTime();
176.             // valist + vsnprintf 处理可变参数
177.             va_list ap;
178.             va_start(ap, format);
179.             char log_info[LOGSIZE] = {0};
180.             vsnprintf(log_info, sizeof(log_info), format, ap);
181.             va_end(ap);
182.             lg._log_info = log_info;
184.             // 打印逻辑
185.             FlushMessage(lg);
186.         }
188.         /// @brief
189.         /// @param ignorelevel
190.         /* DEBUG = 1,   调试信息
191.          *INOF = 2,     提示信息
192.          *WARNING = 3,  警告
193.          *ERROR = 4,    错误，但是不影响服务正常运行
194.          *FATAL = 5,    致命错误，服务无法正常运行 */
195.         void SetIgnoreLevel(int ignorelevel)
196.         {
197.             _ignore_level = ignorelevel;
198.         }
199.         void SetLogFileName(const char *newlogfilename)
200.         {
201.             _log_file_name = newlogfilename;
202.         }
203.         ~Log() {}
205.     private:
206.         int _print_type;
207.         std::string _log_file_name;
208.         int _ignore_level;
209.     };
211.     // 全局的Log对象
212.     Log lg;
214.     // 打印日志信息
215.     // e.g.
216.     // LOG(DEBUG,"this is a test message %d %f %c", 13 , 9.81 , 'A' );
217.     //
219. // 使用日志的一般格式
220. #define LOG(Level, Format, ...)                                            \
221.     do                                                                     \
222.     {                                                                      \
223.         lg.load_message(Level, __FILE__, __LINE__, Format, ##__VA_ARGS__); \
224.     } while (0)
226.     /// 无法设计为inlne——__VA_ARGS只能出现在宏替换中
227.     // inline void LOG(int level,const char* format, ...)
228.     // {
229.     //     lg.load_message(level,__FILE__,__LINE__,format,__VA_ARGS__);
230.     // }
232. // 往文件打印
233. #define ENABLE_FILE()         \
234.     do                        \
235.     {                         \
236.         lg.Enable(FILE_TYPE); \
237.     } while (0)
238. // 往显示器打印
239. #define ENABLE_SCREEN()         \
240.     do                          \
241.     {                           \
242.         lg.Enable(SCREEN_TYPE); \
243.     } while (0)
244. //设置日志忽略等级
245. #define SET_IGNORE_LEVEL(Level)  \
246.     do                           \
247.     {                            \
248.         lg.SetIgnoreLevel(Level); \
249.     } while (0)
250. //设置日志文件名称
251. #define SET_LOG_FILENAME(Name)  \
252.     do                          \
253.     {                           \
254.         lg.SetLogFileName(Name); \
255.     } while (0)
257. }; // namespace log_ddsm

复制代码

以上是日记体系的计划，具体的报告请参考我之前的这篇文章：
   《[HTTP协议]应用层协议HTTP从入门到深刻理解并落地部署自己的云服务(2)实操部署》：
https://blog.csdn.net/2301_79465388/article/details/146114051?spm=1001.2014.3001.5501
  四、基于POSIX线程封装自己的线程模块

1. #pragma once
2. #include <pthread.h>
3. #include <vector>
4. #include <functional>
5. #include <iostream>
6. #include <cassert>
7. #include <cstring>
8. #include "Log.hpp"
10. using namespace log_ddsm;
11. // 理解C++11做的工作，封装自己的原生线程
12. // 创建线程，执行任务
13. namespace ThreadMoudle
14. {
15.     // 支持模板
16.     using func_t = std::function<void(const std::string &name)>;
18.     // 线程控制-自主管理原生线程
19.     class Thread
20.     {
21.     private:
22.         // 线程创建后例行执行的函数
23.         static void *Routine(void *args) // 设置为static，防止this指针干扰类型匹配
24.         {
25.             Thread *self = static_cast<Thread *>(args);
26.             self->Excute();
27.             return nullptr;
28.         }
30.         void Excute()
31.         {
32.             _isrunning = true;
33.             _func(_name);
34.             _isrunning = false;
35.         }
37.     public:
38.         /// @brief 构造线程
39.         /// @param name 线程名
40.         /// @param func 线程要执行的任务
41.         Thread(const std::string &name, func_t func)
42.             : _name(name), _tid(), _isrunning(false), _func(func)
43.         {
44.         }
46.         bool Start()
47.         {
48.             int res = ::pthread_create(&_tid, nullptr, Routine, this); // 要求类型：void*（void*）——有this指针导致类型不匹配
49.             if (res != 0)
50.             {
51.                 LOG(ERROR, "pthread_create failed with error: %s\n", strerror(res));
53.                 // std::cerr << __FILE__ << " " << __LINE__ << ":" << "pthread_creat: error!" << std::endl;
54.                 // printf("pthread_create failed with error: %s\n", strerror(res));
56.                 return false;
57.             }
58.             else
59.             {
60.                 return true;
61.             }
62.         }
64.         bool Stop()
65.         {
66.             if (_isrunning)
67.             {
68.                 int res = ::pthread_cancel(_tid);
69.                 if (res != 0)
70.                 {
71.                     // std::cerr << __FILE__ << " " << __LINE__ << ":" << "pthread_cancel: error!" << std::endl;
72.                     // printf("pthread_cancel failed with error: %s\n", strerror(res));
73.                     LOG(ERROR, "pthread_cancel failed with error: %s\n", strerror(res));
74.                     return false;
75.                 }
76.                 else
77.                 {
78.                     return true;
79.                 }
80.             }
81.             //走到这里，说明线程已经stop了
82.             return true;
83.         }
85.         // Stop之后join
86.         bool Join()
87.         {
88.             if (!_isrunning)
89.             {
90.                 int res = ::pthread_join(_tid, nullptr);
91.                 if (res != 0)
92.                 {
93.                     // std::cout<<"Debug res:"<<res<<std::endl;
95.                     // std::cerr << __FILE__ << " " << __LINE__ << ":" << "pthread_join: error!" << std::endl;
96.                     // printf("pthread_join failed with error: %s\n", strerror(res));
97.                     LOG(ERROR, "pthread_join failed with error: %s\n", strerror(res));
99.                     return false;
100.                 }
101.                 else
102.                 {
103.                     return true;
104.                 }
105.             }
106.             else
107.             {
108.                 // Running时join
109.                 LOG(INFO, "join while Running\n");
111.                 // std::cerr << "join while Running" << std::endl;
112.                 return false;
113.             }
114.         }
115.         ~Thread() {}
117.         bool Status() { return _isrunning; }
118.         std::string Name() { return _name; }
120.     private:
121.         std::string _name; // 线程的名字
122.         pthread_t _tid;    // 线程的类型
123.         bool _isrunning;   // 线程是否在运行
124.         func_t _func;      // 线程将要执行的回调函数
125.     };
127. }; // namespace ThreadMoudle

复制代码

以上代码实现了一个线程管理模块 ThreadMoudle，其中包含一个线程类 Thread。它封装了POSIX线程（pthread）的创建、启动、制止和加入（join）操纵，并提供了一些辅助功能。
计划要点

• 线程管理：
  
  
  
  • 该模块封装了POSIX线程的基本操纵，提供了一个易于利用的接口来管理线程的创建、启动、制止和加入操纵。
    
  
  
• 可重用性和可扩展性：
  
  
  
  • 通过利用std::function<void(const std::string &name)>，可以将恣意符合该签名的回调函数传递给线程对象，从而实现任务的多样性和机动性。
    
  
  
• 线程安全性：
  
  
  
  • 通过内部状态标记（_isrunning）和得当的日记记录，确保线程的状态和操纵的一致性，制止了多次启动或制止同一线程的问题。
    
  
  

主要功能

• 线程创建和启动：
  
  
  
  • Thread::Start()：调用pthread_create创建一个新线程，并启动实行Routine函数。Routine函数将调用实例的Excute方法实行具体任务。
    
  • Routine函数被声明为静态函数，以制止this指针干扰范例匹配。
    
  
  
• 线程实行具体任务：
  
  
  
  • Thread::Excute()：设置线程状态为运行中，并调用回调函数_func实行具体任务。
    
  
  
• 线程制止：
  
  
  
  • Thread::Stop()：调用pthread_cancel取消运行中的线程。对于不符合逻辑的情况：线程未运行，则记录错误日记。
    
  
  
• 线程加入：
  
  
  
  • Thread::Join()：调用pthread_join等待线程制止。对于不符合逻辑的情况：线程仍在运行，则记录错误日记。
    
  
  
• 日记记录：
  
  
  
  • 利用日记模块（log_ddsm）记录线程操纵中的错误信息，便于调试和问题排查。
    
  
  

利用示例

1. #include "Thread.hpp"
3. void MyTask(const std::string &name)
4. {
5.     std::cout << "Thread " << name << " is running." << std::endl;
6.     int count = 5;
7.     while(count--)
8.     {
9.         std::cout<< count<<std::endl;
10.     }
11. }
13. int main()
14. {
15.     ThreadMoudle::Thread myThread("TestThread", MyTask);
16.     if (myThread.Start())
17.     {
18.         std::cout << "Thread started successfully." << std::endl;
19.     }
21.     // 等待线程完成任务
22.     myThread.Join();
24.     return 0;
25. }

复制代码

结果：

在这个示例中：

• 创建一个名为"TestThread"的线程对象，并传递任务函数MyTask。
  
• 调用myThread.Start()启动线程。
  
• 等待线程完成任务后，调用myThread.Join()等待线程制止。
  

---

五、基于懒汉模式的线程池主体逻辑计划

1. #pragma once
2. #include <queue>
3. #include <unistd.h>
4. #include "Log.hpp"
5. #include "Thread.hpp" //using pthread lib
7. using std::cout;
8. using std::endl;
10. using ThreadMoudle::func_t;
11. using namespace log_ddsm;
13. enum NUM
14. {
15.     DEFAULTNUM = 10
16. };
18. /// @brief
19. /// @tparam T:任务类型
20. //懒汉模式设计单例ThreadPool
22. template <typename T>
23. class ThreadPool
24. {
25. private:
26.     void LockQueue() // 加锁
27.     {
28.         pthread_mutex_lock(&_taskqueue_mutex);
29.     }
30.     void UnLockQueue() // 解锁
31.     {
32.         pthread_mutex_unlock(&_taskqueue_mutex);
33.     }
34.     void Sleep() // 在条件变量_taskqueue_cond下等待
35.     {
36.         pthread_cond_wait(&_taskqueue_cond, &_taskqueue_mutex);
37.     }
39.     void WakeUp() // 叫醒一个进程
40.     {
41.         pthread_cond_signal(&_taskqueue_cond);
42.     }
43.     void WakeUpAll() // 叫醒所有线程
44.     {
45.         pthread_cond_broadcast(&_taskqueue_cond);
46.     }
47.     bool IsEmpty()
48.     {
49.         return _taskqueue.empty();
50.     }
52.     // thread将要执行的函数逻辑
53.     void HandlerTask(const std::string &name)
54.     {
55.         // 一直循环：等待任务——执行任务——等待任务，直到运行到break逻辑
56.         while (true)
57.         {
58.             LockQueue();
60.             /*  优先级上，先考虑任务队列不为空，只有任务队列为空，再考虑是否要线程退出*/
61.             while (IsEmpty() && _isrunning) // 没有任务则等待,while——在唤醒之后先判断是否队列仍然有任务，否则再次等待
62.             {
63.                 _sleep_num++;
64.                 Sleep();
65.                 _sleep_num--;
66.             }
67.             if (IsEmpty() && !_isrunning)
68.             {
69.                 UnLockQueue();
71.                 LOG(DEBUG, "thread quit\n");
72.                 // cout << "thread quit" << endl;
73.                 break;
74.             }
76.             T t; // 从taskqueue中获取任务
77.             t = _taskqueue.front();
78.             _taskqueue.pop();
79.             UnLockQueue();
81.             // 执行任务，在解锁之后进行，执行任务要与从队列中获取任务并发
82.             // 不能在临界区中进行
83.             t();
85.             LOG(DEBUG, "%s", name.c_str());
86.             // cout << name << " : ";
87.             t.Debug();
88.         }
89.     }
90.     // 缺省个数——10
91.     ThreadPool(size_t threadnum)
92.         : _threadnum(threadnum), _taskqueue(), _isrunning(false), _sleep_num(0) // 初始化列中调用has a的构造？？
93.     {
94.         // 初始化
95.         pthread_mutex_init(&_taskqueue_mutex, nullptr);
96.         pthread_cond_init(&_taskqueue_cond, nullptr);
97.     }
98.     // 初始化线程池
99.     void Init()
100.     {
101.         func_t func = std::bind(&ThreadPool::HandlerTask, this, std::placeholders::_1);
102.         for (int i = 0; i < _threadnum; ++i)
103.         {
104.             std::string name = "thread - " + std::to_string(i + 1);
106.             _threads.emplace_back(name, func); // 减少一次拷贝，直接在vector内构造
107.             // usleep(10000);
108.             // cout << "emplaced i+1 :" << i + 1 << endl;
109.         }
110.     }
112.     // 开始运行
113.     void Start()
114.     {
115.         _isrunning = true; // 单进程访问不需要保护
117.         for (int i = 0; i < _threadnum; ++i)
118.         {
119.             _threads[i].Start();
120.             LOG(DEBUG, "thread - %d started \n", i + 1);
121.         }
122.     }
124. public:
125.     static ThreadPool<T> *GetInstance()
126.     {
127.         if (_ptp == nullptr)
128.         {
129.             LockGuard lockguard(&_sig_mutex);
130.             if (_ptp == nullptr)
131.             {
132.                 LOG(INFO, "create threadspool\n");
133.                 _ptp = new ThreadPool<T>(DEFAULTNUM);//线程安全问题
134.                 _ptp->Init();
135.                 _ptp->Start();
136.             }
137.         }
138.         else
139.         {
140.             LOG(INFO, "get threadspool\n");
141.         }
142.         return _ptp;
143.     }
145.     // 停止运行
146.     void Stop()
147.     {
148.         LockQueue();
149.         _isrunning = false;
150.         WakeUpAll();
151.         UnLockQueue();
152.     }
154.     // 向任务队列加入任务
155.     void PushTask(const T &in)
156.     {
157.         LockQueue();
158.         if (_isrunning == true) // 防止出现线程池想要退出但是 却被一直分配任务，导致无法退出
159.         {
160.             _taskqueue.push(in); // 加入任务
161.             if (_sleep_num > 0)
162.                 WakeUp();
163.         }
164.         UnLockQueue();
165.     }
167.     ~ThreadPool()
168.     {
169.         for (int i = 0; i < _threadnum; ++i)
170.         {
171.             LOG(INFO, "thread-%d joined\n", i + 1);
172.             _threads[i].Join();
173.         }
174.         pthread_mutex_destroy(&_taskqueue_mutex);
175.         pthread_cond_destroy(&_taskqueue_cond);
176.     }
178. private:
179.     size_t _threadnum;                          // 线程池中线程的个数
180.     std::vector<ThreadMoudle::Thread> _threads; // vector维护线程池
182.     std::queue<T> _taskqueue; // 任务队列
183.     bool _isrunning;          // 标识线程池是否在运行
185.     /*多线程对taskqueue访问需要整体使用——加锁*/
186.     pthread_mutex_t _taskqueue_mutex; // 用于维护taskqueue的mutex cond
187.     pthread_cond_t _taskqueue_cond;
189.     int _sleep_num; // 休眠的线程个数
191.     static ThreadPool<T> *_ptp; // 单例对象
192.     static pthread_mutex_t _sig_mutex;
193. };
195. // 静态成员在类外初始化
196. template <class T>
197. ThreadPool<T> *ThreadPool<T>::_ptp = nullptr;
199. // 第一次保护获取单例的mutex
200. template <class T>
201. pthread_mutex_t ThreadPool<T>::_sig_mutex = PTHREAD_MUTEX_INITIALIZER;

复制代码

类 ThreadPool 的成员函数详解

• LockQueue() 和 UnLockQueue()：
  
  
  
  • 这两个函数分别用来加锁息争锁 _taskqueue_mutex 互斥锁。
    
  • 目标是保护任务队列 _taskqueue 的访问，确保在多线程环境下对任务队列的操纵是线程安全的。
    
  1. void LockQueue() // 加锁
  2. {
  3.     pthread_mutex_lock(&_taskqueue_mutex);
  4. }
  6. void UnLockQueue() // 解锁
  7. {
  8.     pthread_mutex_unlock(&_taskqueue_mutex);
  9. }

  复制代码
  
  
• Sleep() 和 WakeUp()、WakeUpAll()：
  
  
  
  • Sleep() 函数使线程在条件变量 _taskqueue_cond 上等待，等待条件满足时被叫醒。
    
  • WakeUp() 函数叫醒一个在 _taskqueue_cond 上等待的线程。
    
  • WakeUpAll() 函数叫醒所有在 _taskqueue_cond 上等待的线程。
    
  • 这些函数用于实现线程休眠和叫醒机制，当任务队列为空时，线程进入休眠状态；当有新任务加入时，叫醒线程实行任务。
    
  1. void Sleep() // 在条件变量_taskqueue_cond下等待
  2. {
  3.     pthread_cond_wait(&_taskqueue_cond, &_taskqueue_mutex);
  4. }
  6. void WakeUp() // 叫醒一个进程
  7. {
  8.     pthread_cond_signal(&_taskqueue_cond);
  9. }
  11. void WakeUpAll() // 叫醒所有线程
  12. {
  13.     pthread_cond_broadcast(&_taskqueue_cond);
  14. }

  复制代码
  
  
• IsEmpty()：
  
  
  
  • 检查任务队列是否为空。
    
  • 返回布尔值，表现任务队列是否为空。
    
  1. bool IsEmpty()
  2. {
  3.     return _taskqueue.empty();
  4. }

  复制代码
  
  
• HandlerTask(const std::string &name)：
  
  
  
  • 线程实行的主要函数逻辑。
    
  • 线程在循环中等待任务队列中的任务，当有任务时，从任务队列中取出任务并实行。
    
  • 当任务队列为空且线程池不再运行时，线程退出循环，竣事实行。
    
  1. // thread将要执行的函数逻辑
  2. void HandlerTask(const std::string &name)
  3. {
  4.     // 一直循环：等待任务——执行任务——等待任务，直到运行到break逻辑
  5.     while (true)
  6.     {
  7.         LockQueue();
  9.         /*  优先级上，先考虑任务队列不为空，只有任务队列为空，再考虑是否要线程退出*/
  10.         while (IsEmpty() && _isrunning) // 没有任务则等待,while——在唤醒之后先判断是否队列仍然有任务，否则再次等待
  11.         {
  12.             _sleep_num++;
  13.             Sleep();
  14.             _sleep_num--;
  15.         }
  16.         if (IsEmpty() && !_isrunning)
  17.         {
  18.             UnLockQueue();
  20.             LOG(DEBUG, "thread quit\n");
  21.             break;
  22.         }
  24.         T t; // 从taskqueue中获取任务
  25.         t = _taskqueue.front();
  26.         _taskqueue.pop();
  27.         UnLockQueue();
  29.         // 执行任务，在解锁之后进行，执行任务要与从队列中获取任务并发
  30.         // 不能在临界区中进行
  31.         t();
  33.         LOG(DEBUG, "%s", name.c_str());
  34.         // cout << name << " : ";
  35.         t.Debug();
  36.     }
  37. }

  复制代码
  
  
• 构造函数 ThreadPool(size_t threadnum)：
  
  
  
  • 初始化线程池，包括线程数量、任务队列、互斥锁和条件变量。
    
  • 设置 _threadnum 为线程池中的线程数量，初始化其他成员变量。
    
  1. ThreadPool(size_t threadnum)
  2.     : _threadnum(threadnum), _taskqueue(), _isrunning(false), _sleep_num(0)
  3. {
  4.     pthread_mutex_init(&_taskqueue_mutex, nullptr);
  5.     pthread_cond_init(&_taskqueue_cond, nullptr);
  6. }

  复制代码
  
  
• Init()：
  
  
  
  • 初始化线程池中的线程。
    
  • 创建 thread 对象，并将 HandlerTask 绑定为线程实行的任务函数。这一操纵的目标是为了使emplace_back传入Thread构造函数的范例匹配。
    
  1. void Init()
  2. {
  3.     func_t func = std::bind(&ThreadPool::HandlerTask, this, std::placeholders::_1);
  4.     for (int i = 0; i < _threadnum; ++i)
  5.     {
  6.         std::string name = "thread - " + std::to_string(i + 1);
  7.         _threads.emplace_back(name, func);
  8.     }
  9. }

  复制代码
  
  
• Start()：
  
  
  
  • 启动线程池中的线程。
    
  • 设置 _isrunning 为 true，然后启动线程池中每个线程。
    
  1. void Start()
  2. {
  3.     _isrunning = true;
  4.     for (int i = 0; i < _threadnum; ++i)
  5.     {
  6.         _threads[i].Start();
  7.         LOG(DEBUG, "thread - %d started \n", i + 1);
  8.     }
  9. }

  复制代码
  
  
• GetInstance()：
  
  
  
  • 获取线程池的单例实例。
    
  • 利用双重检查锁定机制确保线程安全。
    
  1. static ThreadPool<T> *GetInstance()
  2. {
  3.     if (_ptp == nullptr)
  4.     {
  5.         LockGuard lockguard(&_sig_mutex);
  6.         if (_ptp == nullptr)
  7.         {
  8.             LOG(INFO, "create threadspool\n");
  9.             _ptp = new ThreadPool<T>(DEFAULTNUM);
  10.             _ptp->Init();
  11.             _ptp->Start();
  12.         }
  13.     }
  14.     else
  15.     {
  16.         LOG(INFO, "get threadspool\n");
  17.     }
  18.     return _ptp;
  19. }

  复制代码
  
  
• Stop()：
  
  
  
  • 制止线程池的运行。
    
  • 设置 _isrunning 为 false，并叫醒所有休眠的线程。
    
  1. void Stop()
  2. {
  3.     LockQueue();
  4.     _isrunning = false;
  5.     WakeUpAll();
  6.     UnLockQueue();
  7. }

  复制代码
  
  
• PushTask(const T &in)：
  
  
  
  • 向任务队列中添加任务。
    
  • 如果线程池正在运行，添加任务到任务队列，并叫醒一个休眠的线程。
    
  1. void PushTask(const T &in)
  2. {
  3.     LockQueue();
  4.     if (_isrunning == true)
  5.     {
  6.         _taskqueue.push(in);
  7.         if (_sleep_num > 0)
  8.             WakeUp();
  9.     }
  10.     UnLockQueue();
  11. }

  复制代码
  
  
• 析构函数 ~ThreadPool()：
  
  
  
  • 销毁线程池，等待所有线程竣事并释放资源。
    
  • 销毁互斥锁和条件变量。
    
  1. ~ThreadPool()
  2. {
  3.     for (int i = 0; i < _threadnum; ++i)
  4.     {
  5.         LOG(INFO, "thread-%d joined\n", i + 1);
  6.         _threads[i].Join();
  7.     }
  8.     pthread_mutex_destroy(&_taskqueue_mutex);
  9.     pthread_cond_destroy(&_taskqueue_cond);
  10. }

  复制代码
  
  

静态成员初始化

• 静态成员变量 _ptp 和 _sig_mutex 在类外举行初始化。
  

1. template <class T>
2. ThreadPool<T> *ThreadPool<T>::_ptp = nullptr;
4. template <class T>
5. pthread_mutex_t ThreadPool<T>::_sig_mutex = PTHREAD_MUTEX_INITIALIZER;

复制代码

六、总结

   ThreadPool 类实现了一个线程池的基本功能，固然你也可以在这个线程池的根本上扩充实现自己需要的功能。
计划中利用单例模式确保只有一个线程池实例,利用互斥锁和条件变量保证多线程环境下的线程安全,利用日记记录监控线程池的运行状态。
  

---

完~
转载请注明出处

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