线程池技术

对于大量并发任务，传统方式处理任务特点：
一个请求一个线程来处理请求任务，大量线程的创建和销毁将消耗过多的系统资源，增加线程上下文切换的开销。

线程池技术特点：
在系统中预先创建一定数量的线程，当任务请求到来时，从线程池中分配一个预先创建的线程去处理任务，线程处理完任务后还可以重用，不会销毁线程而是等待下一次任务的到来。这样，避免大量的线程创建和销毁动作，从而实现节省系统资源。

线程池好处：
1）对于多核处理器，由于线程会被分配到多个CPU，会提高并行处理效率；
2）每个线程独立阻塞，可以防止由于主线程被阻塞而导致其他请求得不到响应的问题。

线程池分类：
1）半同步半异步线程池
2）领导者追随者线程池

半同步半异步线程池模型：

第一层：同步服务层，处理来自上层的任务请求，上层的请求可能是并发的，这些请求不会马上被处理，而是会被放到一个同步排队层中，等待处理。
第二层：同步排队层，来自上层的任务请求都会加到排队层中等待处理。
第三层：异步服务层，有多个线程同时处理排队层中的任务，异步服务层从排队层中取出任务并行的处理。

这种三层结构可以最大程度处理上层并发请求。对于上层来说，只需要将任务丢到同步队列即可，至于谁处理、什么时候出来，上层无需关心，主线程也不会阻塞，也能发起新请求。任务如何处理，是靠异步服务层（第三层）的多线程异步并行完成的。

两者区别在于任务队列的实现。领导者/追随者模式的核心是线程池，一个或多个线程扮演追随者角色，并在线程池同步器上排队等待扮演领导者角色，其中一个线程将会被选择为领导者，等待时间在句柄集中的句柄上发生。当有一个事件发生时，当前领导者线程将一个追随者线程提升为新的领导者；原来的领导者接着扮演处理线程的角色。领导者/追随者模式实现更复杂。本文主要讲解半同步半异步线程池。

关键技术

线程池有2个活动：
1）外面不停的往线程池添加任务；
2）线程池内部不停地取任务执行。

半同步半异步线程池活动图如下：

从内部看，线程池各线程不断从任务队列取任务执行；从外部看，用户不断往任务队列添加任务。可看出，任务队列是线程池运作的核心。

同步队列

同步队列是半同步半异步线程池三层结构的中间层，即上面的任务队列，主要作用：保证队列中共享数据的线程安全，为上层同步服务层提供添加新任务的接口，为下层异步服务层提供取任务的接口。同时，限制任务数的上限值，避免任务过多导致内存暴涨。

同步队列实现较为简单，主要包括：互斥锁、条件变量、右值引用、std::move、std::forward等C++技术。

同步队列典型实现代码：

#include <list>
#include <thread>
#include <mutex>
#include <condition_variable>

template<typename T>
class SyncQueue
{
public:
    SyncQueue(int maxSize)
    : m_maxSize(maxSize), m_needStop(false)
    {}

    void Put(const T& x)
    {
        Add(x);
    }
    void Put(T&& x)
    {
        Add(std::forward<T>(x));
    }

    void Take(std::list<T>& list)
    {
        std::unique_lock<std::mutex> locker(m_mutex);
        m_notEmpty.wait(locker, [this]{ return m_needStop || NotEmpty(); });

        if (m_needStop) return;
        list = std::move(m_queue);
        m_notFull.notify_one();
    }
    void Take(T& t)
    {
        std::unique_lock<std::mutex> locker(m_mutex);
        m_notEmpty.wait(locker, [this]{ return m_needStop || NotEmpty(); });

        if (m_needStop) return;
        t = m_queue.front();
        m_queue.pop_front();
        m_notFull.notify_one();
    }

    void Stop()
    {
        {
            std::unqiue_lock<std::mutex> locker(m_mutex);
            m_needStop = true;
        }
        m_notFull.notify_all();
        m_notEmpty.notify_all();
    }

    bool Empty()
    {
        std::lock_guard<std::mutex> locker(m_mutex);
        return m_queue.empty();
    }

    bool Full()
    {
        std::lock_guard<std::mutex> locker(m_mutex);
        return m_queue.size() == m_maxSize;
    }

    size_t Size()
    {
        std::lock_guard<std::mutex> locker(m_mutex);
        return m_queue.size();
    }

    // FIXME: Why owns Size(), still need Count()?
    // Not thread safe
    int Count()
    {
        return m_queue.size();
    }

private:
    template<typename F>
    void Add(F&& x)
    {
        std::unique_lock<std::mutex> locker(m_mutex);
        m_notFull.wait(locker, [this]{ return m_needStop || NotFull(); });
        if (m_needStop) return;

        m_queue.push_back(std::forward<F>(x));
        m_notEmpty.notify_one();
    }

    // Not thread safe
    bool NotFull() const
    {
        bool full = m_queue.size() >= m_maxSize;
        if (full) {
            std::cout << "Buffer is full, need to wait..." << std::endl;
        }
        return !full;
    }

    // Not thread safe
    bool NotEmpty() const
    {
        bool empty = m_queue.empty();
        if (empty) {
            std::cout << "Buffer is empty, need to wait... thread id: "
                      << std::this_thread::get_id() << std::endl;
        }
        return !empty;
    }

    std::list<T> m_queue;                     // task buffer
    std::mutex m_mutex;                       // mutex lock
    std::condition_variable m_notEmpty;
    std::condition_variable m_notFull;
    int m_maxSize;                            // queue's max size
    bool m_needStop;                          // stop flag
};

为什么有了接口Take(T& t)，还会有一个重载的Take(std::list& list)？
因为前者只能一次获取一个数据，每条数据都加锁获取效率很低，后者只加一次锁就获取所有数据，方法是将list m_queue直接通过std::move移交给参数list，这之后m_queue会被清空。

Take函数

主要作用是从同步队列取数据（任务），通过std::mutex + std::unique_lock + 条件变量 确保线程安全。条件变量m_notEmpty.wait等待两个条件：1）m_needStop 队列需要停止标志；2）NotEmpty() 缓冲区非空函数返回true。

由于使用了条件变量，可能有其他线程等待条件满足，所以当获取数据完毕后，需要用m_notFull（队列非满）来唤醒notify_one()等待该条件的某个线程，以重新竞争锁。

    void Take(std::list<T>& list)
    {
        std::unique_lock<std::mutex> locker(m_mutex);
        // 阻塞等待lambda表达式返回true
        m_notEmpty.wait(locker, [this]{ return m_needStop || NotEmpty(); });

        if (m_needStop) return;
        list = std::move(m_queue); // 将m_queue移交给list, m_queue会被清空
        m_notFull.notify_one();    // 唤醒某个等待在该条件变量上的线程
    }

Add函数

主要作用是将任务添加到同步队列末尾。public接口Put()实际上转发给私有接口Add()，且直接在Add()中加锁。
注：这种惯例不是特别推荐，更建议在public接口中加锁，private接口中不加锁。

    template<typename F>
    void Add(F&& x)
    {
        std::unique_lock<std::mutex> locker(m_mutex);
        m_notFull.wait(locker, [this]{ return m_needStop || NotFull(); });
        if (m_needStop) return;

        m_queue.push_back(std::forward<F>(x));
        m_notEmpty.notify_one();
    }

Stop函数

同步队列包含了一组不间断运行的子线程，如何停止？
可以利用Stop函数，设置m_needStop为true，因为各子线程主循环循环会查询该条件，以决定是否退出线程。

    void Stop()
    {
        {
            std::unqiue_lock<std::mutex> locker(m_mutex);
            m_needStop = true;
        }
        // 注意这里是唤醒所有等待条件的线程
        m_notFull.notify_all(); 
        m_notEmpty.notify_all();
    }

线程池

前面讲的同步队列是排队层，而一个完整线程池包括：同步服务层、排队层、异步服务层。这是一种生产者——消费者模型，同步层是生产者，不断将新任务加入排队层，因此线程池需要提供添加任务的接口供生产者使用；消费者是异步层，线程池预先创建一组线程处理排队层中的任务。

线程池与生产者——消费者模型关系：

线程池典型实现：

#include <list>
#include <thread>
#include <functional>
#include <memory>
#include <atomic>
#include "SyncQueue.h"

const int MaxTaskCount = 100;
class ThreadPool
{
public:
    using Task = std::function<void()>;
    // thread::hardware_concurrency(): number of concurrent threads
    ThreadPool(int numThreads = std::thread::hardware_concurrency())
    : m_queue(MaxTaskCount)
    {
        Start(numThreads);
    }
    ~ThreadPool()
    {
        // stop the thread pool if not stop
        Stop();
    }

    void Stop()
    {
        // ensure invoke StopThreadGroup() just one time
        std::call_once(m_flag, [this]{ StopThreadGroup(); });
    }

    void AddTask(Task&& task)
    {
        m_queue.Put(std::forward<Task>(task));
    }
    void AddTask(const Task& task)
    {
        m_queue.Put(task);
    }

private:
    void Start(int numThreads)
    {
        m_running = true;
        // create thread group
        for (int i = 0; i < numThreads; ++i) {
            m_threadgroup.push_back(
                    std::make_shared<std::thread>(&ThreadPool::RunInThread, this));
        }
    }
    void RunInThread()
    {
        while (m_running) {
            // take task from queue and execute
            std::list<Task> list;
            m_queue.Take(list);

            for (auto& task : list) {
                if (!m_running) return;

                task();
            }
        }
    }

    void StopThreadGroup()
    {
        m_queue.Stop();
        m_running = false;

        for (auto thread : m_threadgroup) { // wait for thread ending
            if (thread) thread->join();
        }
        m_threadgroup.clear();
    }

    std::list<std::shared_ptr<std::thread>> m_threadgroup; // thread array to process task
    SyncQueue<Task> m_queue;     // synchronous queue
    std::atomic_bool m_running;  // stop thread pool flag
    std::once_flag m_flag;       // run once
};

应用线程池

下面例子展示如何使用半同步半异步的线程池。线程池将初始创建两个线程，外部线程不断向线程中添加新任务，内部线程将并行处理同步队列中的任务。

void TestThdPool()
{
    ThreadPool pool;
    pool.Start(2);

    std::thread thd1([&pool]{
        for (int i = 0; i < 10; ++i) {
            auto thdId = std::this_thread::get_id();
            pool.AddTask([thdId]{
               std::cout << "同步层线程1的ID: " << thdId << std::endl;
            });
        }
    });

    std::thread thd2([&pool]{
        for (int i = 0; i < 10; ++i) {
            auto thdId = std::this_thread::get_id();
            pool.AddTask([thdId]{
                std::cout << "同步层线程2的线程ID: " << std::endl;
            });
        }
    });

    std::this_thread::sleep_for(std::chrono::seconds(2));
    getchar();
    pool.Stop();
    thd1.join();
    thd2.join();
}

参考

[1]祁宇. 深入应用C++11 : 代码优化与工程级应用 : In-Depth C++11 : code optimization and engineering level application[M]. 机械工业出版社, 2015.
[2]spdlog日志库源码：线程池thread_pool