1.任务队列
类声明(Task,TaskQueue)
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 #ifndef C___PRACTICE_TASKQUEUE_H #define C___PRACTICE_TASKQUEUE_H #include <queue> #include <pthread.h> using function = void (*)(void * arg);struct Task { Task (){ working = nullptr ; arg = nullptr ; } Task (function working, void * arg){ this ->working = working; this ->arg = arg; } function working; void * arg; }; class TaskQueue {private : std::queue<Task> taskQueue; pthread_mutex_t QueueLock; public : TaskQueue (); ~TaskQueue (); void addTask (Task& task) ; void addTask (function working, void * arg) ; Task takeTask () ; inline size_t getTaskSize () { return taskQueue.size (); } }; #endif
其中Task 是任务类,里边有两个成员,分别是两个指针 void(*)(void*) 和 void*
另外一个类TaskQueue 是任务队列,提供了添加任务、取出任务、存储任务、获取任务个数、线程同步的功能。
类定义
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 #include "TaskQueue.h" TaskQueue::TaskQueue () { pthread_mutex_init (&QueueLock, nullptr ); } TaskQueue::~TaskQueue () { pthread_mutex_destroy (&QueueLock); } void TaskQueue::addTask (Task& task) { pthread_mutex_lock (&QueueLock); taskQueue.push (task); pthread_mutex_unlock (&QueueLock); } void TaskQueue::addTask (function working, void *arg) { pthread_mutex_lock (&QueueLock); taskQueue.push (Task (working, arg)); pthread_mutex_unlock (&QueueLock); } Task TaskQueue::takeTask () { pthread_mutex_lock (&QueueLock); Task t; if (!taskQueue.empty ()) { t = taskQueue.front (); taskQueue.pop (); } pthread_mutex_unlock (&QueueLock); return t; }
2.线程池
类声明(ThreadPool)
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 #ifndef C___PRACTICE_THREADPOOL_H #define C___PRACTICE_THREADPOOL_H #include "TaskQueue.h" class ThreadPool {private : TaskQueue* taskQueue = nullptr ; pthread_t managerID; pthread_t * threadIDs; int minNum; int maxNum; int busyNum; int liveNum; int exitNum; pthread_mutex_t threadLock; pthread_mutex_t busyLock; pthread_cond_t notEmpty; bool shutdown; static void * worker (void * arg) ; static void * manager (void * arg) ; void threadExit () ; public : ThreadPool (int min,int max); ~ThreadPool (); void addTask (Task task) ; inline int getBusyNumber () ; inline int getAliveNumber () ; }; #endif
类定义
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 #include "ThreadPool.h" #include <iostream> #include <cstring> #include <unistd.h> using std::cout;using std::endl;#define CHECK_TIME 5 #define ADD_NUM 2 #define EXIT_NUM 2 template <typename T>inline void safe_delete_void_ptr (void * &target) ;ThreadPool::ThreadPool (int min, int max) { do { taskQueue = new (std::nothrow)TaskQueue; if (taskQueue == nullptr ) { cout << "new taskQueue error" << endl; break ; } threadIDs = new (std::nothrow)pthread_t [max]; if (threadIDs == nullptr ) { cout << "new threadIDs error" << endl; break ; } memset (threadIDs, 0 , sizeof (pthread_t ) * max); minNum = min; maxNum = max; busyNum = 0 ; liveNum = min; exitNum = 0 ; if (pthread_mutex_init (&threadLock, nullptr ) != 0 || pthread_mutex_init (&busyLock, nullptr ) != 0 || pthread_cond_init (¬Empty, nullptr ) != 0 ) { cout << "init the lock or cond error" << endl; break ; } shutdown = false ; pthread_create (&managerID, nullptr , manager, this ); for (int i = 0 ; i < min; ++i) { pthread_create (&threadIDs[i], nullptr , worker, this ); } return ; }while (false ); if (threadIDs) { delete [] threadIDs; threadIDs = nullptr ; } if (taskQueue) { delete taskQueue; taskQueue = nullptr ; } } void * ThreadPool::worker (void * arg) { auto pool = static_cast <ThreadPool*>(arg); while (true ) { pthread_mutex_lock (&pool->threadLock); while (pool->taskQueue->getTaskSize ()==0 && !pool->shutdown) { cout<<"thread " <<pthread_self ()<<" is waiting" <<endl; pthread_cond_wait (&pool->notEmpty, &pool->threadLock); if (pool->exitNum > 0 ) { pool->exitNum--; if (pool->liveNum > pool->minNum) { pool->liveNum--; pthread_mutex_unlock (&pool->threadLock); pool->threadExit (); } } } if (pool->shutdown) { pthread_mutex_unlock (&pool->threadLock); pool->threadExit (); } Task task = pool->taskQueue->takeTask (); pthread_mutex_unlock (&pool->threadLock); pthread_mutex_lock (&pool->busyLock); pool->busyNum++; pthread_mutex_unlock (&pool->busyLock); cout << "thread " << pthread_self () << " start working" << endl; task.working (task.arg); safe_delete_void_ptr <int >(task.arg); task.arg = nullptr ; cout << "thread " << pthread_self () << " end working" << endl; pthread_mutex_lock (&pool->busyLock); pool->busyNum--; pthread_mutex_unlock (&pool->busyLock); } } void * ThreadPool::manager (void *arg) { auto pool = static_cast <ThreadPool*>(arg); while (!pool->shutdown) { sleep (CHECK_TIME); pthread_mutex_lock (&pool->threadLock); size_t queueSize = pool->taskQueue->getTaskSize (); int liveNum = pool->liveNum; int busyNum = pool->busyNum; pthread_mutex_unlock (&pool->threadLock); if (queueSize>liveNum&&liveNum<pool->maxNum) { pthread_mutex_lock (&pool->threadLock); int counter = 0 ; for (int i=0 ;i<pool->maxNum&&counter<ADD_NUM&&pool->liveNum<pool->maxNum;i++) { if (pool->threadIDs[i]==0 ) { pthread_create (&pool->threadIDs[i],nullptr ,worker,pool); counter++; pool->liveNum++; } } pthread_mutex_unlock (&pool->threadLock); } if (busyNum*2 <liveNum&&liveNum>pool->minNum) { pthread_mutex_lock (&pool->threadLock); pool->exitNum = EXIT_NUM; pthread_mutex_unlock (&pool->threadLock); for (int i=0 ;i<EXIT_NUM;i++) { pthread_cond_signal (&pool->notEmpty); } } } return nullptr ; } void ThreadPool::threadExit () { pthread_t tid = pthread_self (); for (int i = 0 ; i < maxNum; ++i) { if (threadIDs[i] == tid) { threadIDs[i] = 0 ; cout << "threadExit() called, threadID: " << tid << " exiting..." << endl; break ; } } pthread_exit (NULL ); } void ThreadPool::addTask (Task task) { if (shutdown) { return ; } taskQueue->addTask (task); pthread_cond_signal (¬Empty); } int ThreadPool::getBusyNumber () { pthread_mutex_lock (&busyLock); int busyNum = this ->busyNum; pthread_mutex_unlock (&busyLock); return busyNum; } int ThreadPool::getAliveNumber () { pthread_mutex_lock (&threadLock); int liveNum = this ->liveNum; pthread_mutex_unlock (&threadLock); return liveNum; } ThreadPool::~ThreadPool () { this ->shutdown = true ; pthread_join (this ->managerID, nullptr ); for (int i = 0 ; i < this ->liveNum; ++i) { pthread_cond_signal (¬Empty); } pthread_mutex_destroy (&threadLock); pthread_mutex_destroy (&busyLock); pthread_cond_destroy (¬Empty); if (taskQueue) { delete taskQueue; } if (threadIDs) { delete [] threadIDs; } } template <typename T>void safe_delete_void_ptr (void * &target) { if (target) { T* temp = static_cast <T*>(target); delete temp; temp = nullptr ; target = nullptr ; } }
遇到的问题
注意25行用new分配内存C++中,new操作符默认不会返回NULL,而是会在分配内存失败时抛出一个std::bad_alloc异常。这样做的好处是,你不需要在每次使用new操作符后检查返回值是否为NULL,而是可以用try/catch语句统一处理异常。如果你想让new操作符在分配内存失败时返回NULL,而不是抛出异常,你可以使用std::nothrow参数
解决方案:
try/catch语句统一处理异常
std::nothrow参数
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 try { taskQueue = new TaskQueue; } catch (std::bad_alloc &e) { cout << "new taskQueue error" << endl; break ; } try { threadIDs = new pthread_t [max]; } catch (std::bad_alloc &e) { cout << "new threadIDs error" << endl; break ; }
注意这里133行代码,C++中不能用delete直接删除一个void*指针,因为delete需要调用被删除对象的析构函数,而void指针没有类型信息,所以无法确定调用哪个析构函数。这样做会导致未定义行为,可能会出现崩溃或内存泄漏。你必须在delete之前把void指针转换回它原来的类型,才能正确地释放内存。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 template <typename T>inline void safe_delete_void_ptr (void * &target) ;template <typename T>void safe_delete_void_ptr (void * &target) { if (target) { T* temp = static_cast <T*>(target); delete temp; temp = nullptr ; target = nullptr ; } }
[ C++指针编程你要小心的陷阱——如何优雅的释放指针void*_c++释放void指针_二流小宝的博客-CSDN博客](https://blog.csdn.net/SweetTool/article/details/77688337?ops_request_misc={"request_id"%3A"168977949416800222881353"%2C"scm"%3A"20140713.130102334.."}&request_id=168977949416800222881353&biz_id=0&utm_medium=distribute.pc_search_result.none-task-blog-2~all~baidu_landing_v2~default-1-77688337-null-null.142^v90^insert_down28v1,239^v2^insert_chatgpt&utm_term=c%2B%2B delete void 指针&spm=1018.2226.3001.4187)
3.修改为模板类处理指针释放问题
TaskQueue.h TaskQueue.cpp ThreadPool.h ThreadPool.cpp test.cpp 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 #ifndef C___PRACTICE_TASKQUEUE_H #define C___PRACTICE_TASKQUEUE_H #include <queue> #include <pthread.h> using function = void (*)(void * arg);template <typename T>struct Task { Task <T>(){ working = nullptr ; arg = nullptr ; } Task <T>(function working, void * arg){ this ->working = working; this ->arg = (T*)arg; } function working; T* arg; }; template <typename T>class TaskQueue { private : std::queue<Task<T>> taskQueue; pthread_mutex_t QueueLock; public : TaskQueue (); ~TaskQueue (); void addTask (Task<T> task) ; void addTask (function working, void * arg) ; Task<T> takeTask () ; inline size_t getTaskSize () { return taskQueue.size (); } }; #endif
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 #include "TaskQueue.h" #include <iostream> template <typename T>TaskQueue<T>::TaskQueue () { pthread_mutex_init (&QueueLock, nullptr ); } template <typename T>TaskQueue<T>::~TaskQueue () { pthread_mutex_destroy (&QueueLock); } template <typename T>void TaskQueue<T>::addTask (Task<T> task){ pthread_mutex_lock (&QueueLock); taskQueue.push (task); pthread_mutex_unlock (&QueueLock); } template <typename T>void TaskQueue<T>::addTask (function working, void *arg){ pthread_mutex_lock (&QueueLock); taskQueue.push (Task <T>(working, arg)); pthread_mutex_unlock (&QueueLock); } template <typename T>Task<T> TaskQueue<T>::takeTask () { pthread_mutex_lock (&QueueLock); Task<T> t; if (!taskQueue.empty ()) { t = taskQueue.front (); taskQueue.pop (); } pthread_mutex_unlock (&QueueLock); return t; }
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 #ifndef C___PRACTICE_THREADPOOL_H #define C___PRACTICE_THREADPOOL_H #include "TaskQueue.h" #include "TaskQueue.cpp" template <typename T>class ThreadPool { private : TaskQueue<T>* taskQueue; pthread_t managerID; pthread_t * threadIDs; int minNum; int maxNum; int busyNum; int liveNum; int exitNum; pthread_mutex_t threadLock; pthread_mutex_t busyLock; pthread_cond_t notEmpty; bool shutdown; static void * worker (void * arg) ; static void * manager (void * arg) ; void threadExit () ; public : ThreadPool (int min,int max); ~ThreadPool (); void addTask (Task<T> task) ; inline int getBusyNumber () ; inline int getAliveNumber () ; }; #endif
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 #include "ThreadPool.h" #include <iostream> #include <cstring> #include <unistd.h> using std::cout;using std::endl;#define CHECK_TIME 5 #define ADD_NUM 2 #define EXIT_NUM 2 template <typename T>ThreadPool<T>::ThreadPool (int min, int max) { do { taskQueue = new TaskQueue<T>; if (taskQueue == nullptr ) { cout << "new taskQueue error" << endl; break ; } threadIDs = new pthread_t [max]; if (threadIDs == nullptr ) { cout << "new threadIDs error" << endl; break ; } memset (threadIDs, 0 , sizeof (pthread_t ) * max); minNum = min; maxNum = max; busyNum = 0 ; liveNum = min; exitNum = 0 ; if (pthread_mutex_init (&threadLock, nullptr ) != 0 || pthread_mutex_init (&busyLock, nullptr ) != 0 || pthread_cond_init (¬Empty, nullptr ) != 0 ) { cout << "init the lock or cond error" << endl; break ; } shutdown = false ; pthread_create (&managerID, nullptr , manager, this ); for (int i = 0 ; i < min; ++i) { pthread_create (&threadIDs[i], nullptr , worker, this ); } return ; }while (false ); if (threadIDs) { delete [] threadIDs; threadIDs = nullptr ; } if (taskQueue) { delete taskQueue; taskQueue = nullptr ; } } template <typename T>void * ThreadPool<T>::worker (void * arg){ auto pool = static_cast <ThreadPool*>(arg); while (true ) { pthread_mutex_lock (&pool->threadLock); while (pool->taskQueue->getTaskSize ()==0 && !pool->shutdown) { cout<<"thread " <<pthread_self ()<<" is waiting" <<endl; pthread_cond_wait (&pool->notEmpty, &pool->threadLock); if (pool->exitNum > 0 ) { pool->exitNum--; if (pool->liveNum > pool->minNum) { pool->liveNum--; pthread_mutex_unlock (&pool->threadLock); pool->threadExit (); } } } if (pool->shutdown) { pthread_mutex_unlock (&pool->threadLock); pool->threadExit (); } Task task = pool->taskQueue->takeTask (); pthread_mutex_unlock (&pool->threadLock); pthread_mutex_lock (&pool->busyLock); pool->busyNum++; pthread_mutex_unlock (&pool->busyLock); cout << "thread " << pthread_self () << " start working" << endl; task.working (task.arg); delete task.arg; task.arg = nullptr ; cout << "thread " << pthread_self () << " end working" << endl; pthread_mutex_lock (&pool->busyLock); pool->busyNum--; pthread_mutex_unlock (&pool->busyLock); } } template <typename T>void * ThreadPool<T>::manager (void *arg){ auto pool = static_cast <ThreadPool*>(arg); while (!pool->shutdown) { sleep (CHECK_TIME); pthread_mutex_lock (&pool->threadLock); size_t queueSize = pool->taskQueue->getTaskSize (); int liveNum = pool->liveNum; int busyNum = pool->busyNum; pthread_mutex_unlock (&pool->threadLock); if (queueSize>liveNum&&liveNum<pool->maxNum) { pthread_mutex_lock (&pool->threadLock); int counter = 0 ; for (int i=0 ;i<pool->maxNum&&counter<ADD_NUM&&pool->liveNum<pool->maxNum;i++) { if (pool->threadIDs[i]==0 ) { pthread_create (&pool->threadIDs[i],nullptr ,worker,pool); counter++; pool->liveNum++; } } pthread_mutex_unlock (&pool->threadLock); } if (busyNum*2 <liveNum&&liveNum>pool->minNum) { pthread_mutex_lock (&pool->threadLock); pool->exitNum = EXIT_NUM; pthread_mutex_unlock (&pool->threadLock); for (int i=0 ;i<EXIT_NUM;i++) { pthread_cond_signal (&pool->notEmpty); } } } return nullptr ; } template <typename T>void ThreadPool<T>::threadExit (){ pthread_t tid = pthread_self (); for (int i = 0 ; i < maxNum; ++i) { if (threadIDs[i] == tid) { threadIDs[i] = 0 ; cout << "threadExit() called, threadID: " << tid << " exiting..." << endl; break ; } } pthread_exit (NULL ); } template <typename T>void ThreadPool<T>::addTask (Task<T> task){ if (shutdown) { return ; } taskQueue->addTask (task); pthread_cond_signal (¬Empty); } template <typename T>int ThreadPool<T>::getBusyNumber (){ pthread_mutex_lock (&busyLock); int busyNum = this ->busyNum; pthread_mutex_unlock (&busyLock); return busyNum; } template <typename T>int ThreadPool<T>::getAliveNumber (){ pthread_mutex_lock (&threadLock); int liveNum = this ->liveNum; pthread_mutex_unlock (&threadLock); return liveNum; } template <typename T>ThreadPool<T>::~ThreadPool () { this ->shutdown = true ; pthread_join (this ->managerID, nullptr ); for (int i = 0 ; i < this ->liveNum; ++i) { pthread_cond_signal (¬Empty); } pthread_mutex_destroy (&threadLock); pthread_mutex_destroy (&busyLock); pthread_cond_destroy (¬Empty); if (taskQueue) { delete taskQueue; } if (threadIDs) { delete [] threadIDs; } }
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 #include <iostream> #include "ThreadPool.h" #include "ThreadPool.cpp" #include <unistd.h> void working (void * arg) { int num = *(int *)arg; std::cout << "thread " << pthread_self () << " is working, number = " << num << std::endl; sleep (1 ); } int main () { ThreadPool<int > pool (3 , 10 ) ; for (int i=0 ;i<100 ;i++) { int *num = new int ; *num = i+100 ; pool.addTask (Task <int >(working, num)); } sleep (30 ); return 0 ; }
特殊:当有类模板时,类模板的声明和定义分开写时,需要包含头文件和源文件,否则会报找不到定义的错误