用C++实现的带队列,并且当队列为空及任务数大于最大队列长度时自动wait的线程池 ( by quqi99 )
2012-06-14 16:14
429 查看
用C++实现的带队列,并且当队列为空及任务数大于最大队列长度时自动wait的线程池 ( by quqi99 )
作者:张华 发表于:2012-06-14
1) 带队列的线程池
2) 当队列为空, 通过pthread_mutex_lock实现java中的wait, 用pthread_cond_signal实现java中的notify功能
3) 当任务数大于所设定的队列长度时,为防内存溢出, 同样通过pthread_mutex_lock实现java中的wait, 用pthread_cond_signal实现java中的notify功能实现自动等待
代码如下:
1) ThreadPool.h
/*
* ThreadPool.h
*
* Created on: 14 Jun 2012
* Author: Zhang Hua
*/
#ifndef THREADPOOL_H_
#define THREADPOOL_H_
#include <queue>
#include <string>
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <iostream>
using namespace std;
class Task {
protected:
int pos;
string input;
public:
Task() {
}
virtual ~Task() {
}
Task(int pos, string input) {
this->pos = pos;
this->input = input;
}
int getPos() {
return this->pos;
}
string getInput() {
return this->input;
}
virtual void run()=0;
};
class ThreadPool {
private:
int m_threadNum;
int m_maxQueueSize;
static queue<Task*>* taskQueue; //因为Task里有纯虚函数,所以为Task*
static pthread_cond_t stop_if_empty, stop_if_full;
static pthread_mutex_t m_mutex;
typedef struct threadarg {
int para1;
} ARG;
void clear();
public:
ThreadPool(int, int);
~ThreadPool();
static void* ThreadFunc(void *arg);
int addTask(Task*);
int waitUnitTaskFinish();
};
#endif /* THREADPOOL_H_ */
2) ThreadPool.cpp
/*
* ThreadPool.cpp
*
* Created on: 14 Jun 2012
* Author: zhang hua
*/
#include "ThreadPool.h"
queue<Task*>* ThreadPool::taskQueue = new queue<Task*>();
pthread_mutex_t ThreadPool::m_mutex = PTHREAD_MUTEX_INITIALIZER; //互斥量,用于线程同步
pthread_cond_t ThreadPool::stop_if_empty = PTHREAD_COND_INITIALIZER; //条件量,队列为空的条件量
pthread_cond_t ThreadPool::stop_if_full = PTHREAD_COND_INITIALIZER; //条件量,队列大于最大队列长度时的条件量
ThreadPool::ThreadPool(int threadNum, int maxQueueSize) {
this->m_threadNum = threadNum;
this->m_maxQueueSize = maxQueueSize;
pthread_t tid = 0;
ARG* threadarg = (ARG*) malloc(sizeof(ARG));
threadarg->para1 = maxQueueSize;
for (int i = 0; i < m_threadNum; i++) {
pthread_create(&tid, NULL, ThreadFunc, (void *) threadarg);
}
}
void *ThreadPool::ThreadFunc(void *arg) {
while (true) {
ARG* threadarg = (ARG *) arg;
int maxQueueSize = threadarg->para1;
int pid = pthread_self();
cout << "[" << pid << "] thread is running ... " << endl;
pthread_mutex_lock(&m_mutex);
while (taskQueue->empty()) {
cout << "[" << pid << "]"
<< "队列为空,将等待,直到stop_if_empty条件量的再次触发线程才往下执行 ..." << endl;
pthread_cond_wait(&stop_if_empty, &m_mutex);
}
if (!taskQueue->empty() && (int) taskQueue->size() < maxQueueSize) {
cout << "[" << pid << "]" << "触发stop_if_full条件量..." << endl;
pthread_cond_signal(&stop_if_full);
}
Task* task = taskQueue->front();
taskQueue->pop();
pthread_mutex_unlock(&m_mutex);
task->run();
}
return 0;
}
int ThreadPool::addTask(Task* task) {
int pid = pthread_self();
pthread_mutex_lock(&m_mutex);
while ((int) taskQueue->size() >= this->m_maxQueueSize) {
cout << "[" << pid << "]"
<< "队列已满,将等待,直到stop_if_full条件量的再次触发线程才往下执行 ..." << endl;
pthread_cond_wait(&stop_if_full, &m_mutex);
}
taskQueue->push(task);
cout << "[" << pid << "]" << "触发stop_if_empty条件量..." << endl;
pthread_cond_signal(&stop_if_empty);
cout << "[" << pid << "]" << "add to queue ..." << task->getPos() << endl;
pthread_mutex_unlock(&m_mutex);
return 0;
}
int ThreadPool::waitUnitTaskFinish() {
while (!taskQueue->empty()) {
sleep(5);
}
clear();
return 0;
}
void ThreadPool::clear() {
while (!taskQueue->empty()) {
taskQueue->pop();
}
delete taskQueue;
pthread_cond_destroy(&stop_if_empty);
pthread_cond_destroy(&stop_if_full);
}
ThreadPool::~ThreadPool() {
clear();
}
3) PerfermanceTest.cpp 测试程序
//============================================================================
// Name : PerfermanceTest.cpp
// Author : zhang hua
// Version :
// Copyright :
// Description : 一个带队列的线程池,如果任务负载大于线程池的最大队列长度将自动等待
//============================================================================
#include"ThreadPool.h"
#include<iostream>
#include<sys/time.h>
using namespace std;
class WorkTask: public Task {
public:
WorkTask(int pos, string input) {
this->pos = pos;
this->input = input;
}
void run() {
int pid = pthread_self();
cout << "[" << pid << "]" << "直正干活的线程任务... " << this->pos << endl;
}
};
int main() {
int pid = pthread_self();
string input = "this is test ";
int threadName = 10;
int maxQueueSize = 100;
ThreadPool threadpool(threadName, maxQueueSize);
for (int i = 0; i < 10000; i++) {
cout << "[" << pid << "]" << "addTask " << i << endl;
WorkTask* workTask = new WorkTask(i, input);
threadpool.addTask(workTask);
}
threadpool.waitUnitTaskFinish();
}
作者:张华 发表于:2012-06-14
版权声明:可以任意转载,转载时请务必以超链接形式标明文章原始出处和作者信息及本版权声明
( http://blog.csdn.net/quqi99 )
用J***A, shell, python都写过很多程序,但不算学校的用C++写程序这个算第一个. 看到java中的线程通过wait, notify可以很方便的实现线程的同步, 于是, 想用c++实现一个, 于是有了下列这个线程池,它的特点是:1) 带队列的线程池
2) 当队列为空, 通过pthread_mutex_lock实现java中的wait, 用pthread_cond_signal实现java中的notify功能
3) 当任务数大于所设定的队列长度时,为防内存溢出, 同样通过pthread_mutex_lock实现java中的wait, 用pthread_cond_signal实现java中的notify功能实现自动等待
代码如下:
1) ThreadPool.h
/*
* ThreadPool.h
*
* Created on: 14 Jun 2012
* Author: Zhang Hua
*/
#ifndef THREADPOOL_H_
#define THREADPOOL_H_
#include <queue>
#include <string>
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <iostream>
using namespace std;
class Task {
protected:
int pos;
string input;
public:
Task() {
}
virtual ~Task() {
}
Task(int pos, string input) {
this->pos = pos;
this->input = input;
}
int getPos() {
return this->pos;
}
string getInput() {
return this->input;
}
virtual void run()=0;
};
class ThreadPool {
private:
int m_threadNum;
int m_maxQueueSize;
static queue<Task*>* taskQueue; //因为Task里有纯虚函数,所以为Task*
static pthread_cond_t stop_if_empty, stop_if_full;
static pthread_mutex_t m_mutex;
typedef struct threadarg {
int para1;
} ARG;
void clear();
public:
ThreadPool(int, int);
~ThreadPool();
static void* ThreadFunc(void *arg);
int addTask(Task*);
int waitUnitTaskFinish();
};
#endif /* THREADPOOL_H_ */
2) ThreadPool.cpp
/*
* ThreadPool.cpp
*
* Created on: 14 Jun 2012
* Author: zhang hua
*/
#include "ThreadPool.h"
queue<Task*>* ThreadPool::taskQueue = new queue<Task*>();
pthread_mutex_t ThreadPool::m_mutex = PTHREAD_MUTEX_INITIALIZER; //互斥量,用于线程同步
pthread_cond_t ThreadPool::stop_if_empty = PTHREAD_COND_INITIALIZER; //条件量,队列为空的条件量
pthread_cond_t ThreadPool::stop_if_full = PTHREAD_COND_INITIALIZER; //条件量,队列大于最大队列长度时的条件量
ThreadPool::ThreadPool(int threadNum, int maxQueueSize) {
this->m_threadNum = threadNum;
this->m_maxQueueSize = maxQueueSize;
pthread_t tid = 0;
ARG* threadarg = (ARG*) malloc(sizeof(ARG));
threadarg->para1 = maxQueueSize;
for (int i = 0; i < m_threadNum; i++) {
pthread_create(&tid, NULL, ThreadFunc, (void *) threadarg);
}
}
void *ThreadPool::ThreadFunc(void *arg) {
while (true) {
ARG* threadarg = (ARG *) arg;
int maxQueueSize = threadarg->para1;
int pid = pthread_self();
cout << "[" << pid << "] thread is running ... " << endl;
pthread_mutex_lock(&m_mutex);
while (taskQueue->empty()) {
cout << "[" << pid << "]"
<< "队列为空,将等待,直到stop_if_empty条件量的再次触发线程才往下执行 ..." << endl;
pthread_cond_wait(&stop_if_empty, &m_mutex);
}
if (!taskQueue->empty() && (int) taskQueue->size() < maxQueueSize) {
cout << "[" << pid << "]" << "触发stop_if_full条件量..." << endl;
pthread_cond_signal(&stop_if_full);
}
Task* task = taskQueue->front();
taskQueue->pop();
pthread_mutex_unlock(&m_mutex);
task->run();
}
return 0;
}
int ThreadPool::addTask(Task* task) {
int pid = pthread_self();
pthread_mutex_lock(&m_mutex);
while ((int) taskQueue->size() >= this->m_maxQueueSize) {
cout << "[" << pid << "]"
<< "队列已满,将等待,直到stop_if_full条件量的再次触发线程才往下执行 ..." << endl;
pthread_cond_wait(&stop_if_full, &m_mutex);
}
taskQueue->push(task);
cout << "[" << pid << "]" << "触发stop_if_empty条件量..." << endl;
pthread_cond_signal(&stop_if_empty);
cout << "[" << pid << "]" << "add to queue ..." << task->getPos() << endl;
pthread_mutex_unlock(&m_mutex);
return 0;
}
int ThreadPool::waitUnitTaskFinish() {
while (!taskQueue->empty()) {
sleep(5);
}
clear();
return 0;
}
void ThreadPool::clear() {
while (!taskQueue->empty()) {
taskQueue->pop();
}
delete taskQueue;
pthread_cond_destroy(&stop_if_empty);
pthread_cond_destroy(&stop_if_full);
}
ThreadPool::~ThreadPool() {
clear();
}
3) PerfermanceTest.cpp 测试程序
//============================================================================
// Name : PerfermanceTest.cpp
// Author : zhang hua
// Version :
// Copyright :
// Description : 一个带队列的线程池,如果任务负载大于线程池的最大队列长度将自动等待
//============================================================================
#include"ThreadPool.h"
#include<iostream>
#include<sys/time.h>
using namespace std;
class WorkTask: public Task {
public:
WorkTask(int pos, string input) {
this->pos = pos;
this->input = input;
}
void run() {
int pid = pthread_self();
cout << "[" << pid << "]" << "直正干活的线程任务... " << this->pos << endl;
}
};
int main() {
int pid = pthread_self();
string input = "this is test ";
int threadName = 10;
int maxQueueSize = 100;
ThreadPool threadpool(threadName, maxQueueSize);
for (int i = 0; i < 10000; i++) {
cout << "[" << pid << "]" << "addTask " << i << endl;
WorkTask* workTask = new WorkTask(i, input);
threadpool.addTask(workTask);
}
threadpool.waitUnitTaskFinish();
}
相关文章推荐
- Windows下C++实现线程池功能(固定线程数,自动调整未实现)(
- C++ 实现线程安全的任务队列
- LINUX C++ 线程池简单实现之双队列
- 用队列实现自动排号机(C++实现)
- 优先队列(C++实现)
- C++第九周【任务一】定义Complex类中的<<和>>运算符的重载,实现输入和输出
- C++ 线程池的简易实现
- 服务器文件自动打包备份到电脑,可以通过flashfxp及电脑的计划任务实现
- 螺旋队列C++代码实现实例及运行结果
- 【c++版数据结构】链队列的实现
- 使用java.util.concurrent实现的线程池、消息队列功能
- Linux利用inotify-tools的inotifywait实现:当文件夹内容改变时自动执行一段脚本
- 用css样式实现在超出长度后,自动显示为。。。
- 线程池原理及创建(C++实现)
- 单链队列,队列的链式存储结构C++实现
- 数据结构之队列的基本操作入队出队初始化删除-c++代码实现及运行实例结果
- 托管C++线程锁实现 c++11线程池
- 算法导论 第六章优先队列C++实现
- 一个HTTP.二进制POST和HTTP指定长度接收的C++实现
- C++实现 3.7猫狗队列