Linux —— 线程池

目录

1.认识线程池

什么是线程池

线程池的优点

线程池的使用场景

2.线程池的实现

线程池的组成

线程池之间的关系 

线程池的实现代码

任务类型

线程池类型

主程序

---

1.认识线程池

什么是线程池

线程池（Thread Pool）是线程的一种使用方式。它通过预先创建一组线程并管理它们的生命周期，来提高多线程应用程序的性能和资源利用率。线程池的核心思想是减少在创建和销毁线程时所产生的开销。

线程池的优点

• 减少开销：线程的创建和销毁是有开销的，线程池通过复用线程减少了这种开销。

• 提高响应速度：当任务到达时，线程池中已经有现成的线程可以立即执行任务，而不需要等待线程创建。

• 提高线程的可控性：线程池可以统一管理线程的生命周期、数量和优先级，使得多线程编程更加可控。

• 防止资源耗尽：通过限制线程的数量，线程池可以防止系统因创建过多线程而耗尽资源。

线程池的使用场景

1. Web 服务器处理请求：Web 服务器需要同时处理大量客户端请求，每个请求通常是一个短期任务（如 HTTP 请求）。使用线程池可以避免为每个请求创建和销毁线程的开销，提高服务器的响应速度和资源利用率。

2. 数据库连接池：数据库操作通常是 I/O 密集型任务，频繁创建和销毁数据库连接会消耗大量资源。通过线程池管理数据库连接，可以复用连接，减少创建和销毁连接的开销。

3. 异步任务处理：某些任务不需要立即执行，可以异步处理（如日志记录、邮件发送、消息推送等）。将任务提交到线程池中异步执行，避免阻塞主线程。

4. 批量数据处理：需要对大量数据进行处理（如文件解析、数据清洗、批量计算等），这些任务可以并行执行。将数据分片后提交到线程池中并行处理，提高处理效率。

关于线程池的应用还有很多，笔者就不一 一赘述了。

2.线程池的实现

线程池的组成

线程池内部主要有一个任务队列用于存储任务，还有一批线程从任务队列中获取任务并执行。任务队列中的任务通过一个控制线程传递进来，并且控制线程还要管理好线程池的生命周期，但是控制线程并不属于线程池。

线程池之间的关系 

线程池代码中的角色有主线程和线程池中的从线程，主线程只有一个，从线程有多个，主线程往任务队列中放数据，从线程从任务队列中获取数据；此时的任务队列就是主线程和从线程之间的共享资源（也是临界资源）。因此，我们要讨论线程池中各角色之间的关系，只有讨论清楚关系了才能写出正确的并且健壮的代码。

• 主线程和主线程之间 —— 主线程只有一个，不讨论。
• 主线程和从线程之间 —— 互斥和同步：主线程在放数据的时候，从线程不能来获取数据，否则获取到的数据可能不完整，他们之间需要互斥的访问临界资源。而主线程生产数据之后，需要通知从线程来拿，任务队列中没有数据的时候，从线程需要进行排队等待，因此，他们之间具有同步关系。
• 从线程和从线程之间 —— 互斥和同步：一个线程在拿数据，另一个线程就不能拿，否则可能拿到一样的数据，因此，他们之间需要互斥。为了避免某些进程竞争锁的能力太强导致的线程饥饿问题，当任务队列为空的时候，多个线程需要在条件变量下排队等待，因此，他们之间需要保持同步。

线程池的实现代码

我们设计的代码需要一个任务类型、一个线程池类型、一个主程序即可。

任务类型

我们的任务类型只是简单的做一下加法即可，有想法的读者可以设计自己的任务类型。

1. #include <iostream>
2. #include <string>
4. class Task
5. {
6. public:
7. Task() {}
8. Task(int num1, int num2) : _num1(num1), _num2(num2), _result(0)
9. {
10. }
11. void Excute()
12. {
13. _result = _num1 + _num2;
14. }
15. void ResultToString() // 打印计算结果
16. {
17. std::cout << "Excute Task: " << std::to_string(_num1) + "+" + std::to_string(_num2) + "=" + std::to_string(_result) << std::endl;
18. }
19. std::string DebugToString() // 打印生产的任务
20. {
21. return std::to_string(_num1) + "+" + std::to_string(_num2) + "=?";
22. }
23. ~Task()
24. {}
25. private:
26. int _num1;
27. int _num2;
28. int _result;
29. };

复制代码

线程池类型

1. #include <iostream>
2. #include <queue>
3. #include <pthread.h>
4. #include <unistd.h>
5. #include "task.hpp"
7. #define THREAD_NUM 5
9. template<class t="">
10. class ThreadPool
11. {
12. private:
13. int _thread_num; // 线程个数
14. std::queue<t*> _task_queue; // 任务队列
15. pthread_mutex_t _lock; // 互斥量，用于保护临界资源 —— 任务队列
16. pthread_cond_t _cond; // 条件变量，维护线程池中的线程按照
17. private:
18. void LockQueue()
19. {
20. pthread_mutex_lock(&_lock);
21. }
23. void UnLockQueue()
24. {
25. pthread_mutex_unlock(&_lock);
26. }
28. void WakeUpOne()
29. {
30. pthread_cond_signal(&_cond);
31. }
33. void ThreadWait()
34. {
35. pthread_cond_wait(&_cond, &_lock);
36. }
38. bool IsEmptyQueue()
39. {
40. return _task_queue.empty();
41. }
43. static void *handler(void *arg)
44. {
45. ThreadPool* tp_ptr = (ThreadPool*)arg;
46. while(true)
47. {
48. // 申请锁
49. tp_ptr->LockQueue();
50. // 判断任务队列是否为空 如果为空就让线程去等待
51. while(tp_ptr->IsEmptyQueue())
52. {
53. tp_ptr->ThreadWait();
54. }
55. // 代码走到这里说明任务队列不为空
57. Task *task; // 输出型参数，用于获取任务
58. tp_ptr->GetTask(&task); // 获取任务
59. tp_ptr->UnLockQueue(); // 释放锁
60. task->Excute(); // 执行任务
61. task->ResultToString(); // 打印执行结果
62. delete task; // 任务是malloc的，使用完需要delete
63. }
64. return NULL;
65. }
67. public:
68. ThreadPool(int num = THREAD_NUM)
69. :_thread_num(num)
70. {
71. // 构造时初始化互斥量和条件变量
72. pthread_mutex_init(&_lock, NULL);
73. pthread_cond_init(&_cond, NULL);
74. }
76. ~ThreadPool()
77. {
78. // 析构时销毁互斥量和条件变量
79. pthread_mutex_destroy(&_lock);
80. pthread_cond_destroy(&_cond);
81. }
83. bool InitThreadPool() {
84. pthread_t tid;
85. // 创建一批线程
86. for (int i = 0; i < _thread_num; i++)
87. {
88. int ret = pthread_create(&tid, NULL, handler, this);
89. if (ret != 0)
90. {
91. std::cout<<"create pool thread error\n";
92. return false;
93. }
94. }
96. return true;
97. }
99. void PushTask(Task *task)
100. {
101. // 加锁
102. LockQueue();
103. // 向任务队列中放数据
104. _task_queue.push(task);
105. // 通知线程池中正在等待的线程来获取任务
106. WakeUpOne();
107. // 解锁
108. UnLockQueue();
109. }
111. void GetTask(Task **task)
112. {
113. *task = _task_queue.front();
114. _task_queue.pop();
115. }
116. };

复制代码

主程序

1. #include "ThreadPool.hpp"
3. int main()
4. {
5. // 定义线程池
6. ThreadPool<task> thread_pool;
7. // 初始化线程池
8. if( thread_pool.InitThreadPool())
9. {
10. std::cout << "Create threadpool success" << std::endl;
11. }
13. srand(time(NULL));
14. while(true) {
15. // 主线程创建任务
16. int num1 = rand()%1000;
17. int num2 = rand()%1000;
18. Task* task = new Task(num1, num2);
19. // 打印创建的任务
20. std::cout << "Create Task: " << task->DebugToString() << std::endl;
21. // 向任务队列中放入任务
22. thread_pool.PushTask(task);
23. // 每个1秒创建一个线程
24. sleep(1);
25. }
27. return 0;
28. }

复制代码

运行结果：

• 可以看到主线程创建的任务被线程池中的线程获取并执行了。