- 当前位置：首页 >应用开发 >我们一起手撸一个线程池
我们一起手撸一个线程池
发布时间：2025-11-04 00:11:41 来源：创站工坊 作者：IT科技类资讯
本文转载自微信公众号「程序喵大人」，手撸作者程序喵大人 。个线转载本文请联系程序喵大人公众号。程池
之前分享过一次手写线程池 - C语言版，手撸然后有朋友问是个线否有C++线程池实现的文章：
其实关于C++线程池的文章我好久以前写过，但估计很多新朋友都没有看到过，程池这里也重新发一下!
本人在开发过程中经常会遇到需要使用线程池的手撸需求，但查了一圈发现在C++中完备的个线线程池第三方库还是比较少的，于是程池打算自己搞一个，链接地址文章最后附上，手撸目前还只是个线初版，可能还有很多问题，程池望各位指正。手撸
线程池都需要什么功能?个线
个人认为线程池需要支持以下几个基本功能：
核心线程数(core_threads)：线程池中拥有的最少线程个数，初始化时就会创建好的程池线程，常驻于线程池。 最大线程个数(max_threads)：线程池中拥有的最大线程个数，max_threads>=core_threads，当任务的个数太多线程池执行不过来时，内部就会创建更多的线程用于执行更多的b2b信息网任务，内部线程数不会超过max_threads，多创建出来的线程在一段时间内没有执行任务则会自动被回收掉，最终线程个数保持在核心线程数。 超时时间(time_out)：如上所述，多创建出来的线程在time_out时间内没有执行任务就会被回收。 可获取当前线程池中线程的总个数。 可获取当前线程池中空闲线程的个数。 开启线程池功能的开关。 关闭线程池功能的开关，可以选择是否立即关闭，立即关闭线程池时，当前线程池里缓存的任务不会被执行。如何实现线程池?下面是自己实现的线程池逻辑。
线程池中主要的数据结构
1. 链表或者数组：用于存储线程池中的线程。
2. 队列：用于存储需要放入线程池中执行的任务。
3. 条件变量：当有任务需要执行时，用于通知正在等待的线程从任务队列中取出任务执行。
代码如下：
class ThreadPool { public: using PoolSeconds = std::chrono::seconds; /** 线程池的香港云服务器配置 * core_threads: 核心线程个数，线程池中最少拥有的线程个数，初始化就会创建好的线程，常驻于线程池 * * max_threads: >=core_threads，当任务的个数太多线程池执行不过来时， * 内部就会创建更多的线程用于执行更多的任务，内部线程数不会超过max_threads * * max_task_size: 内部允许存储的最大任务个数，暂时没有使用 * * time_out: Cache线程的超时时间，Cache线程指的是max_threads-core_threads的线程, * 当time_out时间内没有执行任务，此线程就会被自动回收 */ struct ThreadPoolConfig { int core_threads; int max_threads; int max_task_size; PoolSeconds time_out; }; /** * 线程的状态：有等待、运行、停止 */ enum class ThreadState { kInit = 0, kWaiting = 1, kRunning = 2, kStop = 3 }; /** * 线程的种类标识：标志该线程是核心线程还是Cache线程，Cache是内部为了执行更多任务临时创建出来的 */ enum class ThreadFlag { kInit = 0, kCore = 1, kCache = 2 }; using ThreadPtr = std::shared_ptr<std::thread>; using ThreadId = std::atomic<int>; using ThreadStateAtomic = std::atomic<ThreadState>; using ThreadFlagAtomic = std::atomic<ThreadFlag>; /** * 线程池中线程存在的基本单位，每个线程都有个自定义的ID，有线程种类标识和状态 */ struct ThreadWrapper { ThreadPtr ptr; ThreadId id; ThreadFlagAtomic flag; ThreadStateAtomic state; ThreadWrapper() { ptr = nullptr; id = 0; state.store(ThreadState::kInit); } }; using ThreadWrapperPtr = std::shared_ptr<ThreadWrapper>; using ThreadPoolLock = std::unique_lock<std::mutex>; private: ThreadPoolConfig config_; std::list<ThreadWrapperPtr> worker_threads_; std::queue<std::function<void()>> tasks_; std::mutex task_mutex_; std::condition_variable task_cv_; std::atomic<int> total_function_num_; std::atomic<int> waiting_thread_num_; std::atomic<int> thread_id_; // 用于为新创建的云南idc服务商线程分配ID std::atomic<bool> is_shutdown_now_; std::atomic<bool> is_shutdown_; std::atomic<bool> is_available_; };线程池的初始化
在构造函数中将各个成员变量都附初值，同时判断线程池的config是否合法。
ThreadPool(ThreadPoolConfig config) : config_(config) { this->total_function_num_.store(0); this->waiting_thread_num_.store(0); this->thread_id_.store(0); this->is_shutdown_.store(false); this->is_shutdown_now_.store(false); if (IsValidConfig(config_)) { is_available_.store(true); } else { is_available_.store(false); } } bool IsValidConfig(ThreadPoolConfig config) { if (config.core_threads < 1 || config.max_threads < config.core_threads || config.time_out.count() < 1) { return false; } return true; }如何开启线程池功能?
创建核心线程数个线程，常驻于线程池，等待任务的执行，线程ID由GetNextThreadId()统一分配。
// 开启线程池功能 bool Start() { if (!IsAvailable()) { return false; } int core_thread_num = config_.core_threads; cout << "Init thread num " << core_thread_num << endl; while (core_thread_num-- > 0) { AddThread(GetNextThreadId()); } cout << "Init thread end" << endl; return true; }如何关闭线程?
这里有两个标志位，is_shutdown_now置为true表示立即关闭线程，is_shutdown置为true则表示先执行完队列里的任务再关闭线程池。
// 关掉线程池，内部还没有执行的任务会继续执行 void ShutDown() { ShutDown(false); cout << "shutdown" << endl; } // 执行关掉线程池，内部还没有执行的任务直接取消，不会再执行 void ShutDownNow() { ShutDown(true); cout << "shutdown now" << endl; } // private void ShutDown(bool is_now) { if (is_available_.load()) { if (is_now) { this->is_shutdown_now_.store(true); } else { this->is_shutdown_.store(true); } this->task_cv_.notify_all(); is_available_.store(false); } }如何为线程池添加线程?
见AddThread()函数，默认会创建Core线程，也可以选择创建Cache线程，线程内部会有一个死循环，不停的等待任务，有任务到来时就会执行，同时内部会判断是否是Cache线程，如果是Cache线程，timeout时间内没有任务执行就会自动退出循环，线程结束。
这里还会检查is_shutdown和is_shutdown_now标志，根据两个标志位是否为true来判断是否结束线程。
void AddThread(int id) { AddThread(id, ThreadFlag::kCore); } void AddThread(int id, ThreadFlag thread_flag) { cout << "AddThread " << id << " flag " << static_cast<int>(thread_flag) << endl; ThreadWrapperPtr thread_ptr = std::make_shared<ThreadWrapper>(); thread_ptr->id.store(id); thread_ptr->flag.store(thread_flag); auto func = [this, thread_ptr]() { for (;;) { std::function<void()> task; { ThreadPoolLock lock(this->task_mutex_); if (thread_ptr->state.load() == ThreadState::kStop) { break; } cout << "thread id " << thread_ptr->id.load() << " running start" << endl; thread_ptr->state.store(ThreadState::kWaiting); ++this->waiting_thread_num_; bool is_timeout = false; if (thread_ptr->flag.load() == ThreadFlag::kCore) { this->task_cv_.wait(lock, [this, thread_ptr] { return (this->is_shutdown_ || this->is_shutdown_now_ || !this->tasks_.empty() || thread_ptr->state.load() == ThreadState::kStop); }); } else { this->task_cv_.wait_for(lock, this->config_.time_out, [this, thread_ptr] { return (this->is_shutdown_ || this->is_shutdown_now_ || !this->tasks_.empty() || thread_ptr->state.load() == ThreadState::kStop); }); is_timeout = !(this->is_shutdown_ || this->is_shutdown_now_ || !this->tasks_.empty() || thread_ptr->state.load() == ThreadState::kStop); } --this->waiting_thread_num_; cout << "thread id " << thread_ptr->id.load() << " running wait end" << endl; if (is_timeout) { thread_ptr->state.store(ThreadState::kStop); } if (thread_ptr->state.load() == ThreadState::kStop) { cout << "thread id " << thread_ptr->id.load() << " state stop" << endl; break; } if (this->is_shutdown_ && this->tasks_.empty()) { cout << "thread id " << thread_ptr->id.load() << " shutdown" << endl; break; } if (this->is_shutdown_now_) { cout << "thread id " << thread_ptr->id.load() << " shutdown now" << endl; break; } thread_ptr->state.store(ThreadState::kRunning); task = std::move(this->tasks_.front()); this->tasks_.pop(); } task(); } cout << "thread id " << thread_ptr->id.load() << " running end" << endl; }; thread_ptr->ptr = std::make_shared<std::thread>(std::move(func)); if (thread_ptr->ptr->joinable()) { thread_ptr->ptr->detach(); } this->worker_threads_.emplace_back(std::move(thread_ptr)); }如何将任务放入线程池中执行?
见如下代码，将任务使用std::bind封装成std::function放入任务队列中，任务较多时内部还会判断是否有空闲线程，如果没有空闲线程，会自动创建出最多(max_threads-core_threads)个Cache线程用于执行任务。
// 放在线程池中执行函数 template <typename F, typename... Args> auto Run(F &&f, Args &&... args) -> std::shared_ptr<std::future<std::result_of_t<F(Args...)>>> { if (this->is_shutdown_.load() || this->is_shutdown_now_.load() || !IsAvailable()) { return nullptr; } if (GetWaitingThreadSize() == 0 && GetTotalThreadSize() < config_.max_threads) { AddThread(GetNextThreadId(), ThreadFlag::kCache); } using return_type = std::result_of_t<F(Args...)>; auto task = std::make_shared<std::packaged_task<return_type()>>( std::bind(std::forward<F>(f), std::forward<Args>(args)...)); total_function_num_++; std::future<return_type> res = task->get_future(); { ThreadPoolLock lock(this->task_mutex_); this->tasks_.emplace([task]() { (*task)(); }); } this->task_cv_.notify_one(); return std::make_shared<std::future<std::result_of_t<F(Args...)>>>(std::move(res)); }如何获取当前线程池中线程的总个数?
int GetTotalThreadSize() { return this->worker_threads_.size(); }如何获取当前线程池中空闲线程的个数?
waiting_thread_num值表示空闲线程的个数，该变量在线程循环内部会更新。
int GetWaitingThreadSize() { return this->waiting_thread_num_.load(); }简单的测试代码
int main() { cout << "hello" << endl; ThreadPool pool(ThreadPool::ThreadPoolConfig{4, 5, 6, std::chrono::seconds(4)}); pool.Start(); std::this_thread::sleep_for(std::chrono::seconds(4)); cout << "thread size " << pool.GetTotalThreadSize() << endl; std::atomic<int> index; index.store(0); std::thread t([&]() { for (int i = 0; i < 10; ++i) { pool.Run([&]() { cout << "function " << index.load() << endl; std::this_thread::sleep_for(std::chrono::seconds(4)); index++; }); // std::this_thread::sleep_for(std::chrono::seconds(2)); } }); t.detach(); cout << "=================" << endl; std::this_thread::sleep_for(std::chrono::seconds(4)); pool.Reset(ThreadPool::ThreadPoolConfig{4, 4, 6, std::chrono::seconds(4)}); std::this_thread::sleep_for(std::chrono::seconds(4)); cout << "thread size " << pool.GetTotalThreadSize() << endl; cout << "waiting size " << pool.GetWaitingThreadSize() << endl; cout << "---------------" << endl; pool.ShutDownNow(); getchar(); cout << "world" << endl; return 0; }香港云服务器IT技术网服务器租用企商汇源码库源码下载亿华云IT资讯网益华科技运维纵横极客码头智能时代亿华智造益强编程堂益强数据堂编程之道思维库益强编程舍益华IT技术论坛益强资讯优选亿华智慧云全栈开发亿华云亿华科技益华科技码上建站汇智坊技术快报益强智未来益强科技码力社IT资讯网亿华灵动创站工坊亿华互联益强前沿资讯益强智囊团亿华云计算多维IT资讯云站无忧极客编程
- Copyright © 2025 Powered by 我们一起手撸一个线程池,创站工坊 滇ICP备2023006006号-44sitemap