当前位置：首页 > ds >正文

Linux多线程---线程池实现

ds 2025/7/20 1:33:02

线程池的定义

线程池（Thread Pool）是一种多线程处理模式，用于管理和复用线程资源，避免频繁创建和销毁线程带来的性能开销。其核心思想是预先创建一定数量的线程，当有任务提交时，从线程池中获取可用线程执行任务，任务完成后线程不销毁而是返回线程池等待下一个任务。

看完定义，你可能会觉得线程池和进程池好像很像，他们都是管理并发执行单元的技术，但是两者在实现机制，资源消耗和使用场景方面有很大差异。

首先，线程池内的线程是共享进程地址空间的，线程切换开销很小，而进程池内的进程都有自己的独立空间，切换开销很大。

同时，由于线程间共享进程地址空间，所以线程间通信就无需借助IPC，但是需要锁和条件变量控制线程之间的互斥和同步，线程池内的线程资源隔离性也因此比较差，一个线程崩溃会影响整个进程。

线程池适用场景

I/O 密集型任务（如网络请求、文件读写）：线程阻塞时不影响其他线程执行。
需要共享内存的场景（如缓存数据共享）。
任务执行时间短、频繁创建销毁线程的场景（如 Web 服务器请求处理）。

线程池的实现

主线程将任务push进入任务队列，其他多线程用于从任务队列获取任务并执行。

代码：

首先我们需要一个ThreadPool类来管理线程池就像进程池有ProcessPool一样。

先描述再组织，线程池中有线程数量、多线程、任务队列等基本内容，任务队列通过queue管理，线程间互斥和同步需要锁和条件变量。

#pragma once#include <vector>
#include <queue>
#include <pthread.h>
#include <string>#define NUM 5//默认线程池内线程数template<typename T>
class ThreadPool
{
public:ThreadPool(int num=NUM):_threadnum(num){pthread_mutex_init(&_mutex,nullptr);pthread_cond_init(&cond,nullptr);}~ThreadPool(){pthread_mutex_destroy(&_mutex);pthread_cond_destroy(&cond);}private:int _threadnum;queue<T> _tasklist;pthread_mutex_t _mutex;//锁和条件变量保证线程间的互斥同步pthread_cond_t _cond;};

完成基础定义，后面还需要完成：

线程池的初始化
线程执行的函数(执行任务)
任务列表push

ThreadPool.hpp:

#pragma once#include <vector>
#include<iostream>
#include <queue>
#include <pthread.h>
#include <string>#define NUM 5 // 默认线程池内线程数template <typename T>
class ThreadPool
{
private:void Lock(){pthread_mutex_lock(&_mutex);}void Unlock(){pthread_mutex_unlock(&_mutex);}void Wait(){pthread_cond_wait(&_cond, &_mutex);}void Wakeup(){pthread_cond_signal(&_cond);}public:ThreadPool(int num = NUM) : _threadnum(num){pthread_mutex_init(&_mutex, nullptr);pthread_cond_init(&_cond, nullptr);}static void *threadroutine(void *arg){pthread_detach(pthread_self());//线程分离无需joinThreadPool *self = static_cast<ThreadPool *>(arg);while (true){self->Lock();while (self->_tasklist.empty())//任务列表空则等待{self->Wait();}T task = self->_tasklist.front(); self->_tasklist.pop();self->Unlock();  task();}}void InitThreadpool(){for (int i = 0; i < _threadnum; i++){pthread_t tid;pthread_create(&tid, nullptr, threadroutine, this);std::cout<<"thread create success..."<<std::endl;}}void push(const T& task){Lock();_tasklist.push(task);Wakeup();//唤醒条件变量通知进程可以获取任务了Unlock();}~ThreadPool(){pthread_mutex_destroy(&_mutex);pthread_cond_destroy(&_cond);}private:int _threadnum;std::queue<T> _tasklist;pthread_mutex_t _mutex; // 锁和条件变量保证线程间的互斥同步pthread_cond_t _cond;
};

为什么线程中需要锁和条件变量？

为了控制线程间的互斥和同步，线程中的任务队列是会被多给线程访问的临界资源，为了保护临界资源我们使用锁。

线程从任务队列中取任务的前提条件是任务队列中有任务，如果执行任务的线程不停的拿锁访问任务队列，同时任务队列中又没有任务可执行，会导致主线程无法拿到锁无法放置任务进而导致效率低下。所以任务队列中没有任务的时候，设置条件变量让线程阻塞并且把拿到的锁释放出来，等待任务队列中有任务出现，然后被唤醒。

注意：

当某线程被唤醒时，其可能是被异常或是伪唤醒，或者是一些广播类的唤醒线程操作而导致所有线程被唤醒，使得在被唤醒的若干线程中，只有个别线程能拿到任务。此时应该让被唤醒的线程再次判断是否满足被唤醒条件，所以在判断任务队列是否为空时，应该使用while进行判断，而不是if。
任务的执行是属于每一个线程自己的，所以不需要用锁保护起来，并且任务的执行时间如果很长会耽误其他线程拿到任务，导致线程池不是并发的。

为什么线程池中的线程执行例程threadroutine需要设置为静态方法？

因为threadroutine方法有必须的格式，返回void*类型参数类型也只能是void*，然后类方法会自动带上this指针，导致不符合格式，static修饰的类方法不会带this指针，但是我们要用到this指针怎么办呢，那就再pthread _create的时候将this指针传入进去，使用的时候强转就行。

任务类型的设计

我们将线程池进行了模板化，因此线程池当中存储的任务类型可以是任意的，但无论该任务是什么类型的，在该任务类当中都必须包含一个Run方法，当我们处理该类型的任务时只需调用该Run方法即可。

task.hpp:

#pragma once#include <iostream>//任务类
class Task
{
public:Task(int x = 0, int y = 0, char op = 0): _x(x), _y(y), _op(op){}~Task(){}//处理任务的方法void Run(){int result = 0;switch (_op){case '+':result = _x + _y;break;case '-':result = _x - _y;break;case '*':result = _x * _y;break;case '/':if (_y == 0){std::cerr << "Error: div zero!" << std::endl;return;}else{result = _x / _y;}break;case '%':if (_y == 0){std::cerr << "Error: mod zero!" << std::endl;return;}else{result = _x % _y;}break;default:std::cerr << "operation error!" << std::endl;return;}std::cout << "thread[" << pthread_self() << "]:" << _x << _op << _y << "=" << result << std::endl;}void operator()(){Run();}private:int _x;int _y;char _op;
};

主线程的作用就是创建好线程池后不断地像任务队列中push任务。

main.cc:

#include<iostream>
#include"ThreadPool.hpp"
#include"task.hpp"
#include<unistd.h>int main()
{ThreadPool<Task>* tp=new ThreadPool<Task>(5);tp->InitThreadpool();const char* op = "+-*/%";while(true){int a=rand()%10+1;int b=rand()%5+2;int index = rand() % 5;Task t(a,b,op[index]);tp->push(t);sleep(1);}return 0;
}

执行结果：