• linux C语言 手写线程池

    main.cpp

    1. #include 
    2. #include "threadpool.h"
    3. #include 
    4. #include 
    5. #include 
    6.  
    7. void taskFunc(void* arg)
    8. {
    9.     int num = *((int*)arg);
    10.     printf("thread %ld is working, number = %d\n", pthread_self(), num);
    11.     sleep(1);
    12. }
    13.  
    14. int main()
    15. {
    16.     printf("%s 向你问好!\n", "threadpool");
    17.     // 创建线程池
    18.     ThreadPool* pool = threadPoolCreate(3, 10, 100);
    19.     for (int i = 0; i < 100; i++)
    20.     {
    21.         int* num = (int*)malloc(sizeof(int));
    22.         *num = i + 100;
    23.         threadPoolAdd(pool, taskFunc, num);
    24.     }
    25.     sleep(30);
    26.     threadPoolDestory(pool);
    27.     return 0;
    28. }

    threadpool.h

    1. #pragma once
    2. #include 
    3. class threadpool
    4. {
    5. };
    6. typedef struct ThreadPool ThreadPool;
    7. // 创建线程池并初始化
    8. ThreadPool* threadPoolCreate(int min, int max, int queueSize);
    9.  
    10. // 销毁线程池
    11. int threadPoolDestory(ThreadPool* pool);
    12.  
    13. // 给线程池添加任务
    14. void threadPoolAdd(ThreadPool* pool, void(*func)(void*), void* arg);
    15.  
    16. // 获取线程池中工作的线程个数
    17. int threadPoolBusyNum(ThreadPool* pool);
    18.  
    19. // 获取线程池中活着的线程的个数
    20. int threadPoolAliveNum(ThreadPool* pool);
    21.  
    22. void* worker(void* arg);
    23. void* manager(void* arg);
    24. void threadExit(ThreadPool* pool);
    25.  
    26.  
    27.  
    28.

     threadpool.cpp

     

    1. #include "threadpool.h"
    2. #include 
    3. #include 
    4. #include 
    5. #include  
    6.  
    7. const int NUMBER = 2;
    8.  
    9. // 任务结构体
    10. typedef struct Task
    11. {
    12. 	void (*function)(void* arg);
    13. 	void* arg;
    14. }Task;
    15.  
    16. // 线程池结构体
    17. struct ThreadPool
    18. {
    19. 	// 任务队列
    20. 	Task* taskQ;
    21. 	int queueCapacity; // 队列容量
    22. 	int queueSize;     // 当前任务个数
    23. 	int queueFront;    // 队头 -- 取数据
    24. 	int queueRear;	   // 队尾 -- 放数据
    25.  
    26. 	pthread_t managerId;  // 管理者线程ID
    27. 	pthread_t* threadIDs; // 工作的线程ID
    28. 	int minNum; // 最小线程数
    29. 	int maxNum; // 最大线程数
    30. 	int busyNum;// 忙碌线程数
    31. 	int liveNum;// 存活线程数
    32. 	int exitNum;// 要销毁线程个数
    33. 	pthread_mutex_t mutexPool; // 锁整个线程池
    34. 	pthread_mutex_t mutexBusy; // 锁busyNum变量
    35. 	pthread_cond_t notFull;    // 任务队列是不是满了
    36. 	pthread_cond_t notEmpty;   // 任务队列是不是空了
    37.  
    38. 	int shutdown; // 是不是要销毁线程, 销毁为1, 不销毁为0
    39. };
    40.  
    41. ThreadPool* threadPoolCreate(int min, int max, int queueSize)
    42. {
    43. 	ThreadPool* pool = (ThreadPool*)malloc(sizeof(ThreadPool));
    44. 	do
    45. 	{
    46. 		if (pool == NULL)
    47. 		{
    48. 			printf("malloc threadpool fail .. \n");
    49. 			break;
    50. 		}
    51. 		pool->threadIDs = (pthread_t*)malloc(sizeof(pthread_t) * max);
    52. 		if (pool->threadIDs == NULL)
    53. 		{
    54. 			printf("malloc threadIDs fail .. \n");
    55. 			break;
    56. 		}
    57. 		memset(pool->threadIDs, 0, sizeof(pthread_t) * max);
    58. 		pool->minNum = min;
    59. 		pool->maxNum = max;
    60. 		pool->busyNum = 0;
    61. 		pool->liveNum = min; // 和最小个数相等
    62. 		pool->exitNum = 0;
    63.  
    64. 		if (pthread_mutex_init(&pool->mutexPool, NULL) != 0 ||
    65. 			pthread_mutex_init(&pool->mutexBusy, NULL) != 0 ||
    66. 			pthread_cond_init(&pool->notEmpty, NULL) != 0 ||
    67. 			pthread_cond_init(&pool->notFull, NULL) != 0)
    68. 		{
    69. 			printf("mutex init fail ... \n");
    70. 			break;
    71. 		}
    72. 		pool->taskQ = (Task*)malloc(sizeof(Task) * queueSize);
    73. 		pool->queueCapacity = queueSize;
    74. 		pool->queueSize = 0;
    75. 		pool->queueFront = 0;
    76. 		pool->queueRear = 0;
    77. 		pool->shutdown = 0;
    78.  
    79. 		// 创建线程
    80. 		pthread_create(&pool->managerId, NULL, manager, pool);
    81. 		for (int i = 0; i < min; i++)
    82. 		{
    83. 			pthread_create(&pool->threadIDs[i], NULL, worker, pool);
    84. 		}
    85. 		return pool;
    86. 	} while (0);
    87.  
    88. 	// 释放资源
    89. 	if (pool && pool->threadIDs)
    90. 	{
    91. 		free(pool->threadIDs);
    92. 	}
    93. 	if (pool && pool->taskQ)
    94. 	{
    95. 		free(pool->taskQ);
    96. 	}
    97. 	if (pool)
    98. 	{
    99. 		free(pool);
    100. 	}
    101.  
    102. 	return NULL;
    103. }
    104.  
    105. int threadPoolDestory(ThreadPool* pool)
    106. {
    107. 	if (pool == NULL)
    108. 	{
    109. 		return -1;
    110. 	}
    111. 	// 关掉线程池
    112. 	pool->shutdown = 1;
    113. 	// 阻塞回收管理者线程
    114. 	// 唤醒阻塞的消费者线程
    115. 	for (int i = 0; i < pool->liveNum; i++)
    116. 	{
    117. 		pthread_cond_signal(&pool->notEmpty);
    118. 	}
    119. 	pthread_join(pool->managerId, NULL);
    120.  
    121. 	// 释放堆内存
    122. 	if (pool->taskQ)
    123. 	{
    124. 		free(pool->taskQ);
    125. 	}
    126. 	if (pool->threadIDs)
    127. 	{
    128. 		free(pool->threadIDs);
    129. 	}
    130.  
    131. 	pthread_mutex_destroy(&pool->mutexPool);
    132. 	pthread_mutex_destroy(&pool->mutexBusy);
    133. 	pthread_cond_destroy(&pool->notEmpty);
    134. 	pthread_cond_destroy(&pool->notFull);
    135. 	free(pool);
    136. 	pool = NULL;
    137. 	return 0;
    138. }
    139.  
    140. void threadPoolAdd(ThreadPool* pool, void(*func)(void*), void* arg)
    141. {
    142. 	pthread_mutex_lock(&pool->mutexPool);
    143. 	while (pool->queueSize==pool->queueCapacity && !pool->shutdown)
    144. 	{
    145. 		// 阻塞生产者线程
    146. 		pthread_cond_wait(&pool->notFull, &pool->mutexPool);
    147. 	}
    148. 	if (pool->shutdown)
    149. 	{
    150. 		pthread_mutex_unlock(&pool->mutexPool);
    151. 		return;
    152. 	}
    153. 	// 添加任务
    154. 	pool->taskQ[pool->queueRear].function = func;
    155. 	pool->taskQ[pool->queueRear].arg = arg;
    156. 	pool->queueRear = (pool->queueRear + 1) % pool->queueCapacity;
    157. 	pool->queueSize++;
    158. 	pthread_cond_signal(&pool->notEmpty);
    159. 	pthread_mutex_unlock(&pool->mutexPool);
    160. }
    161.  
    162. int threadPoolBusyNum(ThreadPool* pool)
    163. {
    164. 	pthread_mutex_lock(&pool->mutexBusy);
    165. 	int busyNum = pool->busyNum;
    166. 	pthread_mutex_unlock(&pool->mutexBusy);
    167.  
    168. 	return busyNum;
    169. }
    170.  
    171. int threadPoolAliveNum(ThreadPool* pool)
    172. {
    173. 	pthread_mutex_lock(&pool->mutexPool);
    174. 	int aliveNum = pool->liveNum;
    175. 	pthread_mutex_unlock(&pool->mutexPool);
    176. 	return aliveNum;
    177. }
    178.  
    179. void* worker(void* arg)
    180. {
    181. 	ThreadPool* pool = (ThreadPool*)arg;
    182. 	while (1)
    183. 	{
    184. 		pthread_mutex_lock(&pool->mutexPool);
    185. 		// 当前任务队列是否为空
    186. 		while (pool->queueSize == 0 && !pool->shutdown)
    187. 		{
    188. 			// 阻塞工作线程
    189. 			pthread_cond_wait(&pool->notEmpty, &pool->mutexPool);
    190.  
    191. 			// 判断是不是要销毁线程
    192. 			if (pool->exitNum > 0)
    193. 			{
    194. 				pool->exitNum--;
    195. 				if (pool->liveNum > pool->minNum)
    196. 				{
    197. 					pool->liveNum--;
    198. 					pthread_mutex_unlock(&pool->mutexPool);
    199. 					threadExit(pool);
    200. 				}
    201. 			}
    202. 		}
    203.  
    204. 		// 判断线程池是否被关闭了
    205. 		if (pool->shutdown)
    206. 		{
    207. 			pthread_mutex_unlock(&pool->mutexPool);
    208. 			threadExit(pool);
    209. 		}
    210.  
    211. 		// 从任务队列中取出一个任务
    212. 		Task task;
    213. 		task.function = pool->taskQ[pool->queueFront].function;
    214. 		task.arg = pool->taskQ[pool->queueFront].arg;
    215. 		// 移动头节点
    216. 		pool->queueFront = (pool->queueFront + 1) % pool->queueCapacity;
    217. 		pool->queueSize--;
    218. 		// 解锁
    219. 		pthread_cond_signal(&pool->notFull);
    220. 		pthread_mutex_unlock(&pool->mutexPool);
    221.  
    222. 		printf("thread %ld start working ... \n", pthread_self());
    223. 		pthread_mutex_lock(&pool->mutexBusy);
    224. 		pool->busyNum++;
    225. 		pthread_mutex_unlock(&pool->mutexBusy);
    226. 		task.function(task.arg);
    227.  
    228. 		free(task.arg);
    229. 		task.arg = NULL;
    230.  
    231. 		// 结束工作
    232. 		printf("thread %ld end working ... \n", pthread_self());
    233. 		pthread_mutex_lock(&pool->mutexBusy);
    234. 		pool->busyNum--;
    235. 		pthread_mutex_unlock(&pool->mutexBusy);
    236.  
    237. 	}
    238.  
    239. 	return NULL;
    240. }
    241.  
    242. void* manager(void* arg)
    243. {
    244. 	ThreadPool* pool = (ThreadPool*)arg;
    245. 	while (!pool->shutdown)
    246. 	{
    247. 		// 每隔3s检测一次
    248. 		sleep(3);
    249.  
    250. 		// 取出线程池中任务的数量和当前线程的数量
    251. 		pthread_mutex_lock(&pool->mutexPool);
    252. 		int queueSize = pool->queueSize;
    253. 		int liveNum = pool->liveNum;
    254. 		pthread_mutex_unlock(&pool->mutexPool);
    255.  
    256. 		// 取出忙的线程的数量
    257. 		pthread_mutex_lock(&pool->mutexBusy);
    258. 		int busyNum = pool->busyNum;
    259. 		pthread_mutex_unlock(&pool->mutexBusy);
    260.  
    261. 		// 添加线程
    262. 		// 当前任务的个数>穿暖和的线程个数 && 存活的线程数<最大线程数
    263. 		if (queueSize > liveNum && liveNum < pool->maxNum)
    264. 		{
    265. 			pthread_mutex_lock(&pool->mutexPool);
    266. 			int counter = 0;
    267. 			for (int i = 0; i < pool->maxNum && counterliveNummaxNum; i++)
    268. 			{
    269. 				if (pool->threadIDs[i] == 0) {
    270. 					pthread_create(&pool->threadIDs[i], NULL, worker, pool);
    271. 					counter++;
    272. 					pool->liveNum++;
    273. 				}
    274. 			}
    275. 			pthread_mutex_unlock(&pool->mutexPool);
    276.  
    277. 		}
    278. 		// 销毁线程
    279. 		// 忙的线程*2 < 存活的线程数  &&  存活的线程 > 最小线程数
    280. 		if (busyNum*2 < liveNum && liveNum > pool->minNum)
    281. 		{
    282. 			pthread_mutex_lock(&pool->mutexPool);
    283. 			pool->exitNum = NUMBER;
    284. 			pthread_mutex_unlock(&pool->mutexPool);
    285. 			// 让工作的线程自杀
    286. 			for (int i = 0; i < NUMBER; i++)
    287. 			{
    288. 				pthread_cond_signal(&pool->notEmpty);
    289. 			}
    290. 		}
    291. 	}
    292. 	return nullptr;
    293. }
    294.  
    295. void threadExit(ThreadPool* pool)
    296. {
    297. 	pthread_t tid = pthread_self();
    298. 	for (int i = 0; i < pool->maxNum; i++)
    299. 	{
    300. 		if (pool->threadIDs[i] == tid) 
    301. 		{
    302. 			pool->threadIDs[i] = 0;
    303. 			printf("threadExit called, %ld exiting ...\n", tid);
    304. 			break;
    305. 		}
    306. 	}
    307. 	pthread_exit(NULL);
    308. }

  • 相关阅读:
    浅谈如何更好的进行需求评审
    网络安全(黑客)自学
    Chrome使用本地修改过的js替换原js内容
    linux c++ 开发 - 05- 使用CMake创建一个动态库
    VMware Ubuntu 关闭自动更新
    【无标题】
    web前端期末大作业：美食文化网页设计与实现——美食餐厅三级(HTML+CSS+JavaScript)
    决策树 #数据挖掘 #Python
    记一个带批注、表头样式的导入导出excel方法（基于easyexcel）
    Linux常用基础命令三
  • 原文地址：https://blog.csdn.net/sono_io/article/details/133951746