【Linux】系统编程简单线程池(C++)

目录

【1】线程池概念

【1.1】线程池

【1.2】线程池的应用场景

【1.3】线程池的种类

【1.4】线程池示例

【2】线程池代码

---

【1】线程池概念

【1.1】线程池

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

【1.2】线程池的应用场景

• 需要大量的线程来完成任务，且完成任务的时间比较短。 WEB服务器完成网页请求这样的任务，使用线程池技术是非常合适的。因为单个任务小，而任务数量巨大，你可以想象一个热门网站的点击次数。 但对于长时间的任务，比如一个Telnet连接请求，线程池的优点就不明显了。因为Telnet会话时间比线程的创建时间大多了。

• 对性能要求苛刻的应用，比如要求服务器迅速响应客户请求。

• 接受突发性的大量请求，但不至于使服务器因此产生大量线程的应用。突发性大量客户请求，在没有线程池情况下，将产生大量线程，虽然理论上大部分操作系统线程数目最大值不是问题，短时间内产生大量线程可能使内存到达极限， 出现错误.

【1.3】线程池的种类

• 半同步/半异步模式

• 领导者/跟随着模式

【1.4】线程池示例

• 创建固定数量线程池，循环从任务队列中获取任务对象，获取到任务对象后，执行任务对象中的任务接口。

【2】线程池代码

【Makefile文件】

# 创建变量关联关系
cc=c++
standard=-std=c++11
linkLib=-l pthread
​
# 创建编译文件依赖关系
myThreadPool:ThreadPool.cc$(cc) -o $@ $^ $(standard) $(linkLib)  
​
# 创建删除命令
.PHONY:clean
clean:rm -f myThreadPool

【Task.hpp文件】

#pragma once
#include <iostream>
#include <string>
#include <functional>
​
/* 计算任务 */
class CalTask
{
public:using func_t = std::function<int(int, int, char)>;
​
public:/* 构造函数 */CalTask() {}
​/* 构造函数 */CalTask(int x, int y, char op, func_t func): _x(x), _y(y), _op(op), _callBalk(func){}
​
public:/* 仿函数 */std::string operator()(){int result = _callBalk(_x, _y, _op);
​char buffer[64];snprintf(buffer, sizeof(buffer), "%d %c %d = %d", _x, _op, _y, result);return buffer;}
​
public:/* 返回打印公式 */std::string ToTaskString(){char buffer[64];snprintf(buffer, sizeof(buffer), "%d %c %d = ？", _x, _op, _y);return buffer;}
​
private:int _x;int _y;char _op;func_t _callBalk;
};
​
/* 执行计算的方法 */
int MyCalculate(int x, int y, char op)
{int result = 0;switch (op){case '+':result = x + y;break;
​case '-':result = x - y;break;
​case '*':result = x * y;break;
​case '/':{if (y == 0){std::cerr << "div zero error！" << std::endl;result = -1;}else{result = x / y;}break;}
​case '%':{if (y == 0){std::cerr << "mod zero error！" << std::endl;result = -1;}else{result = x % y;}break;}
​default:break;}
​return result;
}
​
/* 保存任务 */
class SaveTask
{
public:using func_t = std::function<void(const std::string &)>;
​
public:/* 构造函数 */SaveTask() {}/* 构造函数 */SaveTask(const std::string &message, func_t func): _message(message), _callBalk(func){}
​
public:/* 仿函数 */void operator()(){_callBalk(_message);}
​
private:std::string _message;func_t _callBalk;
};
​
/* 保存方法 */
void Save(const std::string& massage){std::string target = "./log.txt";FILE *fp = fopen(target.c_str(), "a+");if(fp == NULL){std::cerr << "fopen fail!" << std::endl;return;}
​fputs(massage.c_str(), fp);fputs("\n", fp);fclose(fp);
}

【Thread.hpp文件】

#pragma once
#include <iostream>
#include <functional>
#include <cstdio>
#include <cassert>
​
namespace ThreadNs
{/* 线程连接上下文 */class Thread;class ConnectText{public:/* 构造函数 */ConnectText() : _this(nullptr), _args(nullptr) {}/* 析构函数 */~ConnectText() {}
​public:Thread *_this;void *_args;};
​
​/* 线程类封装 */class Thread{public:using func_t = std::function<void *(void *)>; // // 从定义类似函数指针类型：返回值是：void*  参数是：void*
​public:/* 构造函数 */Thread(const int number = 0){// 创建线程的名称char buffer[64];snprintf(buffer, sizeof(buffer), "Thread-%d", number);_tName = buffer;}
​/* 析构函数 */~Thread() {}public:/* 线程启动 */void Start(const func_t& func, void *args = nullptr){// 建立关系_func = func;_args = args;
​// 创建线程后，启动ConnectText *cnt = new ConnectText();cnt->_this = this;cnt->_args = _args;
​int n = pthread_create(&_tid, nullptr, StartRoutine, (void *)cnt);assert(n == 0); (void)n;       // 编译debug的方式时assert是存在的，release方式assert是不存在的，到时n就是定义了，但是没有被使用的变量。// 在有的编译器下会有warning。}
​/* 线程等待 */void Join(){int n = pthread_join(_tid, nullptr);assert(n == 0);(void)n;}
​public:/* 获取线程名字 */std::string GetThreadName(){return _tName;}
​private:/* 线程函数 */static void *StartRoutine(void *args){ // 在类内创建线程，想让线程执行对应的方法，需要将方法设置称为staticConnectText *cnt = static_cast<ConnectText *>(args);void *exRet = cnt->_this->RoutineRun(cnt->_args);delete cnt;return exRet;}
​void *RoutineRun(void *args){return _func(args);}
​private:pthread_t _tid;     // 线程idstd::string _tName; // 线程名称func_t _func;       // 线程函数void *_args;        // 线程参数};
}

【ThreadPool.hpp文件】

#pragma once
#include <iostream>
#include <vector>
#include <queue>
#include <pthread.h>
#include "Thread.hpp"
#include "LockGuard.hpp"
​
/* 引用命名空间 */
using namespace ThreadNs;
​
/* 线程上下文数据 */
template <class T>
class ThreadPool;
​
template <class T>
class ThreadData
{
public:ThreadData(ThreadPool<T> *tp, const std::string &n): _threadPool(tp), _name(n){}
​
public:ThreadPool<T> *_threadPool;std::string _name;
};
​
​
​
/* 线程池封装 */
const int g_num = 5;    // 设置线程数
template <class T>
class ThreadPool
{
public:/* 构造函数 */ThreadPool(const int &num = g_num): _threadNum(num){// 初始化线程锁pthread_mutex_init(&_mutex, nullptr);// 初始化线程条件变量pthread_cond_init(&_cond, nullptr);
​// 将每个线程地址Push到线程地址容器中for (int i = 0; i < _threadNum; i++){_threads.push_back(new Thread(i + 1));}}
​/* 析构函数 */~ThreadPool(){// 释放线程锁pthread_mutex_destroy(&_mutex);// 释放线程条件变量pthread_cond_destroy(&_cond);
​// 遍历释放for (auto &t : _threads){delete t;}}
​
public:/* 开启线程池 */void Run(){// 启动线程for (const auto &t : _threads){ThreadData<T> *td = new ThreadData<T>(this, t->GetThreadName());t->Start(HandlerTask, td);std::cout << t->GetThreadName() << ": Start..." << std::endl;}
​// // 阻塞式等待回收线程// for (const auto &t : _threads)// {//     t->Join();//     std::cout << t->GetThreadName() << ": Recycle..." << std::endl;// }}
​
public:/* 获取锁 */pthread_mutex_t *Mutex() { return &_mutex; }/* 线程加锁 */void Lock() { pthread_mutex_lock(&_mutex); }/* 线程解锁 */void UnLock() { pthread_mutex_unlock(&_mutex); }/* 线程等待 */void ThreadWait() { pthread_cond_wait(&_cond, &_mutex); }/* 判断是否有任务 */bool IsEmpty() { return _taskQueue.empty(); }
​/* 新增任务 */void Push(const T &in){// 加锁->自动解锁LockGuard LockGuard(&_mutex);// 新增任务_taskQueue.push(in);// 环形线程执行pthread_cond_signal(&_cond);}
​/* 执行任务 */T Pop(){// 创建T对象T t;// 获取栈顶任务t = _taskQueue.front();_taskQueue.pop();// 返回任务return t;}
​
private:/* 线程池共享线程 */static void *HandlerTask(void *args){ThreadData<T> *td = static_cast<ThreadData<T> *>(args);
​while (true){// 创建任务对象T t;
​{// 加锁->自动解锁LockGuard LockGuard(td->_threadPool->Mutex());// 如果没有任务，等待任务if (td->_threadPool->IsEmpty()){td->_threadPool->ThreadWait();}t = td->_threadPool->Pop();// 解锁td->_threadPool->UnLock();}
​// 打印信息std::cout << td->_name << "：承接一个任务[" << t.ToTaskString() << "],任务的执行结果是[" << t() << "]" << std::endl;}return nullptr;}
​
private:int _threadNum;                 // 线程数量std::vector<Thread *> _threads; // 存放线程地址的容器std::queue<T> _taskQueue;       // 存放线程任务的队列pthread_mutex_t _mutex;         // 线程锁pthread_cond_t _cond;           // 线程条件变量
};

【ThreadPool.cc文件】

#include <iostream>
#include <memory>
#include <unistd.h>
#include "Task.hpp"
#include "ThreadPool.hpp"
using namespace std;
​
int main() 
{// 智能指针管理unique_ptr<ThreadPool<CalTask>> tP(new ThreadPool<CalTask>());// 启动线程池tP->Run();
​// 手动派发任务int x;int y;char op;while (1){std::cout << "请输入数据1# ";std::cin >> x;std::cout << "请输入数据2# ";std::cin >> y;std::cout << "请输入要进行的运算# ";std::cin >> op;CalTask t(x, y, op, MyCalculate);tP->Push(t);sleep(1);}return 0;
}

【代码测试结果】