任务队列

头文件

#ifndef __TASKQUEUE__H
#define __TASKQUEUE__H#include<queue>
#include<pthread.h>
using namespace std;
using callback = void(*)(void* arg);template<typename T>
struct Task
{Task<T>(){function = nullptr;arg = nullptr;}Task<T>(callback f, void* arg){this->arg = (T*)arg;this->function = f;}callback function;T* arg;
};template<typename T>
class TaskQueue
{
public:TaskQueue();~TaskQueue();//添加任务void addTask(Task<T> task);void addTask(callback f, void* arg);// 取出任务Task<T> takeTask();// 获取当前任务个数inline size_t taskNumber(){return m_taskQ.size();}//private:pthread_mutex_t m_mutex;queue<Task<T>> m_taskQ;};#endif // !__TASKQUEUE__H

源文件

#include "TaskQueue.h"template<typename T>
TaskQueue<T>::TaskQueue()
{pthread_mutex_init(&m_mutex, NULL);
}template<typename T>
TaskQueue<T>::~TaskQueue()
{pthread_mutex_destroy(&m_mutex);
}template<typename T>
void TaskQueue<T>::addTask(Task<T> task)
{pthread_mutex_lock(&m_mutex);m_taskQ.push(task);pthread_mutex_unlock(&m_mutex);
}template<typename T>
void TaskQueue<T>::addTask(callback f, void * arg)
{pthread_mutex_lock(&m_mutex);m_taskQ.push(Task<T>(f,arg));pthread_mutex_unlock(&m_mutex);
}template<typename T>
Task<T> TaskQueue<T>::takeTask()
{Task<T> t;pthread_mutex_lock(&m_mutex);//判断任务队列是否为空if (!m_taskQ.empty()){t = m_taskQ.front();//先取出m_taskQ.pop();// 在弹出}pthread_mutex_unlock(&m_mutex);return t;
}

线程池

线程池头文件

#ifndef __THREADPOOL__H
#define __THREADPOOL__H
#include"TaskQueue.h"
#include"TaskQueue.cpp"
#include<functional>template<class T>
class ThreadPool
{
public:// 创建线程池并初始化ThreadPool(int min, int max);// 销毁线程池~ThreadPool();// 给线程池添加任务void addTask(Task<T> task);// 获取线程池中工作的线程的个数int getBusyNum();// 获取线程池中活着的线程个数int getAliveNum();private:/// 工作的线程，消费者线程任务函数static void* worker(void* arg);// 管理者线程任务函数static void* manager(void* arg);// 单个线程退出void threadExit();private:// 任务队列TaskQueue<T>* taskQ; // 任务队列，数组，所以定义指针pthread_t managerID; // 管理者线程IDpthread_t *threadIDs; // 工作的线程IDint minNum;               // 最少线程数int maxNum;             // 最大线程数int busyNum;            // 在忙中的线程数int liveNum;          // 存活线程数int exitNum;            // 需要杀死的线程数pthread_mutex_t mutexPool;   // 锁整个线程池pthread_cond_t notEmpty;   // 任务队列是不是空了// function<void*(void*)> workerFunc = worker;static const int NUMBER = 2;bool shutDown;        // 是不是需要销毁线程池，销毁为1，不销毁为0
};#endif // !__THREADPOOL__H

线程池源文件

#include "ThreadPool.h"
#include<iostream>
#include<string.h>
#include<string>
#include<unistd.h>
#include<pthread.h>
using namespace std;template<class T>
ThreadPool<T>::ThreadPool(int min, int max):minNum(min), maxNum(max), liveNum(min),busyNum(0),exitNum(0), shutDown(false)
{do{//实例化任务队列taskQ = new TaskQueue<T>;if (taskQ == nullptr){cout << "new taskQ fail .... " << endl;break;}threadIDs = new pthread_t[max]; // 根据最大的线程池设置进行创建空间if (threadIDs == nullptr){cout << "new theadIDs fail .... " << endl;break;}memset(threadIDs, 0, sizeof(pthread_t)*max);if (pthread_mutex_init(&mutexPool, NULL) != 0 ||           pthread_cond_init(&notEmpty, NULL) != 0 )          {cout<<"mutex or cond init error"<<endl;;break;}// 创建线程pthread_create(&managerID, NULL, manager, this);// 管理线程for (int i = 0; i < min; ++i){//printf("creating is %d\n", i);pthread_create(&threadIDs[i], NULL, worker, this); // 工作线程}cout << "threadPoolCreate create sucess  " << endl;      return;} while (0);//是否资源if (threadIDs) delete[] threadIDs;if (taskQ) delete taskQ;}template<class T>
ThreadPool<T>::~ThreadPool()
{//关闭线程池shutDown = true;// 阻塞回收管理者线程pthread_join(managerID, NULL);// 唤醒阻塞的消费者线程for (int i = 0; i < liveNum; i++){pthread_cond_signal(&notEmpty);}// 释放堆内存if (taskQ){delete taskQ;}if (threadIDs){delete[] threadIDs;}pthread_mutex_destroy(&mutexPool);pthread_cond_destroy(&notEmpty);}template<class T>
void ThreadPool<T>::addTask(Task<T> task)
{// 判断线程池是否被关闭if (shutDown){pthread_mutex_unlock(&mutexPool);return;}// 添加任务taskQ->addTask(task);pthread_cond_signal(&notEmpty);}template<class T>
int ThreadPool<T>::getBusyNum()
{pthread_mutex_lock(&mutexPool);int busyNum = this->busyNum;pthread_mutex_unlock(&mutexPool);return busyNum;
}template<class T>
int ThreadPool<T>::getAliveNum()
{pthread_mutex_lock(&mutexPool);int aliveNum = this->liveNum;pthread_mutex_unlock(&mutexPool);return aliveNum;
}template<class T>
void * ThreadPool<T>::worker(void * arg)
{ThreadPool* pool = static_cast<ThreadPool*>(arg); //参数进行类型转换while (true){       pthread_mutex_lock(&pool->mutexPool);// 判断当前任务队列是否为空,且线程没有关闭while (pool->taskQ->taskNumber() == 0 && !pool->shutDown){// 需要阻塞线程,锁和条件变量进行绑定，后面通过条件变量对其进行唤醒pthread_cond_wait(&pool->notEmpty, &pool->mutexPool);//判断是不是需要销毁线程,管理者线程会对exitNum进行初始化if (pool->exitNum > 0){pool->exitNum--;if (pool->liveNum > pool->minNum){pool->liveNum--;pthread_mutex_unlock(&pool->mutexPool);pool->threadExit();}}}// 如果有任务，则继续向下运行，开始消费任务了// 判断线程池是否关闭了if (pool->shutDown){pthread_mutex_unlock(&pool->mutexPool);pool->threadExit();}// 从任务队列取出任务Task<T> task = pool->taskQ->takeTask();// 任务队列应该设计成环形的队列，只需要确定头部和尾部即可，添加任务进行覆盖即可// 移动头结点// 解锁pool->busyNum++;pthread_mutex_unlock(&pool->mutexPool);// 当前子线程开始工作了，因此忙的状态需要标定cout << "thread " << to_string(pthread_self()) << "  start working... " << endl;// 开始工作task.function(task.arg);delete task.arg;task.arg = nullptr;// 当前子线程完成工作，因此把忙的状态取消cout << "thread  " << to_string(pthread_self()) << "  end working... " << endl;pthread_mutex_lock(&pool->mutexPool);pool->busyNum--;pthread_mutex_unlock(&pool->mutexPool);}return nullptr;
}template<class T>
void * ThreadPool<T>::manager(void * arg)
{ThreadPool* pool = static_cast<ThreadPool*>(arg);//参数进行类型转换while (!pool->shutDown){// 每隔3s检测一次sleep(3);//取出线程池中任务的数量和当前线程的数量pthread_mutex_lock(&pool->mutexPool);int queueSize = pool->taskQ->taskNumber();int liveNum = pool->liveNum;int busyNum = pool->busyNum;pthread_mutex_unlock(&pool->mutexPool);// 添加线程,需要制定一个添加的规则，可以根据实际情况进行设计即可// 任务的个数>存活的线程个数 && 存活的线程个数< 最大线程数if (queueSize > liveNum && liveNum < pool->maxNum){pthread_mutex_lock(&pool->mutexPool);int counter = 0;for (int i = 0; i < pool->maxNum && counter < NUMBER && pool->liveNum < pool->maxNum; ++i){if (pool->threadIDs[i] == 0){pthread_create(&pool->threadIDs[i], NULL, worker, pool);counter++;pool->liveNum++;}}pthread_mutex_unlock(&pool->mutexPool);}// 销毁线程// 忙的线程*2 < 存活的线程数 && 存活的线程数>最小线程数if (busyNum * 2 < liveNum && liveNum > pool->minNum){pthread_mutex_lock(&pool->mutexPool);pool->exitNum = NUMBER;pthread_mutex_unlock(&pool->mutexPool);// 让空闲的工作线程自杀for (int i = 0; i < NUMBER; i++){pthread_cond_signal(&pool->notEmpty);}}}return nullptr;
}template<class T>
void ThreadPool<T>::threadExit()
{pthread_t tid = pthread_self();for (int i = 0; i < this->maxNum; i++){if (this->threadIDs[i] == tid){this->threadIDs[i] = 0;cout << "threadExit() called " << to_string(tid) << " exiting ..." << endl;break;}}pthread_exit(NULL);
}

测试代码

#include <iostream>
#include<unistd.h>
#include"ThreadPool.h"
#include"ThreadPool.cpp"
using namespace std;void taskFunc(void* arg)
{int num = *(int*)arg;cout << "thread  " << to_string(pthread_self()) << "  is working, number = " << num << endl;sleep(1);
}int main()
{ cout << " hello world ThreadPool_c++ " << endl;//创建线程池ThreadPool<int> pool(3, 10);for (int i = 0; i < 100; i++){int* num = new int(i+100);//printf("*num = %d\n", *num);pool.addTask(Task<int>(taskFunc,num));}sleep(30);return 0;}

linux下c++版本线程池的实现相关推荐

Linux下C语言线程池的实现（1）
http://hi.baidu.com/lingiloveyou/blog/item/21e57cf3322a6b40342accc7.html 什么时候需要创建线程池呢?简单的说,如果一个应用需要频 ...
LINUX 下C实现线程池《转载》
原文转载自:https://blog.csdn.net/hubi0952/article/details/8045094 1.线程池基本原理在传统服务器结构中, 常是有一个总的监听线程监听有没有 ...
YOLOv4：目标检测（windows和Linux下Darknet 版本）实施
YOLOv4:目标检测(windows和Linux下Darknet 版本)实施 YOLOv4 - Neural Networks for Object Detection (Windows and L ...
Windows和Linux下通用的线程接口
对于多线程开发,Linux下有pthread线程库,使用起来比较方便,而Windows没有,对于涉及到多线程的跨平台代码开发,会带来不便.这里参考网络上的一些文章,整理了在Windows和Linux下 ...
linux 下查看应用版本信息,Linux下查看版本信息
Linux下如何查看版本信息, 包括位数.版本信息以及CPU内核信息.CPU具体型号等. 1.# uname -a (Linux查看版本当前操作系统内核信息) 2.# cat /proc/vers ...
Multi-thread--Windows和Linux下通用的线程接口
对于多线程开发,Linux下有pthread线程库,使用起来比较方便,而Windows没有,对于涉及到多线程的跨平台代码开发,会带来不便.这里参考网络上的一些文章,整理了在Windows和Linux下 ...
Windows/Linux下获取当前线程的ID号
序在多线程场合,为了方便跟踪线程的运行状态,往往需要在程序中添加打印当前线程ID号的功能. 1. Linux下打印当前线程ID pthread_t pthread_self() 2. Windows ...
Linux下JIRA版本5.0.1的安装.破解.汉化
Linux下JIRA版本5.0.1的安装.破解.汉化 2013-11-10 01:36:27 标签:linux jira 安装原创作品,允许转载,转载时请务必以超链接形式标明文章原始出处 .作者 ...
Linux下EasyPanel版本安装及升级
Linux下EasyPanel版本安装及升级本脚本为官方脚本,集成了kangle web服务器和mysql,仅支持centos 5和centos 6. 执行下面的命令即可,安装程序将自动安装或者升级 ...

linux下c++版本线程池的实现

任务队列

头文件

源文件

线程池

线程池头文件

线程池源文件

测试代码

linux下c++版本线程池的实现相关推荐

最新文章

热门文章