多线程笔记：线程池

相关资料

线程池的概念和相关示例可以参考：
C++实现线程池
基于C++11实现线程池的工作原理

本代码相关的C++基础，参考：
c++拷贝构造函数详解
智能指针shared_ptr的用法
深入解析条件变量

其他相关文章
jorion/c++11 多线程（X）
jorionwen/threadtest

线程池示例

调用线程池

#include "TaskPool.h"
#include <chrono>

int main()
{
    TaskPool threadPool;
    threadPool.init(); //初始化线程对象队列

    Task* task = NULL;
    for (int i = 0; i < 10; ++i)
    {
        task = new Task();
        threadPool.addTask(task); //初始化任务对象队列，调度线程时会取出执行
    }
    
    std::this_thread::sleep_for(std::chrono::seconds(2));

    threadPool.stop(); //等待所有工作线程结束

    return 0; //析构
}

线程池的方法

#include "TaskPool.h"

TaskPool::TaskPool() : m_bRunning(false)
{

}

TaskPool::~TaskPool()
{
    removeAllTasks();
}

void TaskPool::init(int threadNum/* = 5*/)
{
    if (threadNum <= 0)
        threadNum = 5;

    m_bRunning = true;

    for (int i = 0; i < threadNum; ++i)
    {
        std::shared_ptr<std::thread> spThread;
        //shared_ptr.reset带参数是初始化，指向new出的thread对象
        //bind绑定了thread对象和其执行函数threadFunc
        spThread.reset(new std::thread(std::bind(&TaskPool::threadFunc, this))); 
        m_threads.push_back(spThread); //thread对象入队
    }
}

void TaskPool::threadFunc() //thread对象唤醒时执行
{
    std::shared_ptr<Task> spTask;
    while (true)
    {
        std::unique_lock<std::mutex> guard(m_mutexList); //RAII实现，作用域结束自动解锁
        while (m_taskList.empty())
        {                 
            if (!m_bRunning)
                break;
            
            //如果获得了互斥锁，但是条件不合适的话，pthread_cond_wait会释放锁，不往下执行。
            //当发生变化后，条件合适，pthread_cond_wait将直接获得锁。
            m_cv.wait(guard);
        }

        if (!m_bRunning)
            break;

        spTask = m_taskList.front(); //取m_taskList的task对象
        m_taskList.pop_front(); //更新m_taskList

        if (spTask == NULL)
            continue;

        spTask->doIt(); //执行task
        spTask.reset(); //shared_ptr.reset不带参数，指向对象的计数-1
    }

    std::unique_lock<std::mutex> guard(m_mutexList); //为了打印的原子性，再加锁
    {
        std::cout << "Exit thread, threadID: " << std::this_thread::get_id() << std::endl;
    }
    
}

void TaskPool::stop()
{
    m_bRunning = false;
    m_cv.notify_all(); //唤醒所有等待条件变量的线程

    //等待所有线程退出
    for (auto& iter : m_threads)
    {
        if (iter->joinable())   //该线程是否可join
            iter->join();       //主线程等待该线程
    }
}

void TaskPool::addTask(Task* task)
{
    std::shared_ptr<Task> spTask;
    spTask.reset(task); //shared_ptr初始化，指向task

    {
        std::lock_guard<std::mutex> guard(m_mutexList);       
        //m_taskList.push_back(std::make_shared<Task>(task));
        m_taskList.push_back(spTask); //Task对象入队
        std::cout << "Add a Task." << std::endl;
    }
    
    m_cv.notify_one(); //唤醒随机一个等待条件变量的线程
}

void TaskPool::removeAllTasks()   //析构时调用
{
    {
        std::lock_guard<std::mutex> guard(m_mutexList);
        for (auto& iter : m_taskList)
        {
            iter.reset();
        }
        m_taskList.clear();
    }
}

类定义

#include <thread>
#include <mutex>
#include <condition_variable>
#include <list>
#include <vector>
#include <memory>
#include <iostream>
#include <functional> //for visual studio build

class Task
{
public:
    virtual void doIt()
    {
        std::cout << "Do a task..." << std::endl;
    }

    virtual ~Task()
    {
        //为了看到一个task的销毁，这里刻意补上其析构函数
        std::cout << "A task destructed..." << std::endl;
    }
};

class TaskPool final
{
public:
    TaskPool();
    ~TaskPool();
    TaskPool(const TaskPool& rhs) = delete;     //delete: 禁用某函数（C++11），阻止拷贝和赋值构造，C++98用private
    TaskPool& operator=(const TaskPool& rhs) = delete;

public:
    void init(int threadNum = 5);   //默认初始化
    void stop();

    void addTask(Task* task);
    void removeAllTasks();

private:
    void threadFunc();

private:
    std::list<std::shared_ptr<Task>>            m_taskList;
    std::mutex                                  m_mutexList;
    std::condition_variable                     m_cv;
    bool                                        m_bRunning;
    std::vector<std::shared_ptr<std::thread>>   m_threads;
};

运行