Thread pool实现

2019-11-25 1351 words 3 minutes

Contents

当我们需要处理多个耗时任务时，每一个任务都不能中途暂停，此时怎么办？

大多数时候我们会考虑使用多线程来处理，因为多线程中的每一个线程都相当于是单独的任务，每个子线程都共享数据。在一个系统中，动态任务比较多的情况下，每当有一个动态任务就创建一个线程去执行它，任务执行完这个线程即销毁，这是理想状态。由于创建子线程的系统开销比较大，所以当任务量多到一定数量级后，这种“即建即跑即销毁”的机制会耗费大量的时间在创建和销毁线程上，真正执行任务的时间比较少，执行效率很低。那怎么办呢？于是有天才的程序猿想到了一个方法，既然创建和销毁线程会占用很多时间，那在启动的时候就把指定量的线程创建好不就可以了，机智如你。

thread pool 原理

如上所述，线程池的出现时为了解决创建和销毁线程很耗费资源而出现的。在启动之初，根据负载，创建指定量的线程。这时又带来了另一个问题，就是在启动之初根本不知道每个线程运行的具体任务，此时线程岂不是在空跑。在windows系统中，线程空跑是会把CPU占用提升到100%的。至此任务队列就出现了，它解决了任务的创建和运行不在同一时刻的问题；但还有一个问题，没有任务执行的时候线程在干嘛？我觉得总不至于在吃饭睡觉打豆豆吧，那这样太没有意义了，其实主要是让它们让出宝贵的CPU资源给有需要的线程使用。这就需要引入一个信号标识“指挥”它们，让它们在有需要的时候“醒来”执行任务，没有任务的时候就“睡觉”。所有线程的主任务就是去任务队列里领取任务，如果没有则等待信号标志。当有新任务加入任务队列的时候，则发信号让其中一个线程“醒来”工作。当主线程需要结束的时候，此时通过信号标志唤醒所有的子线程，“指挥”它们依次退出。

thread pool 实现

ThreadPool声明如下：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
class ThreadPool {
    public:
    typedef std::shared_ptr<Pthread> ThreadPtr;
    typedef std::function<void(void)> Task;
    ThreadPool(int max_count = 1);
    ~ThreadPool() = default;
    void AddTask(const Task &callback);

    void Shutdown();

    private:
    void* ThreadLoop(void*);
    bool TaskEmpty();
    private:
    Condition cv_;
    Mutex mtx_;
    bool shutdown_;
    std::vector<ThreadPtr> threads_;
    std::deque<Task> task_list_;

};

创建ThreadPool实例的时候即指定所需子线程的数量，默认为1，即创建一个子线程

ThreadLoop函数是所有的子线程的创建后运行的函数，实现如下：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
void* ThreadPool::ThreadLoop(void*)
{
	while(true)
    {
        while(!shutdown_ && this->TaskEmpty())
        {
            //no task,wait the new task
            cv_.Wait();
        }
        if(!this->TaskEmpty())
        {
            mtx_.Lock();
            auto &task = task_list_.front();
            task_list_.pop_front();
            mtx_.Unlock();

            if(task)
            {
                task();
            }
        }
        else if(shutdown_)
        {
            break;
        }
    }
  return nullptr;
}

如果任务队列里没有任务，则等待cv_信号量；如果有任务，则执行

在实践中，判断线程休眠条件时最好是循环判断，即while模式，因为信号量有可能会被其他的线程发送全局唤醒信号唤醒

每次通过AddTask加任务时，通过调用信号量的Notify函数就向等待信号量的子线程发送一次信号，以此唤醒子线程执行任务。由于所有的线程都可以操作任务队列，为了保证任务队列的线程安全，必须对其加入锁机制，即mutex。

任务加入时锁机制实现如下：

1
2
3
mtx_.Lock();
task_list_.push_back(callback);
mtx_.Unlock();

任务移出时锁机制如下：

1
2
3
4
mtx_.Lock();
auto &task = task_list_.front();
task_list_.pop_front();
mtx_.Unlock();

总结

初始版本中，最后一个任务执行完之后会有很大概率产生段错误，造程序异常退出。分析原因如下

在A线程的if判断中，任务列表不为空，然后系统调度另一个线程B执行，此线程将任务列表清空后，系统将前面休眠的线程A调度起来，此时该线程执行task_list_.pop_front()函数，由于列表中已经没有了任务，所以系统产生段错误

在进行判断列表为空的操作时加入读写锁后，程序正常。