Fetch the repository succeeded.
#include "Log.h"
#include "ThreadPool.h"
#include "Utils.h"
ThreadPool::ThreadPool(size_t threadNum, ShutdownMode shutdown_mode, size_t maxQueueSize)
: threadNum_(threadNum),
maxQueueSize_(maxQueueSize),
// 使用 类成员变量 threadpool_mutex_ 来初始化 threadpool_cond_
threadpool_cond_(threadpool_mutex_),
shutdown_mode_(shutdown_mode)
{
// 开始循环创建线程
while(threads_.size() < threadNum_)
{
pthread_t thread;
// 如果线程创建成功,则将其压入栈内存中
if(!pthread_create(&thread, nullptr, TaskForWorkerThreads_, this))
{
threads_.push_back(thread);
// // 注意这里只修改已启动的线程数量
// startedThreadNum_++;
}
}
}
ThreadPool::~ThreadPool()
{
// 向任务队列中添加退出线程事件,注意上锁
// 注意在 cond 使用之前一定要上 mutex
{
// 操作 task_queue_ 时一定要上锁
MutexLockGuard guard(threadpool_mutex_);
// 如果需要立即关闭当前的线程池,则
if(shutdown_mode_ == IMMEDIATE_SHUTDOWN)
// 先将当前队列清空
while(!task_queue_.empty())
task_queue_.pop();
// 往任务队列中添加退出线程任务
for(size_t i = 0; i < threadNum_; i++)
{
auto pthreadExit = [](void*) { pthread_exit(0); };
ThreadpoolTask task = { pthreadExit, nullptr };
task_queue_.push(task);
}
// 唤醒所有线程以执行退出操作
threadpool_cond_.notifyAll();
}
for(size_t i = 0; i < threadNum_; i++)
{
// 回收线程资源
pthread_join(threads_[i], nullptr);
}
}
bool ThreadPool::appendTask(void (*function)(void*), void* arguments)
{
// 由于会操作事件队列,因此需要上锁
MutexLockGuard guard(threadpool_mutex_);
// 如果队列长度过长,则将当前task丢弃
if(task_queue_.size() > maxQueueSize_)
return false;
else
{
// 添加task至列表中
ThreadpoolTask task = { function, arguments };
task_queue_.push(task);
// 每当有新事件进入之时,只唤醒一个等待线程
threadpool_cond_.notify();
return true;
}
}
void* ThreadPool::TaskForWorkerThreads_(void* arg)
{
ThreadPool* pool = (ThreadPool*)arg;
// 启动当前线程
ThreadpoolTask task;
// 对于子线程来说,事件循环开始
for(;;)
{
// 首先获取事件
{
// 获取事件时需要上个锁
MutexLockGuard guard(pool->threadpool_mutex_);
/**
* 如果好不容易获得到锁了,但是没有事件可以执行
* 则陷入沉睡,释放锁,并等待唤醒
* NOTE: 注意, pthread_cond_signal 会唤醒至少一个线程
* 也就是说,可能存在被唤醒的线程仍然没有事件处理的情况
* 这时只需循环wait即可.
*/
while(pool->task_queue_.size() == 0)
pool->threadpool_cond_.wait();
// 唤醒后一定有事件
assert(pool->task_queue_.size() != 0);
task = pool->task_queue_.front();
pool->task_queue_.pop();
}
// 执行事件
(task.function)(task.arguments);
}
// 注意: UNREACHABLE, 控制流不可能会到达此处
// 因为线程的退出不会走这条控制流,而是执行退出事件
UNREACHABLE();
return nullptr;
}
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。