C++ 11 实现（同步的）任务队列线程池 正文

发表于

「这是我参与11月更文挑战的第 7 天，活动详情查看：2021最后一次更文挑战」。

参加该活动的第 15 篇文章

参考原文地址： （原创）C++ 半同步半异步线程池

我对文章的格式和错别字进行了调整，并在他的基础上，根据我自己的理解把重点部分进一步解释完善（原作者的代码注释甚少）。以下是正文。

正文

线程池可以开启多个线程高效并行处理任务，一开始各个线程会等待同步队列中的任务到来，任务到来后多个线程会抢着执行，但是当到来的任务太多并且达到上限时，线程则需要等待片刻，任务上限是为了保证内存不会溢出。

image.png
线程池的效率和 CPU 核数相关，多核的话效率会更高，线程数一般取 CPU 数量+ 2 比较合适，否则线程过多，线程频繁切换反而会导致效率降低。

线程池有两个活动过程：

  1. 外面有一个线程不停的为线程池（的任务队列）添加任务；
  2. 线程池内部的线程不停地（从任务队列中）取任务执行。

活动图如下

image.png

线程池中的队列是用的上一篇博文中的同步队列。具体代码：

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
cpp复制代码#include <vector>
#include <thread>
#include <functional>
#include <memory>
#include <atomic>
#include "SyncQueue.hpp"

const int MaxTaskCount = 100;
class ThreadPool
{
public:
    using Task = std::function<void()>;
    
    /// @note 获取硬件支持的并发数
    ThreadPool(int numThreads = std::thread::hardware_concurrency()) : m_queue(MaxTaskCount)
    {
        Start(numThreads);
    }

    ~ThreadPool(void)
    {
        /// @note 主动停止线程池
        Stop();
    }

    /// @note 关闭线程池
    void Stop()
    {
        std::call_once(m_flag, [this]
                       { StopThreadGroup(); }); ///< 保证多线程情况下只调用一次 StopThreadGroup
    }

    /// @note 为任务队列添加任务
    void AddTask(Task &&task)
    {
        m_queue.Put(std::forward<Task>(task));
    }
    
    /// @note 同上
    void AddTask(const Task &task)
    {
        m_queue.Put(task);
    }

private:
    /// @note 开启线程池
    void Start(int numThreads)
    {
        m_running = true;
        /// @note 创建线程组（std::list<std::shared_ptr<std::thread>>）
        for (int i = 0; i < numThreads; ++i)
        {
            m_threadgroup.push_back(std::make_shared<std::thread>(&ThreadPool::RunInThread, this));
        }
    }

    /// @note 各个线程从任务队列中取出任务，然后执行
    void RunInThread()
    {
        while (m_running)
        {
            /// @note 取任务分别执行
            std::list<Task> list;
            m_queue.Take(list);

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

                task();
            }
        }
    }

    /// @note Stop 调用，关闭线程池
    void StopThreadGroup()
    {
        m_queue.Stop();    ///< 任务队列的出队和入队操作中断
        m_running = false; ///< 置为 false ，让内部线程跳出循环并退出

        for (auto thread : m_threadgroup) ///< 等待线程池的线程结束
        {
            if (thread)
                thread->join();
        }
        m_threadgroup.clear();
    }

    std::list<std::shared_ptr<std::thread>> m_threadgroup; ///< 处理任务的线程组
    SyncQueue<Task> m_queue;                               ///< 线程同步的任务队列
    atomic_bool m_running;                                 ///< 是否停止的标志
    std::once_flag m_flag;                                 ///< 用于 std::call_once
};

上面的代码中用到了同步队列 SyncQueue ，它的实现在这里。测试代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
cpp复制代码void TestThdPool()
{
    ThreadPool pool;
    bool runing = true;

    /// @note 不停地给线程池添加任务的子线程
    std::thread thd1([&pool, &runing]
                     {
                         while (runing)
                         {
                             cout << "produce " << this_thread::get_id() << endl;

                             pool.AddTask([]
                                          { std::cout << "consume " << this_thread::get_id() << endl; });
                         }
                     });

    this_thread::sleep_for(std::chrono::seconds(10));
    runing = false;
    pool.Stop();

    thd1.join();
    getchar();
}

上面的测试代码中，thd1 是生产者线程，线程池内部会不断消费生产者产生的任务。在需要的时候可以提前停止线程池，只要调用 Stop 函数就行了。

执行结果如下

image.png

本例中涉及的其他知识点

std::call_once

1
2
3
4
> cpp复制代码template <class Fn, class... Args>
> void call_once (once_flag& flag, Fn&& fn, Args&&...args);
> 
>
  • 第一个参数是 std::once_flag 的对象( once_flag 是不允许修改的，其拷贝构造函数和 operator= 函数都声明为 delete )，
  • 第二个参数可调用实体，即要求只执行一次的代码，后面可变参数是其参数列表。
  • call_once 保证函数 fn 只被执行一次，如果有多个线程同时执行函数 fn 调用，则只有一个活动线程(active call)会执行函数，其他的线程在这个线程执行返回之前会处于 ”passive execution” (被动执行状态) —— 不会直接返回，直到活动线程对 fn 调用结束才返回。对于所有调用函数 fn 的并发线程，数据可见性都是同步的(一致的)。
  • 如果活动线程在执行 fn 时抛出异常，则会从处于 ”passive execution” 状态的线程中挑一个线程成为活动线程继续执行 fn ，依此类推。一旦活动线程返回，所有 ”passive execution” 状态的线程也返回, 不会成为活动线程。(实际上 once_flag 相当于一个锁，使用它的线程都会在上面等待，只有一个线程允许执行。如果该线程抛出异常，那么从等待中的线程中选择一个，重复上面的流程)。

std::thread::hardware_concurrency

  • 功能，获取硬件支持的并发线程数
  • 返回值，正常返回支持的并发线程数，若值非错误定义或不可计算，则返回 0

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%