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

(线程池) 100行以内的简单线程池

news 2025/7/27 22:43:51

文章目录

  • 前言
  • Code
    • ThreadPool.hpp
    • main.cpp
  • 简单讲解
    • 所需头文件
    • using
    • 成员变量
    • 构造
    • 析构
    • 添加任务
    • PS
    • 测试效果
  • END

前言

线程池_百度百科 (baidu.com)

线程池是一种多线程处理形式，处理过程中将任务添加到队列，然后在创建线程后自动启动这些任务。线程池线程都是后台线程。每个线程都使用默认的堆栈大小，以默认的优先级运行，并处于多线程单元中。如果某个线程在托管代码中空闲（如正在等待某个事件）,则线程池将插入另一个辅助线程来使所有处理器保持繁忙。如果所有线程池线程都始终保持繁忙，但队列中包含挂起的工作，则线程池将在一段时间后创建另一个辅助线程但线程的数目永远不会超过最大值。超过最大值的线程可以排队，但他们要等到其他线程完成后才启动。

线程池主要有fixed模式和cached模式。

其中fixed模式实现起来比较简单。本文就是以此来编写。

多线程基础请看：(C++) 多线程之生产者消费者问题_c++ 多线程 生产者消费者_天赐细莲的博客-CSDN博客

cached模式的线程池，可以参考程序喵达人的代码：C++线程池的实现 - 掘金 (juejin.cn)

Code

ThreadPool.hpp

#include <atomic>
#include <condition_variable>
#include <functional>
#include <mutex>
#include <queue>
#include <sstream>
#include <string>
#include <thread>
#include <vector>namespace lotus {
std::string get_threadID() {std::stringstream ss;ss << std::this_thread::get_id();return ss.str();
}class ThreadPool {
public:// 统一将任务bind为 void(*)() 的函数形式using Task_Type = std::function<void()>;private:  // config// 后期根据这个将 fixed 模式改为 cached模式const size_t TASK_COUNT;private:  // taskstd::vector<std::unique_ptr<std::thread>> m_threadControlList;std::queue<Task_Type> m_taskQueue;private:  // thread helperstd::atomic<bool> m_canRun;std::condition_variable m_condVar;std::mutex m_mutex;public:/*** @brief Construct a new Thread Pool object* 尽量不要让线程数 > cpu内核数* @param taskCnt*/ThreadPool(const size_t taskCnt) : TASK_COUNT(taskCnt) {m_canRun = (TASK_COUNT <= std::thread::hardware_concurrency());open_pool();}// copy prohibitedThreadPool(const ThreadPool&) = delete;ThreadPool& operator=(const ThreadPool&) = delete;/*** @brief Destroy the Thread Pool object*/~ThreadPool() {close_pool();}public:/*** @brief* 添加任务，并让条件变量通知一次* 目前不处理返回值* @tparam Fun* @tparam Args* @param fun* @param args*/template <typename Fun, typename... Args>void Add_task(Fun&& fun, Args&&... args) {std::lock_guard<std::mutex> lock(m_mutex);auto task =std::bind(std::forward<Fun>(fun), std::forward<Args>(args)...);m_taskQueue.push(std::move(task));m_condVar.notify_one();}private:/*** @brief Create a thread object*/void create_thread() {std::lock_guard<std::mutex> lock(m_mutex);auto createTaskThread = [this]() {for (;;) {std::unique_lock<std::mutex> lock(m_mutex);while (m_taskQueue.empty() && m_canRun) {m_condVar.wait(lock);}if (false == m_canRun) {break;}auto task = std::move(m_taskQueue.front());m_taskQueue.pop();lock.unlock();task();}  // while 1};auto thPtr = std::make_unique<std::thread>(createTaskThread);m_threadControlList.emplace_back(std::move(thPtr));}private:/*** @brief* 创建一定数量的线程* @param taskCnt*/void open_pool() {for (size_t i = 0; i < TASK_COUNT; i += 1) {create_thread();}}/*** @brief* 运行标志改为 false* 并通知所有线程* 确保所有线程都join完*/void close_pool() {m_canRun = false;m_condVar.notify_all();for (auto&& thPtr : m_threadControlList) {if (thPtr->joinable()) {thPtr->join();}}}
};  // class}  // namespace lotu

main.cpp

#include <iostream>#include "ThreadPool.hpp"
namespace my = lotus;#ifdef _MSC_VER
#define __FUNC_NAME__ __FUNCSIG__
#elif defined(__GNUC__) || defined(__clang__)
#define __FUNC_NAME__ __PRETTY_FUNCTION__
#else
#define __FUNC_NAME__ __func__
#endif/*** @brief* test fun* @param waitTime* @param str*/
void fun(int waitTime, const char* str) {const int N = 5;printf("[%s]time{%d}start\n", __FUNC_NAME__, waitTime);for (int i = 0; i < N; i += 1) {std::this_thread::sleep_for(std::chrono::seconds(waitTime));auto threadId = my::get_threadID();const char* idStr = threadId.c_str();printf("[%s]%s\n", idStr, str);}printf("[%s]time{%d}end\n", __FUNC_NAME__, waitTime);
}/*** @brief* main fun* @param argc* @param argv* @return int*/
int main(int argc, const char** argv) {srand(time(0));printf("[%s] start\n", __FUNC_NAME__);{my::ThreadPool pool(2);pool.Add_task(fun, 1, "11111111111111111111");pool.Add_task(fun, 2, "22222222222222222222");pool.Add_task(fun, 3, "33333333333333333333");system("pause");}printf("[%s] end\n", __FUNC_NAME__);// getchar();
}

简单讲解

所需头文件

#include <atomic>
#include <condition_variable>
#include <functional>
#include <mutex>
#include <queue>
#include <sstream>
#include <string>
#include <thread>
#include <vector>

using

// 将所有任务bind成void(*)()的形式
using Task_Type = std::function<void()>;

成员变量

private:  // config// 后期根据这个将 fixed 模式改为 cached模式const size_t TASK_COUNT;private:  // taskstd::vector<std::unique_ptr<std::thread>> m_threadControlList;// 任务队列std::queue<Task_Type> m_taskQueue;private:  // thread helper// 终止标志std::atomic<bool> m_canRun;// 条件变量std::condition_variable m_condVar;// 互斥量std::mutex m_mutex;

构造

传入fixed的线程数量， 尽量不要 ＞ cpu核数。禁止拷贝操作。

public:// 传入fixed的线程数量ThreadPool(const size_t taskCnt) : TASK_COUNT(taskCnt) {// 尽量不要 ＞ cpu核数m_canRun = (TASK_COUNT <= std::thread::hardware_concurrency());open_pool();}// 禁止拷贝ThreadPool(const ThreadPool&) = delete;ThreadPool& operator=(const ThreadPool&) = delete;

初始化TASK_COUNT数量的线程。

每个线程写死循环，不断等待和执行任务队列的任务。

基于条件变量std::condition_variable来进行阻塞。

private:   void create_thread() {std::lock_guard<std::mutex> lock(m_mutex);auto createTaskThread = [this]() {for (;;) {std::unique_lock<std::mutex> lock(m_mutex);while (m_taskQueue.empty() && m_canRun) {m_condVar.wait(lock);}if (false == m_canRun) {break;}auto task = std::move(m_taskQueue.front());m_taskQueue.pop();lock.unlock();task();}  // while 1};auto thPtr = std::make_unique<std::thread>(createTaskThread);m_threadControlList.emplace_back(std::move(thPtr));}private:void open_pool() {for (size_t i = 0; i < TASK_COUNT; i += 1) {create_thread();}}

析构

public:~ThreadPool() {close_pool();}

将运行标志改为false，由于这里的标志是std::atomic<>。并且其余变量不涉及资源竞争，因此无需加锁。

注意，这里使用join()来保证每个线程是正常退出的。

private:void close_pool() {m_canRun = false;m_condVar.notify_all();for (auto&& thPtr : m_threadControlList) {if (thPtr->joinable()) {thPtr->join();}}}

添加任务

这里使用变参模板技术，将任务包装成std::function<void()>的一个可调用对象。

注意这里传入的是万能引用要使用std::forward<>。

每加入一个任务，就让条件变量进行一次通知notify_one()。

public:template <typename Fun, typename... Args>void Add_task(Fun&& fun, Args&&... args) {std::lock_guard<std::mutex> lock(m_mutex);auto task =std::bind(std::forward<Fun>(fun), std::forward<Args>(args)...);m_taskQueue.push(std::move(task));m_condVar.notify_one();}

PS

由于std::this_thread::get_id()返回的是thread::id。对于cpp可以使用std::cout来进行输出。但是并没有对应的类型转化操作。且，id的大小和数值类型在不同平台，不同编译器中不一致。更不能用于printf虽然有的编译器支持。

因此用std::stringstream流进行一次间接的转化。

std::string get_threadID() {std::stringstream ss;ss << std::this_thread::get_id();return ss.str();
}

测试效果

g++ (x86_64-posix-seh-rev0, Built by MinGW-W64 project) 7.3.0

[int main(int, const char**)] start     
[void fun(int, const char*)]time{1}start
[void fun(int, const char*)]time{2}start
请按任意键继续. . . [2]11111111111111111111
[3]22222222222222222222
[2]11111111111111111111
[2]11111111111111111111
[3]22222222222222222222
[2]11111111111111111111
[2]11111111111111111111
[void fun(int, const char*)]time{1}end
[void fun(int, const char*)]time{3}start
[3]22222222222222222222
[3]22222222222222222222
[2]33333333333333333333
[3]22222222222222222222
[void fun(int, const char*)]time{2}end
[2]33333333333333333333
[2]33333333333333333333
[2]33333333333333333333
[2]33333333333333333333
[void fun(int, const char*)]time{3}end[int main(int, const char**)] end

最后的空行是，对应system("pause")。由于在线程池的析构中使用了join()，因此提前键入也可以。

END

http://www.lryc.cn/news/146395.html

相关文章：

  • Mysql按姓氏从小到大排序的正确sql
  • 【C++】详细介绍模版初阶—函数模版、类模板
  • BananaPi BPI-6202工业控制板全志科技A40i、24V DC输入、RS485接口
  • Python - functools.partial设置回调函数处理异步任务基本使用
  • phpspreadsheet导出excel自动获得列，数字下标
  • 结算日-洛谷
  • Android Native Code开发学习（一）环境配置
  • Python GUI应用程序开发之wxPython使用详解
  • 【电子学会真题】青少年软件编程（C语言）等级考试试卷（一级） 2021年9月
  • 学习完毕JavaSE的感想
  • FastJson的学习
  • python scrapy框架
  • 滑动窗口系列3-Leetcode134题加油站
  • LOIC（low orbit ion cannon）
  • 从格灵深瞳中报稳定盈利，看AI公司的核心竞争力
  • 理解 Databend Cluster key 原理及使用
  • C++day3（类、this指针、类中的特殊成员函数）
  • Qt中的配置文件：实现个性化应用程序配置与保存加载
  • Navicat激活时出现rsa public key not find错误
  • FFmpeg5.0源码阅读——URLContext和URLProtocol
  • Qt的输出
  • 长胜证券：久违普涨再现 大盘回升有望加速
  • WPF .NET 7.0学习整理（一）
  • 数据分析简介
  • 解读未知：文本识别算法的突破与实际应用
  • [第七届蓝帽杯全国大学生网络安全技能大赛 蓝帽杯 2023]——Web方向部分题 详细Writeup
  • el-backtop返回顶部的使用
  • Go 官方标准编译器中所做的优化
  • C语言程序设计——小学生计算机辅助教学系统
  • SQL自动递增的列恢复至从0开始