devbox
天景科技苑
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

【Linux】-- 线程池_linux 线程池例子

作者：天景科技苑 | 2024-08-07 22:43:21

赞

踩

linux 线程池例子

目录

铺垫

内存

线程的角度

线程池

基本代码结构

对于线程池的生产消费的完善

初步实现线程池生产消费

结合日志完善线程池

铺垫

内存

（以STL处理方式，引入提供效率的一种思想）

        通过进行C语言与C++语言的学习中，平时我们使用的内存，都是我们需要了，才进行申请。而又通过系统的学习，我们可以发现，需要才进行申请，这样语言上我们要调用语言的接口：malloc、new等，但是它们的底层也一定是要调用操作系统提供的接口的。而调用系统接口，需要：

  • 陷入内核。
  • 更改CPU状态。
  • 切换页表。
  • 空间不够，操作系统还需要进行查找（空间碎片），进行其内部的内存管理算法（刷新缓冲区进行IO腾出空间、对内存碎片做整理、杀掉不常用的应用）。

        所以，对应的STL容器，对于此方面：增容不是我们要多少给多少，而是直接预先多申请一部分，本质上就是用空间换时间的策略。

线程的角度

        内存是资源，线程本质是一种调用执行流，也是资源。所以如果我们预先创建出一大批的线程，当有任务在来的时候，我们直接让线程去进行处理任务，而不用再去当任务到来时，然后才进行创建线程（本质创建各种数据结构，并且进行初始化、以及一系列为管理的操作），这是很浪费时间的。

        如果我们预先创建出一大批的线程，当有任务来的时候，然后直接让线程去处理任务，而不用再去当任务来的时候，才创建……。

线程池

        一种线程使用模式。线程过多会带来调度开销，进而影响缓存局部性和整体性能。而线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利用，还能防止过分调度。可用线程数量应该取决于可用的并发处理器、处理器内核、内存、网络sockets等的数量。

线程池的应用场景：

  1. 需要大量的线程来完成任务，且完成任务的时间比较短。 WEB服务器完成网页请求这样的任务，使用线程池技术是非常合适的。因为单个任务小，而任务数量巨大，你可以想象一个热门网站的点击次数。 但对于长时间的任务，比如一个Telnet连接请求，线程池的优点就不明显了。因为Telnet会话时间比线程的创建时间大多了。
  2. 对性能要求苛刻的应用，比如要求服务器迅速响应客户请求。
  3. 接受突发性的大量请求，但不至于使服务器因此产生大量线程的应用。突发性大量客户请求，在没有线程池情况下，将产生大量线程，虽然理论上大部分操作系统线程数目最大值不是问题，短时间内产生大量线程可能使内存到达极限。
线程池示例：
  1. 创建固定数量线程池，循环从任务队列中获取任务对象。
  2. 获取到任务对象后，执行任务对象中的任务接口。

Note：

        核心：用来预先申请资源，达到以空间换时间的一种做法。

线程池的框架实现：

        有人向线程池内部push任务，线程池内部的线程就可以自动的处理任务。即：有线程向线程池内部push任务，就一定回有线程在线程池内部拿任务 —— 本质：生产消费模型。

基本代码结构

        并未加入互斥量与信号量。

thread.hpp

        对线程的封装。

  1. #pragma once
  2. #include <string>
  3. #include <pthread.h>
  4. #include <cstdio>
  5.  
  6. // 对线程的封装 - 不是完全必要，但是这样便于后期的统一管理
  7. typedef void*(*fun_t)(void*);
  8.  
  9. // 整合线程的数据
  10. class ThreadData
  11. {
  12. public:
  13.     std::string _name;
  14.     void* _args;
  15. };
  16.  
  17. class Thread
  18. {
  19. public:
  20.     Thread(int num, fun_t callback, void* args):_func(callback)
  21.     {
  22.         char nameBuffer[64];
  23.         snprintf(nameBuffer, sizeof(nameBuffer), "Thread-%d", num);
  24.         _name = nameBuffer;
  25.  
  26.         _tdata._args = args;
  27.         _tdata._name = _name;
  28.     }
  29.  
  30.     void start()
  31.     {
  32.         pthread_create(&_tid, nullptr, _func, (void*)&_tdata);
  33.     }
  34.  
  35.     void join()
  36.     {
  37.         pthread_join(_tid, nullptr);
  38.     }
  39.  
  40.     // 未来不再使用线程id了，因为其是一个地址不便于我们查看
  41.     std::string name()
  42.     {
  43.         return _name;
  44.     }
  45.     
  46.     ~Thread()
  47.     {}
  48.  
  49. private:
  50.    std::string _name;
  51.    fun_t _func;
  52.    ThreadData _tdata;
  53.    pthread_t _tid;
  54. };

ThreadPool.hpp 

  1. #include "thread.hpp"
  2. #include <vector>
  3. #include <queue>
  4. #include <unistd.h>
  5. #include <iostream>
  6.  
  7. const int g_thread_num = 3;
  8.  
  9. template<class T>
  10. class ThreadPool
  11. {
  12. public:
  13.     ThreadPool(int thread_num = g_thread_num):_num(thread_num)
  14.     {
  15.         for(int i = 1; i <= _num; i++)
  16.             _threads.push_back(new Thread(i, routine, nullptr));
  17.     }
  18.  
  19.     // 1.run - 将线程跑起来
  20.     void run()
  21.     {
  22.         int i = 0;
  23.         for(auto& iter : _threads)
  24.         {
  25.             iter->start();
  26.             std::cout << iter->name() << " 启动成功" << std::endl;
  27.         }
  28.     }
  29.  
  30.     // 未来所有执行流所执行的方法 - 核心取任务、执行任务的逻辑
  31.     static void* routine(void* args) // 因为在类当中，如果是一个成员方法，其会有一个隐藏的参数this指针，所以我们需要使用static进行修饰。
  32.     {
  33.         ThreadData *td = (ThreadData*)args;
  34.         while(true)
  35.         {
  36.             std::cout << "我是一个线程，我的名字是：" << td->_name << std::endl;
  37.             sleep(1);
  38.         }
  39.     }
  40.  
  41.     // 2.pushTask - 将任务放到任务池里
  42.     void pushTask()
  43.     {}
  44.  
  45.     void join()
  46.     {
  47.         for(auto& iter : _threads)
  48.             iter->join();
  49.     }
  50.  
  51.     ~ThreadPool()
  52.     {
  53.         for(auto& iter : _threads)
  54.             delete iter;
  55.     }
  56.  
  57. private:
  58.     std::vector<Thread*> _threads;
  59.     int _num;
  60.     std::queue<T> _task_queue;
  61. };

testMain.cc 

  1. #include "threadPool.hpp"
  2. #include <iostream>
  3.  
  4. int main()
  5. {
  6.     ThreadPool<int> *tp = new ThreadPool<int>();
  7.     tp->run();
  8.     tp->join();
  9.     return 0;
  10. }

对于线程池的生产消费的完善

        加入互斥量与信号量，并且加入生产者的生产数据、方法，消费者的消费数据、方法。此处我们需要注意：最难的并不是生产，而是消费。

生产者：

        生产的过程。

  1. // 2.pushTask - 将任务放到任务池里
  2. void pushTask(const T& task)
  3. {
  4.     lockGuard lockguard(&_lock); // 加锁
  5.     _task_queue.push(task); // 压入任务
  6.     pthread_cond_signal(&_cond); // 唤醒线程
  7. } // 自动释放锁

消费者：

        消费的过程。

原：

  1. // 未来所有执行流所执行的方法 - 核心取任务、执行任务的逻辑
  2. static void* routine(void* args) // 因为在类当中，如果是一个成员方法，其会有一个隐藏的参数this指针，所以我们需要使用static进行修饰。
  3. {
  4.     ThreadData *td = (ThreadData*)args;
  5.     while(true)
  6.     {
  7.         std::cout << "我是一个线程，我的名字是：" << td->_name << std::endl;
  8.         sleep(1);
  9.     }
  10. }

        因为，每一个线程创建出来了，它都会执行routine，并且我们是在构造的时候就将其传送给了我们对应的线程，该线程就会在start的时候就会进入我们设置的回调。

  1. //ThreadPool.hpp 
  2. //---------------------------
  3. // 构造函数
  4. ThreadPool(int thread_num = g_thread_num):_num(thread_num)
  5. {
  6.     pthread_mutex_init(&_lock, nullptr);
  7.     pthread_cond_init(&_cond, nullptr);
  8.     for(int i = 1; i <= _num; i++)
  9.         _threads.push_back(new Thread(i, routine, nullptr));
  10. }
  11.  
  12. // 1.run - 将线程跑起来
  13. void run()
  14. {
  15.     for(auto& iter : _threads)
  16.     {
  17.         iter->start();
  18.         std::cout << iter->name() << " 启动成功" << std::endl;
  19.     }
  20. }
  21.  
  22. //Thread.hpp 
  23. //---------------------------
  24. void start()
  25. {
  26.     pthread_create(&_tid, nullptr, _func, (void*)&_tdata);
  27. }

        但是，作为势必要消费数据的routine就带来一些问题（困扰）：消费就代表要访问，存储对应方法的_task_queue，但是_task_queue作为内类的成员方法，而routine是一个被staict修饰的成员方法（其只能够使用静态成员、静态方法，无法使用内类的成员属性、成员方法）。

#：将_task_queue设置为static修饰的

        这个方法是可以的（类内的成员方法，能够使用static修饰的成员），但是是不好的。因为如果定义为static修饰的，未来如果有两个三个等的线程池，这些线程池就相当于_task_queue被大家所共享的，而且对于此类数据还是最好封装住。

        与其让routine拿到_task_queue，不如让其直接拿到整体对象。拿到之后可以直接让其利用函数的方式访问类内的各种属性。

        直接在构造函数的地方传递：this指针。

  1. //ThreadPool.hpp 
  2. //---------------------------
  3. // 构造函数
  4. ThreadPool(int thread_num = g_thread_num):_num(thread_num)
  5. {
  6.     pthread_mutex_init(&_lock, nullptr);
  7.     pthread_cond_init(&_cond, nullptr);
  8.     for(int i = 1; i <= _num; i++)
  9.         _threads.push_back(new Thread(i, routine, this));
  10. }

        此时，对于routine内部只需要强转一下就行了。

  1. // 未来所有执行流所执行的方法 - 核心取任务、执行任务的逻辑
  2. static void* routine(void* args) // 因为在类当中，如果是一个成员方法，其会有一个隐藏的参数this指针，所以我们需要使用static进行修饰。
  3. {
  4.     ThreadData *td = (ThreadData*)args;
  5.     ThreadPool<T>* tp = (ThreadPool<T>*)td->_args;
  6.     while(true)
  7.     {
  8.         // lock
  9.         // while(task_queue_.empty()) wait();
  10.         // 获取任务
  11.         // unlock
  12.  
  13.         // 处理任务
  14.     }
  15. }

初步完善：

  1. public:
  2.     // 返回锁的地址
  3.     pthread_mutex_t *getMutex()
  4.     {
  5.         return &_lock;
  6.     }
  7.  
  8.     bool isEmpty()
  9.     {
  10.         return _task_queue.empty();
  11.     }
  12.  
  13.     void waitCond()
  14.     {
  15.         pthread_cond_wait(&_cond, &_lock);
  16.     }
  17.  
  18.     T getTask()
  19.     {
  20.         T t = _task_queue.front();
  21.         _task_queue.pop();
  22.         return t;
  23.     }
  24. ------------------------
  25.  
  26. // 未来所有执行流所执行的方法 - 核心取任务、执行任务的逻辑
  27. static void *routine(void *args) // 因为在类当中，如果是一个成员方法，其会有一个隐藏的参数this指针，所以我们需要使用static进行修饰。
  28. {
  29.     ThreadData *td = (ThreadData *)args;
  30.     ThreadPool<T> *tp = (ThreadPool<T> *)td->_args;
  31.     while (true)
  32.     {
  33.         T task; // 对应任务的对象
  34.  
  35.         // 利用{}确定加锁的区间
  36.         {
  37.             // lock
  38.             lockGuard lockguard(tp->getMutex());
  39.  
  40.             // while(task_queue_.empty()) wait();
  41.             while (tp->isEmpty())
  42.                 tp->waitCond();
  43.  
  44.             // 获取任务 - 100%有任务
  45.             task = tp->getTask(); // 任务队列是共享的->将任务从共享，拿到自己的私有空间
  46.  
  47.         } // 自动释放锁
  48.  
  49.         // 处理任务
  50.         task(); // 要求每一个任务都要提供一个仿函数
  51.     }
  52. }

初步实现线程池生产消费

提供一个：Task.hpp

        任务类型的封装，重点：包含仿函数。

  1. #pragma once
  2.  
  3. #include <iostream>
  4. #include <string>
  5. #include <functional>
  6.  
  7. typedef std::function<int(int, int)> func_t;
  8.  
  9. class Task
  10. {
  11. public:
  12.     Task(){}
  13.     Task(int x, int y, func_t func):_x(x), _y(y), _func(func)
  14.     {}
  15.  
  16.     int operator ()()
  17.     {
  18.         return _func(_x, _y);
  19.     }
  20. public:
  21.     int _x;
  22.     int _y;
  23.     
  24.     // int type;
  25.     func_t _func;
  26. };

        任务是什么样子，不能够让线程池知道，与线程池没有关系，线程池只需要我们让其干什么就干什么就行了，也就是一定意义上的解耦合。

thread.hpp

  1. #pragma once
  2. #include <string>
  3. #include <pthread.h>
  4. #include <cstdio>
  5.  
  6. // 对线程的封装 - 不是完全必要，但是这样便于后期的统一管理
  7.  
  8. typedef void*(*fun_t)(void*);
  9.  
  10. // 整合线程的数据
  11. class ThreadData
  12. {
  13. public:
  14.     std::string _name;
  15.     void* _args;
  16. };
  17.  
  18. class Thread
  19. {
  20. public:
  21.     Thread(int num, fun_t callback, void* args):_func(callback)
  22.     {
  23.         char nameBuffer[64];
  24.         snprintf(nameBuffer, sizeof(nameBuffer), "Thread-%d", num);
  25.         _name = nameBuffer;
  26.  
  27.         _tdata._args = args;
  28.         _tdata._name = _name;
  29.     }
  30.  
  31.     void start()
  32.     {
  33.         pthread_create(&_tid, nullptr, _func, (void*)&_tdata);
  34.     }
  35.  
  36.     void join()
  37.     {
  38.         pthread_join(_tid, nullptr);
  39.     }
  40.  
  41.     // 未来不再使用线程id了，因为其是一个地址不便于我们查看
  42.     std::string name()
  43.     {
  44.         return _name;
  45.     }
  46.     
  47.     ~Thread()
  48.     {}
  49.  
  50. private:
  51.    std::string _name;
  52.    fun_t _func;
  53.    ThreadData _tdata;
  54.    pthread_t _tid;
  55. };

lockGuard.hpp

  1. #pragma once
  2.  
  3. #include <iostream>
  4. #include <pthread.h>
  5.  
  6. // RAII风格的加锁方式
  7. class lockGuard
  8. {
  9. public:
  10.     lockGuard(pthread_mutex_t *mtx):mtx_(mtx)
  11.     {
  12.         pthread_mutex_lock(mtx_);
  13.  
  14.     }
  15.     ~lockGuard()
  16.     {
  17.         pthread_mutex_unlock(mtx_);
  18.     }
  19.  
  20. private:
  21.     pthread_mutex_t *mtx_;
  22. };

Task.hpp

  1. #pragma once
  2.  
  3. #include <iostream>
  4. #include <string>
  5. #include <functional>
  6.  
  7. typedef std::function<int(int, int)> func_t;
  8.  
  9. class Task
  10. {
  11. public:
  12.     Task(){}
  13.     Task(int x, int y, func_t func):_x(x), _y(y), _func(func)
  14.     {}
  15.  
  16.     void operator()(std::string& name)
  17.     {
  18.         std::cout << "线程" << name << "处理完成，结果是" << _x << "+" << _y << "=" << _func(_x, _y) << std::endl;
  19.     }
  20. public:
  21.     int _x;
  22.     int _y;
  23.  
  24.     // int type;
  25.     func_t _func;
  26. };

threadPool.hpp 

  1. #include "thread.hpp"
  2. #include "lockGuard.hpp"
  3. #include <vector>
  4. #include <queue>
  5. #include <unistd.h>
  6. #include <iostream>
  7.  
  8. const int g_thread_num = 3;
  9.  
  10. template <class T>
  11. class ThreadPool
  12. {
  13. public:
  14.     // 返回锁的地址
  15.     pthread_mutex_t *getMutex()
  16.     {
  17.         return &_lock;
  18.     }
  19.  
  20.     bool isEmpty()
  21.     {
  22.         return _task_queue.empty();
  23.     }
  24.  
  25.     void waitCond()
  26.     {
  27.         pthread_cond_wait(&_cond, &_lock);
  28.     }
  29.  
  30.     T getTask()
  31.     {
  32.         T t = _task_queue.front();
  33.         _task_queue.pop();
  34.         return t;
  35.     }
  36.  
  37. public:
  38.     ThreadPool(int thread_num = g_thread_num) : _num(thread_num)
  39.     {
  40.         for (int i = 1; i <= _num; i++)
  41.         {
  42.             _threads.push_back(new Thread(i, routine, this));
  43.         }
  44.         pthread_mutex_init(&_lock, nullptr);
  45.         pthread_cond_init(&_cond, nullptr);
  46.     }
  47.  
  48.     // 1.run - 将线程跑起来
  49.     void run()
  50.     {
  51.         for (auto &iter : _threads)
  52.         {
  53.             iter->start();
  54.             std::cout << iter->name() << " 启动成功" << std::endl;
  55.         }
  56.     }
  57.  
  58.     // 未来所有执行流所执行的方法 - 核心取任务、执行任务的逻辑
  59.     static void *routine(void *args) // 因为在类当中，如果是一个成员方法，其会有一个隐藏的参数this指针，所以我们需要使用static进行修饰。
  60.     {
  61.         ThreadData *td = (ThreadData *)args;
  62.         ThreadPool<T> *tp = (ThreadPool<T> *)td->_args;
  63.         while (true)
  64.         {
  65.             T task; // 对应任务的对象
  66.             {
  67.                 // lock
  68.                 lockGuard lockguard(tp->getMutex());
  69.  
  70.                 // while(task_queue_.empty()) wait();
  71.                 while (tp->isEmpty())
  72.                     tp->waitCond();
  73.  
  74.                 // 获取任务 - 100%有任务
  75.                 task = tp->getTask(); // 任务队列是共享的->将任务从共享，拿到自己的私有空间
  76.  
  77.             } // 自动释放锁
  78.  
  79.             // 处理任务
  80.             task(td->_name); // 要求每一个任务都要提供一个仿函数
  81.         }
  82.     }
  83.  
  84.     // 2.pushTask - 将任务放到任务池里
  85.     void pushTask(const T &task)
  86.     {
  87.         lockGuard lockguard(&_lock); // 加锁
  88.         _task_queue.push(task);      // 压入任务
  89.         pthread_cond_signal(&_cond); // 唤醒线程
  90.     }                                // 自动释放锁
  91.  
  92.     // void join()
  93.     // {
  94.     //     for (auto &iter : _threads)
  95.     //     {
  96.     //         iter->join();
  97.     //     }
  98.     // }
  99.  
  100.     ~ThreadPool()
  101.     {
  102.         for (auto &iter : _threads)
  103.         {
  104.             iter->join();
  105.             delete iter;
  106.         }
  107.         pthread_mutex_destroy(&_lock);
  108.         pthread_cond_destroy(&_cond);
  109.     }
  110.  
  111. private:
  112.     std::vector<Thread *> _threads;
  113.     int _num;
  114.     std::queue<T> _task_queue;
  115.     pthread_mutex_t _lock;
  116.     pthread_cond_t _cond;
  117. };

 testMain.cc

  1. #include "threadPool.hpp"
  2. #include "Task.hpp"
  3. #include <ctime>
  4. #include <iostream>
  5.  
  6. int main()
  7. {
  8.     srand((unsigned long)time(nullptr) ^ getpid()); // 利用 ^ getpid()让数据更随机
  9.     ThreadPool<Task> *tp = new ThreadPool<Task>();
  10.     tp->run();
  11.  
  12.     while (true)
  13.     {
  14.         // 生产的过程（制作任务的时候，要花时间）
  15.         int x = rand() % 1000;
  16.         usleep(5000); // 模拟制作任务的时候，花费的时间
  17.         int y = rand() % 1000;
  18.  
  19.         Task t(x, y, [](int x, int y)->int{
  20.             return x + y;
  21.         });
  22.  
  23.         std::cout << "制作任务完成" << std::endl;
  24.         // 推送任务到线程池中
  25.         tp->pushTask(t);
  26.  
  27.         sleep(1); // 防止刷新过快
  28.     }
  29.     
  30.     return 0;
  31. }

将上述代码进行优化：

进阶：

  1. // 方案2:
  2. //  queue1,queue2
  3. //  std::queue<T> *p_queue, *c_queue
  4. //  p_queue->queue1
  5. //  c_queue->queue2
  6. //  p_queue -> 生产一批任务之后，swap(p_queue,c_queue),唤醒所有线程/一个线程
  7. //  当消费者处理完毕的时候，你也可以进行swap(p_queue,c_queue)
  8. //  因为我们生产和消费用的是不同的队列，未来我们要进行资源的处理的时候，仅仅是指针
  • 需要加锁的仅仅是swap的地方
  • swap方式：
  1. 满了就swap。
  2. 让一个线程不断地去观察，消费完了，且生产的数据就交换。

        对于上述的多处打印信息的逻辑，其实在正常的时候是需要封装成日志的。一般在公司内部是，任何一款软件，都必须有日志：

        日志中因为要实现，至少：日志等级、时间、日志内容、支持用户自定义。所以参数的传递中，我们需要利用到可变参数列表。

补充：

        可变参数表的使用。

参数：

  1. #include <stdarg.h>
  2.  
  3. void va_start(va_list ap, last);
  4. type va_arg(va_list ap, type);
  5. void va_end(va_list ap);
  6. void va_copy(va_list dest, va_list src);
  • va_list：本质上一个char类型的指针。
  • last：代表可变参数的前面的参数当中的最后一个具体参数。
  • type：具体的类型。
  1. // 完整的日志功能，至少：日志等级 时间 日志内容 支持用户自定义
  2. void logMessage(int level, const char* format, ...)// level：日志等级，format, ...：用户传参、日志对应的信息等。
  3. {
  4.     va_list ap; // va_list本质上就是char类型的指针
  5.     va_start(ap, format); // 让指针指向栈帧对应的结构
  6.     int x = va_arg(ap, int); // 通过具体的类型，来从通过指针提取特定的值 —— 没有参数就返回NULL
  7.  
  8.     va_end(ap); // 将指针设置为空（相当于：ap = nullptr）
  9.  
  10. }

        但是，如果让我们自己提取，那也就太麻烦了，而且使用起来也不舒服。我们可以使用：#include <stdarg.h>中的。

  1. #include <stdarg.h>
  2. // 将我们传入的参数，可变的方式，格式化显示到：
  3.  
  4. // 显示器
  5. int vprintf(const char *format, va_list ap);
  6. // 文件
  7. int vfprintf(FILE *stream, const char *format, va_list ap);
  8. // 支付串
  9. int vsprintf(char *str, const char *format, va_list ap);
  10. // 指定长度的字符串
  11. int vsnprintf(char *str, size_t size, const char *format, va_list ap);

        可以发现它们是与图中的上面四个，#include <stdio.h>中的类似的。

log.hpp：

  1. #pragma once
  2. #include <cstdarg>
  3.  
  4. // 日志是有日志级别的
  5. #define NORMAL 1 // 正常
  6. #define WARNING 2 // 警告 -- 没出错
  7. #define ERROR 3 // 错误 -- 不影响后续执行（一个功能因为条件等，没有执行）
  8. #define FATAL 4 // 致命 -- 代码无法继续向后执行
  9.  
  10. // 完整的日志功能，至少：日志等级 时间 日志内容 支持用户自定义
  11. void logMessage(int level, const char* format, ...)// level：日志等级; format, ...：用户传参、日志对应的信息等。
  12. {
  13.     va_list args;
  14.     va_start(args, format);
  15.     // 这个时候就有一个可变参数列表的起始地址
  16.  
  17.     vprintf(format, args);
  18.  
  19.     va_end(args);
  20. }

testMain.cc：

  1. #include "log.hpp"
  2. #include <iostream>
  3.  
  4. int main()
  5. {
  6.     logMessage(NORMAL, "%s %d %c %f\n", "这是一条日志信息", 1234, 'a', 3.14);
  7.     return 0;
  8. }

结合日志完善线程池

log.hpp

  1. #pragma once
  2. #include <cstdarg>
  3. #include <ctime>
  4.  
  5. // 日志是有日志级别的
  6. #define DEBUG   0
  7. #define NORMAL  1 // 正常
  8. #define WARNING 2 // 警告 -- 没出错
  9. #define ERROR   3 // 错误 -- 不影响后续执行（一个功能因为条件等，没有执行）
  10. #define FATAL   4 // 致命 -- 代码无法继续向后执行
  11.  
  12. const char* gLevelMap[] = {
  13.     "DEBUG",
  14.     "NORMAL",
  15.     "WARNING",
  16.     "ERROR",
  17.     "FATAL",
  18. };
  19.  
  20. #define LOGFILE "./threafpool.log"
  21.  
  22. // 完整的日志功能，至少：日志等级 时间 日志内容 支持用户自定义
  23. void logMessage(int level, const char* format, ...)// level：日志等级; format, ...：用户传参、日志对应的信息等。
  24. {
  25. #ifndef DEBUG_SHOW
  26.     if(level == DEBUG) return;
  27. #endif
  28.     char stdBuffer[1024]; //标准部分
  29.     time_t timestamp = time(nullptr);
  30.  
  31.     //struct tm* localtime = localtime(&timestamp); //- 详细时间输出操作
  32.     //localtime->tm_year;
  33.  
  34.     snprintf(stdBuffer, sizeof stdBuffer, "[%s] [%ld] ", gLevelMap[level], timestamp);
  35.  
  36.     char logBuffer[1024]; //自定义部分
  37.     va_list args;
  38.     va_start(args, format);
  39.     // 这个时候就有一个可变参数列表的起始地址
  40.  
  41.     // 向屏幕中直接打印
  42.     // vprintf(format, args);
  43.  
  44.     // 向缓冲区logBuffer中打印
  45.     vsnprintf(logBuffer, sizeof logBuffer, format, args);
  46.     va_end(args);
  47.  
  48.     // 向屏幕
  49.     printf("%s%s\n", stdBuffer, logBuffer);
  50.  
  51.     // 向文件
  52.     FILE *fp = fopen(LOGFILE, "a");
  53.     fprintf(fp, "%s%s\n", stdBuffer, logBuffer);
  54.     fclose(fp);
  55. }

Task.hpp

  1. #pragma once
  2.  
  3. #include "log.hpp"
  4. #include <iostream>
  5. #include <string>
  6. #include <functional>
  7.  
  8. typedef std::function<int(int, int)> func_t;
  9.  
  10. class Task
  11. {
  12. public:
  13.     Task(){}
  14.     Task(int x, int y, func_t func):_x(x), _y(y), _func(func)
  15.     {}
  16.  
  17.     void operator()(std::string& name)
  18.     {
  19.         //std::cout << "线程" << name << "处理完成，结果是" << _x << "+" << _y << "=" << _func(_x, _y) << std::endl;
  20.         logMessage(WARNING, "%s处理完成: %d+%d=%d | %s | %d", 
  21.         name.c_str(), _x, _y, _func(_x, _y), __FILE__, __LINE__); // __FILE__, __LINE__：预处理符。
  22.     }
  23. public:
  24.     int _x;
  25.     int _y;
  26.  
  27.     // int type;
  28.     func_t _func;
  29. };

预处理符： 

        这就是为什么要有这些预处理符，其一的原因就是日志中会使用到，便于我们找到文件的执行位置。

  1. __FILE__      //进行编译的源文件
  2. __LINE__     //文件当前的行号
  3. __DATE__    //文件被编译的日期
  4. __TIME__    //文件被编译的时间
  5. __STDC__    //如果编译器遵循ANSI C，其值为1，否则未定义

thread.hpp

  1. #pragma once
  2. #include <string>
  3. #include <pthread.h>
  4. #include <cstdio>
  5.  
  6. // 对线程的封装 - 不是完全必要，但是这样便于后期的统一管理
  7.  
  8. typedef void*(*fun_t)(void*);
  9.  
  10. // 整合线程的数据
  11. class ThreadData
  12. {
  13. public:
  14.     std::string _name;
  15.     void* _args;
  16. };
  17.  
  18. class Thread
  19. {
  20. public:
  21.     Thread(int num, fun_t callback, void* args):_func(callback)
  22.     {
  23.         char nameBuffer[64];
  24.         snprintf(nameBuffer, sizeof(nameBuffer), "Thread-%d", num);
  25.         _name = nameBuffer;
  26.  
  27.         _tdata._args = args;
  28.         _tdata._name = _name;
  29.     }
  30.  
  31.     void start()
  32.     {
  33.         pthread_create(&_tid, nullptr, _func, (void*)&_tdata);
  34.     }
  35.  
  36.     void join()
  37.     {
  38.         pthread_join(_tid, nullptr);
  39.     }
  40.  
  41.     // 未来不再使用线程id了，因为其是一个地址不便于我们查看
  42.     std::string name()
  43.     {
  44.         return _name;
  45.     }
  46.     
  47.     ~Thread()
  48.     {}
  49.  
  50. private:
  51.    std::string _name;
  52.    fun_t _func;
  53.    ThreadData _tdata;
  54.    pthread_t _tid;
  55. };

lockGuard.hpp

  1. #pragma once
  2.  
  3. #include <iostream>
  4. #include <pthread.h>
  5.  
  6. // RAII风格的加锁方式
  7. class lockGuard
  8. {
  9. public:
  10.     lockGuard(pthread_mutex_t *mtx):mtx_(mtx)
  11.     {
  12.         pthread_mutex_lock(mtx_);
  13.  
  14.     }
  15.     ~lockGuard()
  16.     {
  17.         pthread_mutex_unlock(mtx_);
  18.     }
  19.  
  20. private:
  21.     pthread_mutex_t *mtx_;
  22. };

threadPool.hpp

  1. #include "thread.hpp"
  2. #include "lockGuard.hpp"
  3. #include "log.hpp"
  4. #include <vector>
  5. #include <queue>
  6. #include <unistd.h>
  7. #include <iostream>
  8.  
  9. const int g_thread_num = 3;
  10.  
  11. template <class T>
  12. class ThreadPool
  13. {
  14. public:
  15.     // 返回锁的地址
  16.     pthread_mutex_t *getMutex()
  17.     {
  18.         return &_lock;
  19.     }
  20.  
  21.     bool isEmpty()
  22.     {
  23.         return _task_queue.empty();
  24.     }
  25.  
  26.     void waitCond()
  27.     {
  28.         pthread_cond_wait(&_cond, &_lock);
  29.     }
  30.  
  31.     T getTask()
  32.     {
  33.         T t = _task_queue.front();
  34.         _task_queue.pop();
  35.         return t;
  36.     }
  37.  
  38. public:
  39.     ThreadPool(int thread_num = g_thread_num) : _num(thread_num)
  40.     {
  41.         for (int i = 1; i <= _num; i++)
  42.         {
  43.             _threads.push_back(new Thread(i, routine, this));
  44.         }
  45.         pthread_mutex_init(&_lock, nullptr);
  46.         pthread_cond_init(&_cond, nullptr);
  47.     }
  48.  
  49.     // 1.run - 将线程跑起来
  50.     void run()
  51.     {
  52.         for (auto &iter : _threads)
  53.         {
  54.             iter->start();
  55.             // std::cout << iter->name() << " 启动成功" << std::endl;
  56.             logMessage(NORMAL, "%s %s", iter->name().c_str(), "启动成功");
  57.         }
  58.     }
  59.  
  60.     // 未来所有执行流所执行的方法 - 核心取任务、执行任务的逻辑
  61.     static void *routine(void *args) // 因为在类当中，如果是一个成员方法，其会有一个隐藏的参数this指针，所以我们需要使用static进行修饰。
  62.     {
  63.         ThreadData *td = (ThreadData *)args;
  64.         ThreadPool<T> *tp = (ThreadPool<T> *)td->_args;
  65.         while (true)
  66.         {
  67.             T task; // 对应任务的对象
  68.             {
  69.                 // lock
  70.                 lockGuard lockguard(tp->getMutex());
  71.  
  72.                 // while(task_queue_.empty()) wait();
  73.                 while (tp->isEmpty())
  74.                     tp->waitCond();
  75.  
  76.                 // 获取任务 - 100%有任务
  77.                 task = tp->getTask(); // 任务队列是共享的->将任务从共享，拿到自己的私有空间
  78.  
  79.             } // 自动释放锁
  80.  
  81.             // 处理任务
  82.             task(td->_name); // 要求每一个任务都要提供一个仿函数
  83.         }
  84.     }
  85.  
  86.     // 2.pushTask - 将任务放到任务池里
  87.     void pushTask(const T &task)
  88.     {
  89.         lockGuard lockguard(&_lock); // 加锁
  90.         _task_queue.push(task);      // 压入任务
  91.         pthread_cond_signal(&_cond); // 唤醒线程
  92.     }                                // 自动释放锁
  93.  
  94.     // void join()
  95.     // {
  96.     //     for (auto &iter : _threads)
  97.     //     {
  98.     //         iter->join();
  99.     //     }
  100.     // }
  101.  
  102.     ~ThreadPool()
  103.     {
  104.         for (auto &iter : _threads)
  105.         {
  106.             iter->join();
  107.             delete iter;
  108.         }
  109.         pthread_mutex_destroy(&_lock);
  110.         pthread_cond_destroy(&_cond);
  111.     }
  112.  
  113. private:
  114.     std::vector<Thread *> _threads;
  115.     int _num;
  116.     std::queue<T> _task_queue;
  117.     pthread_mutex_t _lock;
  118.     pthread_cond_t _cond;
  119. };

testMain.cc

  1. #include "threadPool.hpp"
  2. #include "Task.hpp"
  3. #include "log.hpp"
  4. #include <ctime>
  5. #include <iostream>
  6.  
  7. int main()
  8. {
  9.     logMessage(NORMAL, "%s %d %c %f\n", "这是一条日志信息", 1234, 'a', 3.14);
  10.     srand((unsigned long)time(nullptr) ^ getpid()); // 利用 ^ getpid()让数据更随机
  11.     ThreadPool<Task> *tp = new ThreadPool<Task>();
  12.     tp->run();
  13.  
  14.     while (true)
  15.     {
  16.         // 生产的过程（制作任务的时候，要花时间）
  17.         int x = rand() % 1000;
  18.         usleep(5000); // 模拟制作任务的时候，花费的时间
  19.         int y = rand() % 1000;
  20.  
  21.         Task t(x, y, [](int x, int y)->int{
  22.             return x + y;
  23.         });
  24.  
  25.         // std::cout << "制作任务完成" << std::endl;
  26.         logMessage(DEBUG, "制作任务完成: %d+%d=?", x, y);
  27.         
  28.         // 推送任务到线程池中
  29.         tp->pushTask(t);
  30.  
  31.         sleep(1); // 防止刷新过快
  32.     }
  33.     
  34.     return 0;
  35. }

补充：

对于时间信息的提取与输出：

struct tm中所存储的内容： 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/天景科技苑/article/detail/944857
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号