#include<thread>
#include<mutex>
#include<vector>
#include<iostream>
#include<string>
#include<deque>
using namespace std;
namespace Z1 {
class Task
{
public:
Task(void* arg = NULL, const std::string taskName = "")
: arg_(arg)
, taskName_(taskName)
{
}
virtual ~Task()
{
}
void setArg(void* arg)
{
arg_ = arg;
}
virtual int run() = 0;
protected:
void* arg_;
std::string taskName_;
};
class ThreadPool{
private:
volatile bool isRunning_;//volatile的变量是说这变量可能会被意想不到地改变，这样，编译器就不会去假设这个变量的值了。
int threadNum_;
pthread_t* threads_;
deque<Task *> taskDeque_;
pthread_mutex_t mutex_;
pthread_cond_t condition_;
public:
ThreadPool(int threadNum = 10);
~ThreadPool();
size_t addTask(Task* task);
void stop();
int size();
Task* take();
private:
int createThreads();
static void* threadFunc(void * threadData);
private:
ThreadPool& operator=(const ThreadPool&);
ThreadPool(const ThreadPool&);
};
}
#include<threadpool.h>
#include<assert.h>
namespace Z1 {
ThreadPool::ThreadPool(int threadNum)
{
isRunning_ = true;
threadNum_ = threadNum;
createThreads();
}
ThreadPool::~ThreadPool()
{
stop();
deque<Task *>::iterator pd;
for(pd = taskDeque_.begin();pd != taskDeque_.end();++pd)
delete *pd;
taskDeque_.clear();
}
int ThreadPool::createThreads()
{
pthread_mutex_init(&mutex_,NULL);
pthread_cond_init(&condition_,NULL);
threads_ = new pthread_t[threadNum_];
//创建threadNum_个线程，并开始运行线程，在线程里取任务
//线程函数的参数是threadpool对象，每个线程的参数对象都是一样的（对象地址一样），传对象是因为线程函数是static的
for(int i = 0;i < threadNum_;++i)
pthread_create(&threads_[i],NULL,threadFunc,this);
return 0;
}
size_t ThreadPool:: addTask(Task* task)//任务队列中装的是任务对象指针
{
pthread_mutex_lock(&mutex_);
taskDeque_.push_back(task);
size_t size = taskDeque_.size();
pthread_cond_signal(&condition_);
pthread_mutex_unlock(&mutex_);
return size;
}
void ThreadPool::stop()
{
if(!isRunning_)
return;
isRunning_ = false;
pthread_cond_broadcast(&condition_);//激活全部线程的条件变量，先激活首先获得互斥锁的那一个。
for(int i = 0;i < threadNum_;++i)
pthread_join(threads_[i],NULL);//调用这个函数等待一个线程终止（在主线程中写时，若新开的线程没有终止，则阻塞在这里）
//这个函数类似于多进程中的waitpid(杀死进程，防止进程僵死)。
delete[] threads_;//指向线程ID的指针
threads_ = NULL;
pthread_mutex_destroy(&mutex_);
pthread_cond_destroy(&condition_);
}
int ThreadPool::size()
{
pthread_mutex_lock(&mutex_);
size_t size = taskDeque_.size();
pthread_mutex_unlock(&mutex_);
return size;
}
Task* ThreadPool::take()//从任务队列中取任务（取得的是一个指向任务对象的指针）
{
Task* task = NULL;
while(!task)
{
pthread_mutex_lock(&mutex_);
while(taskDeque_.empty() && isRunning_)//没有任务的话等待（使用while是防止虚假唤醒）
pthread_cond_wait(&condition_,&mutex_);//线程池stop的时候这里会被激活,AND isRunning_ = false
if(!isRunning_)
{
pthread_mutex_unlock(&mutex_);
break;
}
else if(taskDeque_.empty())//防止虚假唤醒需要在唤醒后再做一次判断。
{
pthread_mutex_unlock(&mutex_);
continue;
}
assert(!taskDeque_.empty());
task = taskDeque_.front();
taskDeque_.pop_front();
pthread_mutex_unlock(&mutex_);
}
return task;
}
void* ThreadPool::threadFunc(void* arg)//在pthread_create之后就会执行,然后从任务队列中取任务，没有任务的话等待
{
pthread_t tid = pthread_self();
//static函数没有this指针，所以必须传进来对象，每个线程中的对象地址是相同的，也就是说每个线程中其实是一个对象
ThreadPool* pool = static_cast<ThreadPool*>(arg);
//cout<<pool<<endl;
while (pool->isRunning_)//当前线程执行完任务之后，再从任务队列中取任务
{
Task* task = pool->take();//如果线程池一直在运行，且没有任务，则会等待任务进队列
if (!task)//
{
printf("thread %lu will exit\n", tid);//执行stop函数之后执行（直接调用stop或者析构函数执行），
//因为如果线程池一直在运行，且没有任务，则会等待任务进队列
break;
}
assert(task);
task->run();
}
return 0;
}
}
#include <iostream>
#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <pthread.h>
#include "threadpool.h"
class MyTask: public Z1::Task
{
public:
MyTask(){}
virtual int run()
{
printf("thread[%lu] : %s\n", pthread_self(), (char*)this->arg_);
sleep(1);
return 0;
}
};
int main()
{
char szTmp[] = "hello world";
MyTask taskObj;
taskObj.setArg((void*)szTmp);
Z1::ThreadPool threadPool(10);
for(int i = 0; i < 30; i++)
{
threadPool.addTask(&taskObj);
}
while(1)
{
printf("there are still %d tasks need to process\n", threadPool.size());
if (threadPool.size() == 0)
{
threadPool.stop();
printf("Now I will exit from main\n");
exit(0);
}
sleep(2);
}
sleep(5);
return 0;
}
