C++学习记录：一个小线程池的源码分析

发表于 分类于 阅读次数： 本文字数： 1.3k 阅读时长 ≈ 5 分钟

一、源码一览

  核心代码很简单，就是下面这不到一百行。但是其中使用了很多新C++11的新东西，写的非常优雅，有很多可以学习的地方。

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
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
#ifndef THREAD_POOL_H
#define THREAD_POOL_H

#include <vector>
#include <queue>
#include <memory>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <future>
#include <functional>
#include <stdexcept>

class ThreadPool {
public:
    ThreadPool(size_t);
    template<class F, class... Args>
    auto enqueue(F&& f, Args&&... args) 
        -> std::future<typename std::result_of<F(Args...)>::type>;
    ~ThreadPool();
private:
    // need to keep track of threads so we can join them
    std::vector< std::thread > workers;
    // the task queue
    std::queue< std::function<void()> > tasks;
    
    // synchronization
    std::mutex queue_mutex;
    std::condition_variable condition;
    bool stop;       
};
 
// the constructor just launches some amount of workers
inline ThreadPool::ThreadPool(size_t threads)
    :   stop(false)
{
    for(size_t i = 0;i<threads;++i)
        workers.emplace_back(
            [this]
            {
                for(;;)
                {
                    std::function<void()> task;

                    {
                        std::unique_lock<std::mutex> lock(this->queue_mutex);
                        this->condition.wait(lock,
                            [this]{ return this->stop || !this->tasks.empty(); });
                        if(this->stop && this->tasks.empty())
                            return;
                        task = std::move(this->tasks.front());
                        this->tasks.pop();
                    }

                    task();
                }
            }
        );
}

// add new work item to the pool
template<class F, class... Args>
auto ThreadPool::enqueue(F&& f, Args&&... args) 
    -> std::future<typename std::result_of<F(Args...)>::type>
{
    using return_type = typename std::result_of<F(Args...)>::type;

    auto task = std::make_shared< std::packaged_task<return_type()> >(
            std::bind(std::forward<F>(f), std::forward<Args>(args)...)
        );
        
    std::future<return_type> res = task->get_future();
    {
        std::unique_lock<std::mutex> lock(queue_mutex);

        // don't allow enqueueing after stopping the pool
        if(stop)
            throw std::runtime_error("enqueue on stopped ThreadPool");

        tasks.emplace([task](){ (*task)(); });
    }
    condition.notify_one();
    return res;
}

// the destructor joins all threads
inline ThreadPool::~ThreadPool()
{
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        stop = true;
    }
    condition.notify_all();
    for(std::thread &worker: workers)
        worker.join();
}

#endif

  

二、源码分析

1. 构造部分

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
// the constructor just launches some amount of workers
inline ThreadPool::ThreadPool(size_t threads)
    :   stop(false)
{
    for(size_t i = 0;i<threads;++i)
        workers.emplace_back(
            [this]
            {
                for(;;)
                {
                    std::function<void()> task;

                    {
                        std::unique_lock<std::mutex> lock(this->queue_mutex);
                        this->condition.wait(lock,
                            [this]{ return this->stop || !this->tasks.empty(); });
                        if(this->stop && this->tasks.empty())
                            return;
                        task = std::move(this->tasks.front());
                        this->tasks.pop();
                    }

                    task();
                }
            }
        );
}

  如上为线程池的构造函数部分。其中主要是根据构造传参创建对应数量的线程，线程储存在 vector< std::thread > 中，线程通过 emplace_back 加 lambda 函数直接构造进vector里。

  线程的流程总体也比较简单，就是一个死循环加条件变量阻塞，当被唤醒时进行判定，如果要求线程退出则 return 出去，否则去任务队列 queue< std::function<void()> > 里取任务执行。这块内容的线程安全是通过一个 std::unique_lock 来保证的。

  值得一提的是这里的条件变量的 wait() 操作中添加了判定条件，从而避免虚假唤醒情况；并且获取 task 也是通过右值引用来进行的。
  

2. 析构部分

1
2
3
4
5
6
7
8
9
10
11
// the destructor joins all threads
inline ThreadPool::~ThreadPool()
{
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        stop = true;
    }
    condition.notify_all();
    for(std::thread &worker: workers)
        worker.join();
}

  析构部分总体也比较简单，即通过 notify_all() 唤醒所有条件变量，随后使用 join() 等待所有线程执行完毕，完成线程池的同步退出。

  另外在这个线程池实现中，是通过一个变量 stop 来标记线程池是否退出的，线程安全也是通过一个 std::unique_lock 来保证的。这个锁逻辑意义上是任务队列锁，确保关于任务队列相关内容操作的线程安全。
  

3. 任务入队部分

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
// add new work item to the pool
template<class F, class... Args>
auto ThreadPool::enqueue(F&& f, Args&&... args) 
    -> std::future<typename std::result_of<F(Args...)>::type>
{
    using return_type = typename std::result_of<F(Args...)>::type;

    auto task = std::make_shared< std::packaged_task<return_type()> >(
            std::bind(std::forward<F>(f), std::forward<Args>(args)...)
        );
        
    std::future<return_type> res = task->get_future();
    {
        std::unique_lock<std::mutex> lock(queue_mutex);

        // don't allow enqueueing after stopping the pool
        if(stop)
            throw std::runtime_error("enqueue on stopped ThreadPool");

        tasks.emplace([task](){ (*task)(); });
    }
    condition.notify_one();
    return res;
}

  这是我感觉整个线程池中，写的最优雅的部分，其中运用了大量C++11的新东西。首先这个部分完成实现了任务入队操作，并且可以通过 std::future 系列操作来实现异步获取执行结果。

  其中的实现是通过 std::packaged_task + std::bind 封装了可以异步执行的任务函数，函数返回一个 std::future 对象。通过这个对象即可异步获取线程的执行结果。另外这部分也中也使用了智能指针 std::shared_ptr，来保证task对象部分的智能释放。

  此外，其中的任务队列通过 std::unique_lock 来保证线程安全，并且也通过大括号来限定临界区域。在任务加入任务队列中，会通过条件变量来唤醒一个已阻塞的线程，从而继续任务处理流程。

  

三、小结

  所以这个线程池的任务处理流程就是：开辟线程 -> 阻塞线程 -> 任务入队 -> 唤醒线程 -> 执行线程 -> 阻塞线程.
  总体来说思路很简单，关于执行结果的接收直接通过C++11的新库 std::future 来实现了，并且通过 std::unique_lock 和 std::condition_variable 来保证线程安全和线程等待的的最小化消耗。其中很多功能都是直接使用了C++11封装好的库，所以看起来很清晰明了并且写的也很优雅。但是毕竟代码量很少，所以相比很多完善的线程池还有很多优化的空间，不过当一个轻量化的线程池来使用我感觉也是够用了，而且从其中我也学习到了不少C++11的新库用法。