完整代码:threadpool.hpp
部分核心代码如下:
// 提交一个任务
// 调用.get()获取返回值会等待任务执行完,获取返回值
// 有两种方法可以实现调用类成员,
// 一种是使用 bind: .commit(std::bind(&Dog::sayHello, &dog));
// 一种是用 mem_fn: .commit(std::mem_fn(&Dog::sayHello), this)
template<class F, class... Args>
auto commit(F&& f, Args&&... args) ->future<decltype(f(args...))>
{
if (!_run)
throw runtime_error("commit on ThreadPool is stopped.");
using RetType = decltype(f(args...)); // typename std::result_of<F(Args...)>::type, 函数 f 的返回值类型
auto task = make_shared<packaged_task<RetType()>>(
bind(forward<F>(f), forward<Args>(args)...)
); // 把函数入口及参数,打包(绑定)
future<RetType> future = task->get_future();
{ // 添加任务到队列
lock_guard<mutex> lock{ _lock };//对当前块的语句加锁 lock_guard 是 mutex 的 stack 封装类,构造的时候 lock(),析构的时候 unlock()
_tasks.emplace([task](){ // push(Task{...}) 放到队列后面
(*task)();
});
}
#ifdef THREADPOOL_AUTO_GROW
if (_idlThrNum < 1 && _pool.size() < THREADPOOL_MAX_NUM)
addThread(1);
#endif // !THREADPOOL_AUTO_GROW
_task_cv.notify_one(); // 唤醒一个线程执行
return future;
}
//......
// 执行任务
while (_run)
{
Task task; // 获取一个待执行的 task
{
// unique_lock 相比 lock_guard 的好处是:可以随时 unlock() 和 lock()
unique_lock<mutex> lock{ _lock };
_task_cv.wait(lock, [this]{
return !_run || !_tasks.empty();
}); // wait 直到有 task
if (!_run && _tasks.empty())
return;
task = move(_tasks.front()); // 按先进先出从队列取一个 task
_tasks.pop();
}
_idlThrNum--;
task();//执行任务
_idlThrNum++;
}
使用条件变量,条件等待直到有新任务加入才加锁取任务执行。任务都是无状态的,随机分配到某个线程执行。在实际应用中,有些任务是有状态的,带有上下文,需要调度到指定的线程去处理,可以使用第二种上下文隔离的线程池模型,带上下文的好处是可以业务隔离,取代加锁,即无锁线程池。2、上下文隔离的无锁线程池(isolated_threadpool)
完整代码:isolated_threadpool.hpp
核心代码如下:
// 提交一个任务,可以指定线程执行,否则自动分配
void commit(Task task, int thread_id = -1) {
if (thread_id > (int)threads_.size() - 1) {
cout << "Invalid thread id:" << thread_id << endl;
return;
}
if (thread_id == -1)
thread_id = AllocateThread();
threads_[thread_id]->Schedule(task);
g_task_queue_++;
}
每个线程新建一个上下文,使用一个thread_local变量分别存储,当任务在该线程上执行的时候,可以读取这个线程局部变量,获取该线程上下文的数据,避免加锁。有什么不明白的地方,欢迎留言讨论~