大家好,我是程序员喵大人,今天我们来讲下C++线程池的实现。 本人在开发过程中经常会遇到需要使用线程池的需求,但查了一圈发现在C++中完备的线程池第三方库还是比较少的,于是打算自己搞一个,链接地址文章最后附上,目前还只是初版,可能还有很多问题,望各位指正。 个人认为线程池需要支持以下几个基本功能:
如何实现线程池?下面是自己实现的线程池逻辑。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_; // 用于为新创建的线程分配ID
std::atomic<bool> is_shutdown_now_; std::atomic<bool> is_shutdown_; std::atomic<bool> is_available_; }; 在构造函数中将各个成员变量都附初值,同时判断线程池的config是否合法。
创建核心线程数个线程,常驻于线程池,等待任务的执行,线程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; } 这里有两个标志位,isshutdown_now置为true表示立即关闭线程,isshutdown置为true则表示先执行完队列里的任务再关闭线程池。
见AddThread()函数,默认会创建Core线程,也可以选择创建Cache线程,线程内部会有一个死循环,不停的等待任务,有任务到来时就会执行,同时内部会判断是否是Cache线程,如果是Cache线程,timeout时间内没有任务执行就会自动退出循环,线程结束。 这里还会检查is_shutdown和isshutdown_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线程用于执行任务。
waitingthread_num值表示空闲线程的个数,该变量在线程循环内部会更新。 int GetWaitingThreadSize() { return this->waiting_thread_num_.load(); }
|
|
来自: 西北望msm66g9f > 《编程》