线程池线程池是并发编程核心的模式之一覆盖了大多数需要并发的场景这篇文章记录了对线程池的理解使用简单的语言对线程池的概念进行阐述主要包括线程池的使用场景面临的问题线程池的核心思想和如何使用c代码实现最后用一个io密集的案例进行使用示范。目录线程池目录概念场景存在问题解决方案线程池实现线程池类生成线程池提交任务销毁线程池案例生成数据数据处理任务线程池计调用结果概念目的在保证系统的稳定性下尽量提高系统的性能。思想预先创建若干线程当有任务时直接从池中取出线程执行任务任务完成后线程归还给池。方法线程池通过限制线程数量 复用线程 任务队列缓冲 future 异步回执。场景存在大量并发任务的场景1.io密集型任务如大量读写文件网络和数据库请求2.cpu密集任务如科学计算、大数据处理存在问题1.系统对线程数量存在上限无限制创建可能导致系统不稳定甚至崩溃2.线程创建过多会消耗大量内存资源存在 OOM 风险3.线程分散创建生命周期、异常与资源回收难以统一管理4.频繁创建与销毁线程系统开销大5.线程数量过多上下文切换成本高降低整体吞吐解决方案池化思想预先创建若干线程当有任务时直接从池中取出线程执行任务任务完成后线程归还给池。池统一管理线程创建和销毁线程复用线程数量可控上下文切换开销小。1.线程池限制了线程数量防止系统崩溃2.线程池限制了线程数量防止内存溢出3.线程池统一创建销毁管理简单4.线程复用减少创建/销毁成本5.控制线程数量降低上下文切换线程池实现核心流程1.生成线程池创建若干工作线程使其进入阻塞等待任务状态2.提交任务将任务封装后放入任务队列通过条件变量唤醒线程执行并返回任务回执future3.等待结果调用方通过 future 等待任务执行完成并获取结果实现要素1.线程队列存放工作线程线程死循环默认阻塞状态无任务时阻塞有任务时唤醒2.任务队列存放任务3.条件变量通信任务和线程实现线程与任务间的同步存放任务时唤醒线程4.future/promise, 接收线程执行结果异步获取执行结果future等待线程执行结果线程池类classThreadPool{private:// 线程队列std::vectorstd::threadworkers;// 任务队列std::queuestd::functionvoid()tasks;// 资源锁保护任务队列std::mutex mtx;// 条件变量std::condition_variable cv;// 管理线程池是否停止boolstop;public:// 构造函数生成线程池ThreadPool(size_t threads);// 析构函数销毁线程池~ThreadPool();// 提交任务templateclassFautoenqueue(Ff)-std::futuredecltype(f());};生成线程池输入线程数量size_t threads管理线程池停止变量stop初始化为false循环生成threads个线程线程函数为lambda表达式lambda表达式内为线程函数逻辑线程函数内定义任务task,用于接收任务队列中的任务线程函数内进行死循环线程函数内任务队列tasks为空时线程阻塞等待条件变量cv任务队列tasks非空时线程唤醒从任务队列tasks中取出一个任务执行任务线程函数内任务队列tasks为空且stop为true时线程退出循环线程结束ThreadPool::ThreadPool(size_t threads):stop(false){for(size_t i0;ithreads;i){workers.emplace_back([this]{// 任务std::functionvoid()task;while(true){{std::unique_lockstd::mutexlock(mtx);// 当任务队列为空时等待cv.wait(lock,[this]{returnstop||!tasks.empty();});// 如果线程池已经停止则退出循环if(stoptasks.empty()){return;}// 从任务队列中取出一个任务taskstd::move(tasks.front());tasks.pop();}// 执行任务task();}});}}提交任务输入任务f使用完美转发将任务f封装为std::packaged_task使用future获取任务f回执将封装的任务f放入任务队列tasks中通知条件变量cv唤醒线程执行任务返回任务回执templateclassFautoThreadPool::enqueue(Ff)-std::futuredecltype(f()){// 函数类型usingreturn_typedecltype(f());autotaskstd::make_sharedstd::packaged_taskreturn_type()(std::forwardF(f));// 获取任务的结果std::futurereturn_typerestask-get_future();{std::unique_lockstd::mutexlock(mtx);if(stop){throwstd::runtime_error(enqueue on stopped ThreadPool);}// 将任务放入队列中tasks.emplace([task](){(*task)();});}cv.notify_one();returnres;}销毁线程池销毁时触发将stop设置为true通知所有线程线程池已经停止等待所有线程结束ThreadPool::~ThreadPool(){{std::unique_lockstd::mutexlock(mtx);stoptrue;}cv.notify_all();for(std::threadworker:workers){worker.join();}}案例磁盘io密集型任务如大量读写文件存在大量文件文件中存储了大量数据需要读取文件中的数据进行计算处理每一个任务为读取一个文件处理文件中的数据生成100个文件每个文件中存储100个1000计算所有文件中数字的和结果为10000000生成数据生成100个文件每个文件中随机存储100个数据// 创建txt文件voidcreate_file(){std::vectorstd::stringfiles;std::string filenamedata/file;// 生成100个文件for(inti1;i100;i){files.push_back(filenamestd::to_string(i).txt);}for(constautofile:files){std::ofstreamf(file);//写入100行随机4位数for(inti0;i100;i){frand()%10000std::endl;}f.close();}}数据处理任务计算文件中所有数字的和// 计算文件数据和intcount_files(conststd::stringfilename){std::ifstreamfile(filename);if(!file){throwstd::runtime_error(cannot open filename);}std::string line;// 计算每行数字和size_t sum0;while(std::getline(file,line)){// 将字符串转换为数字size_t numstd::stoul(line);sumnum;}returnsum;}线程池计调用1.生成文件2.根据cpu核心数生成线程池3.遍历文件列表每个任务读取一个文件提交到线程池获取回执4.遍历回执获取每个文件的数据和5.求和intmain(){// 创建文件create_file();// 初始化线程池size_t cpu_coresstd::thread::hardware_concurrency();intthread_countscpu_cores*2;std::coutcpu cores: cpu_coresstd::endl;ThreadPoolpool(thread_counts);// 文件列表std::vectorstd::stringfiles;std::string filenamedata/file;for(inti1;i100;i){files.push_back(filenamestd::to_string(i).txt);}// 提交任务到线程池std::vectorstd::futureintfutures;for(constautofile:files){futures.emplace_back(pool.enqueue([file]{returncount_files(file);}));}// 获取结果std::vectorintresults;for(autof:futures){try{results.push_back(f.get());}catch(conststd::exceptione){std::cerre.what()std::endl;}}// 求和intdata_sum0;for(constautoresult:results){std::coutfilename: \n;std::coutsum: resultstd::endl;data_sumresult;}std::coutdata sum: data_sumstd::endl;return0;}结果sum: 100000 data/file: sum: 100000 data/file: sum: 100000 data/file: sum: 100000 data/file: sum: 100000 data/file: sum: 100000 data/file: sum: 100000 data sum: 10000000结果和预期一致为1000×100×10010000000
并发编程（c++）——2.线程池模式
发布时间：2026/6/12 10:59:10
线程池线程池是并发编程核心的模式之一覆盖了大多数需要并发的场景这篇文章记录了对线程池的理解使用简单的语言对线程池的概念进行阐述主要包括线程池的使用场景面临的问题线程池的核心思想和如何使用c代码实现最后用一个io密集的案例进行使用示范。目录线程池目录概念场景存在问题解决方案线程池实现线程池类生成线程池提交任务销毁线程池案例生成数据数据处理任务线程池计调用结果概念目的在保证系统的稳定性下尽量提高系统的性能。思想预先创建若干线程当有任务时直接从池中取出线程执行任务任务完成后线程归还给池。方法线程池通过限制线程数量 复用线程 任务队列缓冲 future 异步回执。场景存在大量并发任务的场景1.io密集型任务如大量读写文件网络和数据库请求2.cpu密集任务如科学计算、大数据处理存在问题1.系统对线程数量存在上限无限制创建可能导致系统不稳定甚至崩溃2.线程创建过多会消耗大量内存资源存在 OOM 风险3.线程分散创建生命周期、异常与资源回收难以统一管理4.频繁创建与销毁线程系统开销大5.线程数量过多上下文切换成本高降低整体吞吐解决方案池化思想预先创建若干线程当有任务时直接从池中取出线程执行任务任务完成后线程归还给池。池统一管理线程创建和销毁线程复用线程数量可控上下文切换开销小。1.线程池限制了线程数量防止系统崩溃2.线程池限制了线程数量防止内存溢出3.线程池统一创建销毁管理简单4.线程复用减少创建/销毁成本5.控制线程数量降低上下文切换线程池实现核心流程1.生成线程池创建若干工作线程使其进入阻塞等待任务状态2.提交任务将任务封装后放入任务队列通过条件变量唤醒线程执行并返回任务回执future3.等待结果调用方通过 future 等待任务执行完成并获取结果实现要素1.线程队列存放工作线程线程死循环默认阻塞状态无任务时阻塞有任务时唤醒2.任务队列存放任务3.条件变量通信任务和线程实现线程与任务间的同步存放任务时唤醒线程4.future/promise, 接收线程执行结果异步获取执行结果future等待线程执行结果线程池类classThreadPool{private:// 线程队列std::vectorstd::threadworkers;// 任务队列std::queuestd::functionvoid()tasks;// 资源锁保护任务队列std::mutex mtx;// 条件变量std::condition_variable cv;// 管理线程池是否停止boolstop;public:// 构造函数生成线程池ThreadPool(size_t threads);// 析构函数销毁线程池~ThreadPool();// 提交任务templateclassFautoenqueue(Ff)-std::futuredecltype(f());};生成线程池输入线程数量size_t threads管理线程池停止变量stop初始化为false循环生成threads个线程线程函数为lambda表达式lambda表达式内为线程函数逻辑线程函数内定义任务task,用于接收任务队列中的任务线程函数内进行死循环线程函数内任务队列tasks为空时线程阻塞等待条件变量cv任务队列tasks非空时线程唤醒从任务队列tasks中取出一个任务执行任务线程函数内任务队列tasks为空且stop为true时线程退出循环线程结束ThreadPool::ThreadPool(size_t threads):stop(false){for(size_t i0;ithreads;i){workers.emplace_back([this]{// 任务std::functionvoid()task;while(true){{std::unique_lockstd::mutexlock(mtx);// 当任务队列为空时等待cv.wait(lock,[this]{returnstop||!tasks.empty();});// 如果线程池已经停止则退出循环if(stoptasks.empty()){return;}// 从任务队列中取出一个任务taskstd::move(tasks.front());tasks.pop();}// 执行任务task();}});}}提交任务输入任务f使用完美转发将任务f封装为std::packaged_task使用future获取任务f回执将封装的任务f放入任务队列tasks中通知条件变量cv唤醒线程执行任务返回任务回执templateclassFautoThreadPool::enqueue(Ff)-std::futuredecltype(f()){// 函数类型usingreturn_typedecltype(f());autotaskstd::make_sharedstd::packaged_taskreturn_type()(std::forwardF(f));// 获取任务的结果std::futurereturn_typerestask-get_future();{std::unique_lockstd::mutexlock(mtx);if(stop){throwstd::runtime_error(enqueue on stopped ThreadPool);}// 将任务放入队列中tasks.emplace([task](){(*task)();});}cv.notify_one();returnres;}销毁线程池销毁时触发将stop设置为true通知所有线程线程池已经停止等待所有线程结束ThreadPool::~ThreadPool(){{std::unique_lockstd::mutexlock(mtx);stoptrue;}cv.notify_all();for(std::threadworker:workers){worker.join();}}案例磁盘io密集型任务如大量读写文件存在大量文件文件中存储了大量数据需要读取文件中的数据进行计算处理每一个任务为读取一个文件处理文件中的数据生成100个文件每个文件中存储100个1000计算所有文件中数字的和结果为10000000生成数据生成100个文件每个文件中随机存储100个数据// 创建txt文件voidcreate_file(){std::vectorstd::stringfiles;std::string filenamedata/file;// 生成100个文件for(inti1;i100;i){files.push_back(filenamestd::to_string(i).txt);}for(constautofile:files){std::ofstreamf(file);//写入100行随机4位数for(inti0;i100;i){frand()%10000std::endl;}f.close();}}数据处理任务计算文件中所有数字的和// 计算文件数据和intcount_files(conststd::stringfilename){std::ifstreamfile(filename);if(!file){throwstd::runtime_error(cannot open filename);}std::string line;// 计算每行数字和size_t sum0;while(std::getline(file,line)){// 将字符串转换为数字size_t numstd::stoul(line);sumnum;}returnsum;}线程池计调用1.生成文件2.根据cpu核心数生成线程池3.遍历文件列表每个任务读取一个文件提交到线程池获取回执4.遍历回执获取每个文件的数据和5.求和intmain(){// 创建文件create_file();// 初始化线程池size_t cpu_coresstd::thread::hardware_concurrency();intthread_countscpu_cores*2;std::coutcpu cores: cpu_coresstd::endl;ThreadPoolpool(thread_counts);// 文件列表std::vectorstd::stringfiles;std::string filenamedata/file;for(inti1;i100;i){files.push_back(filenamestd::to_string(i).txt);}// 提交任务到线程池std::vectorstd::futureintfutures;for(constautofile:files){futures.emplace_back(pool.enqueue([file]{returncount_files(file);}));}// 获取结果std::vectorintresults;for(autof:futures){try{results.push_back(f.get());}catch(conststd::exceptione){std::cerre.what()std::endl;}}// 求和intdata_sum0;for(constautoresult:results){std::coutfilename: \n;std::coutsum: resultstd::endl;data_sumresult;}std::coutdata sum: data_sumstd::endl;return0;}结果sum: 100000 data/file: sum: 100000 data/file: sum: 100000 data/file: sum: 100000 data/file: sum: 100000 data/file: sum: 100000 data/file: sum: 100000 data sum: 10000000结果和预期一致为1000×100×10010000000
