• C++多线程的用法（包含线程池小项目）

    一些小Tips: 

    • linux环境下，编译多线程程序时，需要加上后缀 -lpthread 引入线程库：
     g++ 7.thread_pool.cpp -lpthread
    • bash下查看一个可执行程序的运行时间：
    time ./a.out
    •   需要引入的库函数有：
    1. #include // 引入线程库
    2. #include // 加入锁机制需要引入库函数mutex
    3. #include // 引入信号量机制
    •  在C++程序中获得本进程的进程id：
    this_thread::get_id()
    • 在C++程序中如何定义信号量、锁：
    1. condition_variable m_cond    // 信号量
    2. std::mutex m_mutex;          // 互斥锁，在非std命名空间的情况下
    • 一个线程的栈所占用的空间有多大？——8M。查看命令：
    ulimit -a

    多线程概念：

    • 进程是资源分配的最基本单位；线程是进程中的概念，是调度的基本单位，共享进程空间资源。
    • 什么是临界资源？——多线程情况下，大家都能访问到的资源。
    • 所谓的多线程只不过就是指定的某一个执行函数为入口函数的另外一套执行流程。

    代码样例：

    1. #include
    2. #include
    3. using namespace std;
    4.  
    5. #define BEGINS(x) namespace x{   
    6. #define ENDS(x) }
    7.  
    8. BEGINS(thread_usage)    // 自定义命名空间
    9. void func() {
    10.     cout << "hello wolrd" << endl;
    11.     return ;
    12. }
    13. int main() {
    14.     thread t1(func);  // t1已经开始运行了
    15.     t1.join();        // 等待t1线程结束
    16.     return 0;
    17. }
    18. ENDS(thread_usage)
    19.  
    20. int main() {
    21.     thread_usage::main();     // 调用该命名空间下的main函数
    22.  
    23.     return 0;
    24. }
    • 那么如何给入口函数传参呢？ 在C++中就很简单：
    1. void print(int a, int b) {
    2.     cout << a << " " << b << endl;
    3.     return ;
    4. }
    5. int main() {
    6.     thread t2(print, 3, 4);
    7.     t2.join();
    8.     return 0;
    9. }

    多线程设计理念：

    • 多线程封装思维：在相应的功能函数内部，不要去访问全局变量，类似于一个原子操作的功能。本身就支持多线程并行。
    • 多线程程序设计：线程功能函数和入口函数要分开。（高内聚低耦合）
    • 能不用锁就不用锁，因为这个锁机制会给程序运行效率带来极大的影响。

    1. 多线程实现素数统计的功能

    1.1. 不加锁版：

    1. #include
    2. #include
    3. #include
    4. #include
    5. using namespace std;
    6.  
    7. #define BEGINS(x) namespace x{
    8. #define ENDS(x) }
    9.  
    10. BEGINS(is_prime)
    11. bool is_prime(int x) {
    12.     for (int i = 2, I = sqrt(x); i <= I; i++) {
    13.         if (x % i == 0) return false;
    14.     }
    15.     return true;
    16. }
    17. // 多线程——功能函数 
    18. int prime_count(int l, int r) {
    19.     // 从l到r范围内素数的数量
    20.     int ans = 0;
    21.     for (int i = l; i <= r; i++) {
    22.         ans += is_prime(i);
    23.     }
    24.     return ans;
    25. }
    26. // 多线程——入口函数 
    27. void worker(int l, int r, int &ret) {
    28.     cout << this_thread::get_id() << "begin" << endl;
    29.     ret = prime_count(l, r);
    30.     cout << this_thread::get_id() << "dnoe" << endl;
    31. }
    32. int main() {
    33.     #define batch 500000
    34.     thread *t[10];
    35.     int ret[10];
    36.     for (int i = 0, j = 1; i < 10; i++, j += batch) {
    37.         t[i] = new thread(worker, j, j + batch - 1, ref(ret[i]));
    38.     }
    39.     for (auto x : t) x->join();
    40.     int ans = 0;
    41.     for (auto x : ret) ans += x;
    42.     for (auto x : t) delete x;
    43.     cout << ans << endl;
    44.     #undef batch
    45.     return 0;
    46. }
    47. ENDS(is_prime)
    48.  
    49. int main() {
    50.     // thread_usage::main();
    51.     is_prime::main();
    52.     return 0;
    53. }
    • 运行时间：总时长0.512s，系统时间0.02s，用户时间1.92s

    1.2. 加锁版：

    1. #include
    2. #include
    3. #include
    4. #include
    5. using namespace std;
    6.  
    7. #define BEGINS(x) namespace x{
    8. #define ENDS(x) }
    9. BEGINS(prime_count2) 
    10. int ans = 0;
    11. std::mutex m_mutex;
    12. bool is_prime(int x) {
    13.     for (int i = 2, I = sqrt(x); i <= I; i++) {
    14.         if (x % i == 0) return false;
    15.     }
    16.     return true;
    17. }
    18. void prime_count(int l, int r) {
    19.     cout << this_thread::get_id() << " begin\n";
    20.     for (int i = l; i <= r; i++) {
    21.         std::unique_lock lock(m_mutex); // 临界区
    22.         ans += is_prime(i);
    23.         lock.unlock();    
    24.     }
    25.     cout << this_thread::get_id() << " done\n";
    26.     return ;
    27. }
    28. int main() {
    29.     #define batch 500000
    30.     thread *t[10];
    31.     for (int i = 0, j = 1; i < 10; i++, j += batch) {
    32.         t[i] = new thread(prime_count, j, j + batch - 1);
    33.     }
    34.     for (auto x : t) x->join();
    35.     for (auto x : t) delete x;
    36.     cout << ans << endl;
    37.     #undef batch
    38.     return 0;
    39. }
    40. ENDS(prime_count2)
    41.  
    42. int main() {
    43.     prime_count2::main();
    44.     return 0;
    45. }
    • 运行时间：总时长5.916s，系统时间7.44s，用户时间6.50s

    显然能够看出，加了锁要比不加锁耗时得多。所以加锁的行为要慎重。

    1.3. 引入原子操作版：

    • 为什么不用++而是用+=：因为+=是原子操作，而++不是，所以++操作在多线程情况下可能存在覆盖写的问题。
    • 为什么使用__sync_fetch_and_add函数：因为这个函数也是原子操作。
    1. #include
    2. #include
    3. #include
    4. #include
    5. using namespace std;
    6.  
    7. #define BEGINS(x) namespace x{
    8. #define ENDS(x) }
    9.  
    10. BEGINS(thread3)
    11. int ans = 0;
    12. bool is_prime(int x) {
    13.     for (int i = 2, I = sqrt(x); i <= I; i++) {
    14.         if (x % i == 0) return false;
    15.     }
    16.     return true;
    17. }
    18. void prime_count(int l, int r) {
    19.     cout << this_thread::get_id() << "begin\n";
    20.     for (int i = l; i <= r; i++) {
    21.         int ret = is_prime(i);
    22.         __sync_fetch_and_add(&ans, ret);
    23.     }
    24.     cout << this_thread::get_id() << "done\n";
    25. }
    26. int main() {
    27.     #define batch 500000
    28.     thread *t[10];
    29.     for (int i = 0, j = 1; i < 10; i++, j += batch) {
    30.         t[i] = new thread(prime_count, j, j + batch - 1);
    31.     }
    32.     for (auto x : t) x->join();
    33.     for (auto x : t) delete x;
    34.     cout << ans << endl;
    35.     #undef batch 
    36.     return 0;
    37. }
    38. ENDS(thread3)
    39.  
    40. int main() {
    41.     thread3::main();
    42.     return 0;
    43. }

    • - `2.06s user`：表示程序在用户态消耗的CPU时间为2.06秒。这是程序执行期间用于执行用户代码的时间。
    • - `0.02s system`：表示程序在内核态消耗的CPU时间为0.02秒。这是程序执行期间用于执行内核代码的时间。
    • - `385% cpu`：表示程序使用了385%的CPU资源。这个值超过100%是因为程序可能在多个CPU核心上并行执行。所以 总时间 < 用户态+内核态时间 是非常正常的。
    • - `0.538 total`：表示程序从开始执行到结束的总时间为0.538秒。这是包括了实际执行时间、等待时间和其他非CPU消耗的时间。

    2. 小项目：线程池的实现

    • 线程作为一种内存资源，在通常的多线程设计模式下，申请线程资源并不可控。而我们自然希望申请内存的动作是可控的。
    • 一个好的解决办法就是限制一个池子的大小来存放一些线程贯穿程序运行周期始终，也就是线程池。在线程池内部，线程数量是可控的。把一个个计算任务打包，扔到一个任务队列中，线程池中的线程争相从任务队列中领取任务并执行。这样就实现了资源的有效利用和管控。
    • 什么是计算任务？——分为过程（函数方法）和数据（函数参数）；
    • 如何进行打包——bind()方法。

    代码实现：

    1. /*************************************************************************
    2.         > File Name: threadpool.cpp
    3.         > Author: jby
    4.         > Mail: 
    5.         > Created Time: Wed 13 Sep 2023 08:48:23 AM CST
    6.  ************************************************************************/
    7.  
    8. #include
    9. #include
    10. #include
    11. #include
    12. #include
    13. #include
    14. #include
    15. #include
    16. #include
    17. using namespace std;
    18.  
    19. #define BEGINS(x) namespace x {
    20. #define ENDS(x) }
    21.  
    22. BEGINS(thread_pool_test)
    23. class Task {
    24. public:
    25.     template<typename FUNC_T, typename ...ARGS>
    26.     Task(FUNC_T func, ARGS ...args) {
    27.         this->func = bind(func, forward(args)...);
    28.     }
    29.     void run() {
    30.         func();
    31.         return ;
    32.     }
    33. private:
    34.     function<void()> func; // 任意函数
    35. };
    36. class ThreadPool {
    37. public:
    38.     ThreadPool(int n = 1) : thread_size(n), threads(n), starting(false) {
    39.         this->start();
    40.         return ;
    41.     }
    42.     void worker() {
    43.         auto id = this_thread::get_id(); // 获得本进程号
    44.         running[id] = true;
    45.         while (running[id]) {
    46.             // 取任务
    47.             Task *t = get_task(); 
    48.             t->run();
    49.             delete t;
    50.         }
    51.         return ;
    52.     }
    53.     void start() {
    54.         if (starting == true) return ; // 如果已经开始了就不用启动了
    55.         for (int i = 0; i < thread_size; i++) {
    56.             threads[i] = new thread(&ThreadPool::worker, this);
    57.         }
    58.         starting = true;
    59.         return ;
    60.     }
    61.     template<typename FUNC_T, typename ...ARGS>
    62.     void add_task(FUNC_T func, ARGS ...args) {
    63.         unique_lock lock(m_mutex); 
    64.         tasks.push(new Task(func, forward(args)...)); // 任务池相当于临界资源
    65.         m_cond.notify_one(); // 生产者消费者模型
    66.         return ;
    67.     }
    68.     void stop() {
    69.         if (starting == false) return ; // 如果已经关了就不用再关了
    70.         for (int i = 0; i < threads.size(); i++) { // 往队列末尾投递毒药任务
    71.             this->add_task(&ThreadPool::stop_runnnig, this);
    72.         }
    73.         for (int i = 0; i < threads.size(); i++) {
    74.             threads[i]->join();
    75.         }
    76.         for (int i = 0; i < threads.size(); i++) {
    77.             delete threads[i]; // 释放那片进程剩余的空间
    78.             threads[i] = nullptr; // 进程指针指向空
    79.         }
    80.         starting = false;
    81.         return ;
    82.     }
    83.     ~ThreadPool() {
    84.         this->stop();
    85.         while (!tasks.empty()) { // 如果任务队列里还有任务没执行完，全部丢弃
    86.             delete tasks.front();
    87.             tasks.pop();
    88.         }
    89.         return ;
    90.     }
    91. private:
    92.     bool starting;
    93.     int thread_size;
    94.     Task *get_task() {
    95.         unique_lock lock(m_mutex);
    96.         while (tasks.empty()) m_cond.wait(lock); // 生产者消费者模型 //子进程的中wait函数对互斥量进行解锁，同时线程进入阻塞或者等待状态。
    97.         Task *t = tasks.front();
    98.         tasks.pop();
    99.         return t;
    100.     }
    101.     std::mutex m_mutex;
    102.     std::condition_variable m_cond;
    103.     vector threads; // 线程池子
    104.     unordered_map<decltype(this_thread::get_id()), bool> running; // 进程号到运行状态的哈希映射
    105.     queue tasks;  // 任务队列
    106.     void stop_runnnig() { // 毒药任务
    107.         auto id = this_thread::get_id();
    108.         running[id] = false;
    109.         return ;
    110.     }
    111. };
    112. bool is_prime(int x) {
    113.     for (int i = 2, I = sqrt(x); i <= I; i++) {
    114.         if (x % i == 0) return false;
    115.     }
    116.     return true;
    117. }
    118. int prime_count(int l, int r) {
    119.     int ans = 0;
    120.     for (int i = l; i <= r; i++) {
    121.         ans += is_prime(i);
    122.     }
    123.     return ans;
    124. }
    125. void worker(int l, int r, int &ret) {
    126.     cout << this_thread::get_id() << "begin\n";
    127.     ret = prime_count(l, r);
    128.     cout << this_thread::get_id() << "done\n";
    129.     return ;
    130. }
    131. int main() {
    132.     #define batch 500000
    133.     ThreadPool tp(5); // 五个任务的窗口队列
    134.     int ret[10];
    135.     for (int i = 0, j = 1; i < 10; i++, j += batch) {
    136.         tp.add_task(worker, j, j + batch - 1, ref(ret[i]));
    137.     }
    138.     tp.stop();
    139.     int ans = 0;
    140.     for (auto x : ret) ans += x;
    141.     cout << ans << endl;
    142.     #undef batch
    143.     return 0;
    144. }
    145. ENDS(thread_pool_test)
    146.  
    147. int main() {
    148.     thread_pool_test::main();
    149.     return 0;
    150. }

  • 相关阅读:
    K8S用户管理体系介绍
    记一次密码控制功能
    IDEA安装与配置
    【Ubuntu搭建MQTT Broker及面板+发布消息、订阅主题】
    SpringCLoud——服务的拆分和远程调用
    设计模式行为型-状态模式
    【Linux】linux中你不得不爱的命令集（上）
    线性代数应用基础补充2
    在进行自动化测试，遇到验证码的问题,怎么办？
    XTU-OJ 1108-a+b
  • 原文地址：https://blog.csdn.net/weixin_55252589/article/details/132843450