我的程序人生

c++ 线程池的实现(原)

看群里有同志老是在找线程池的实现,听说网上曾经发布的都是不正确的,今天我就自己弄了一个,不正确的地方大家指点指点

mutex.hxx 互斥类
 1#ifndef INCLUDE_MUTEX_HH
 2#define INCLUDE_MUTEX_HH
 3#include <pthread.h>
 4
 5class Mutex
 6{
 7public:
 8  Mutex();
 9  virtual ~Mutex();
10  void lock();
11  void unlock();
12  pthread_mutex_t *get_mutex();
13private:
14  pthread_mutex_t mutex;
15}
;
16
17#endif
18

mutex.cxx互斥实现类
#include "mutex.hxx"
#include 
"error.hxx"
Mutex::Mutex()
{
  
if(pthread_mutex_init(&mutex,NULL))
    
{
      perror(
"pthread_mutex_init error");
      
throw MutexError("pthread_mutex_init error");
    }

}

Mutex::
~Mutex()
{
  
if(pthread_mutex_destroy(&mutex))
    
{
      perror(
"pthread_mutex_destroy error");
      
throw MutexError("pthread_mutex_destroy error");
    }

}


void Mutex::lock()
{
  pthread_mutex_lock(
&mutex);
}


void Mutex::unlock()
{
  pthread_mutex_unlock(
&mutex);
}


pthread_mutex_t 
*Mutex::get_mutex()
{
  
return &mutex;
}


error.hxx 异常类型
#ifndef INCLUDE_ERROR_HH
#define INCLUDE_ERROR_HH
#include 
<stdexcept>

class MutexError:public std::runtime_error
{
public:
  MutexError(
const std::string& what)
    :std::runtime_error(what.c_str())
  
{}
  MutexError(
const char* const what)
    :std::runtime_error(what)
  
{}
}
;

#endif

task.hxx 任务类,所有的任务需要实现此接口
#ifndef INCLUDE_TASK_HH
#define INCLUDE_TASK_HH

#include 
<string>
#include 
"mutex.hxx"
//class Mutex;
class Task
{
  friend 
bool operator<(const Task& t1,const Task& t2);
public:
  Task(
const std::string& taskName=std::string(),int level=0);
  
virtual ~Task(){};
  
void setLevel(int level);
  std::
string taskName()const;
  std::
string taskName();
  
void setName(const std::string&);
  
virtual void run()=0;

private:
  Mutex mutex;
  
int level_;
  std::
string taskName_;

}
;
#endif

task.cxx 任务实现代码
#include "task.hxx"
//#include "mutex.hxx"

Task::Task(
const std::string& name,int level)
  :taskName_(name),level_(level)
{
}


void Task::setLevel(int level)
{
  mutex.
lock();
  level_
=level;
  mutex.unlock();
}

std::
string Task::taskName()const
{
  
return taskName_;
}

std::
string Task::taskName()
{
  
return taskName_;
}

void Task::setName(const std::string& name)
{
  mutex.
lock();
  taskName_
=name;
  mutex.unlock();
}

bool operator<(const Task& t1,const Task& t2)
{
  
return t1.level_<t2.level_;
}


池头文件 pool.hxx
#ifndef INCLUDE_POOL_HH
#define INCLUDE_POOL_HH
#include 
<pthread.h>
#include 
<queue>
#include 
<list>
#include 
"mutex.hxx"
class Task;
class ThreadPool
  :
private Mutex
{
public:
  ThreadPool(
int);
  
~ThreadPool();
  
void addTask(Task*);
  
void wait();
  
void release(const pthread_t&);
  Task
* get();
  
void setTimeout(long t);
private:
  typedef std::list
<pthread_t>::iterator ThreadIterator;
  pthread_cond_t release_cond;
  pthread_cond_t task_cond;
  
static void* threadFunc(void*);
  
void init(int);
  std::priority_queue
<Task*> tasks;
  std::list
<pthread_t> idleThreads;
  std::list
<pthread_t> busyThreads;
  
long timeout_second;
}
;

#endif

池实现文件
#include "pool.hxx"
#include 
"task.hxx"
#include 
<algorithm>
#include 
<ctime>
#include 
<iostream>

ThreadPool::ThreadPool(
int threadNumber)
  :timeout_second(
10)
{
  pthread_cond_init(
&release_cond,NULL);
  pthread_cond_init(
&task_cond,NULL);
  init(threadNumber);
}


ThreadPool::
~ThreadPool()
{
  pthread_cond_destroy(
&release_cond);
  pthread_cond_destroy(
&task_cond);
}


void ThreadPool::init(int threadNumber)
{
  
for(int i=0;i<threadNumber;i++)
    
{
      pthread_t t;
      pthread_create(
&t,NULL,threadFunc,this);
      busyThreads.push_back(t);
    }

}


void ThreadPool::setTimeout(long t)
{
  
if(t>0)
    timeout_second
=t;
}


void ThreadPool::addTask(Task* task)
{
  
lock();
  tasks.push(task);
  pthread_cond_signal(
&task_cond);
  unlock();
}


Task
* ThreadPool::get()
{
  
struct timespec timeout;
  timeout.tv_sec
=time(NULL)+timeout_second;
  timeout.tv_nsec
=0;
  
lock();
  
if(tasks.empty())
    
{
      pthread_cond_timedwait(
&task_cond,get_mutex(),&timeout);
    }

  
if(tasks.empty())
    
{
      std::cout
<<"empty"<<std::endl;
      unlock();
      
return NULL;
    }

  Task 
*task=tasks.top();
  tasks.pop();
  unlock();
  
return task;
}


void * ThreadPool::threadFunc(void* args)
{
  ThreadPool
* pool=static_cast<ThreadPool*>(args);
  Task
* task;
  
while((task=pool->get())!=NULL)
    
{
      task
->run();
    }

  pool
->release(pthread_self());
}


void ThreadPool::release(const pthread_t& t)
{
  
lock();
  ThreadIterator it;
  it
=std::find(busyThreads.begin(),busyThreads.end(),t);
  
if(it!=busyThreads.end())
    
{
      busyThreads.erase(it);
    }

  idleThreads.push_back(t);
  pthread_cond_signal(
&release_cond);
  unlock();
}

void ThreadPool::wait()
{
  
lock();
  
while(!busyThreads.empty())
    
{
      
struct timespec timeout;
      timeout.tv_sec
=time(NULL)+10;
      timeout.tv_nsec
=0;

      pthread_cond_timedwait(
&release_cond,get_mutex(),&timeout);
    }


  
for(ThreadIterator it=idleThreads.begin();it!=idleThreads.end();it++)
    
{
      pthread_join(
*it,NULL);
    }

  unlock();
  
}



测试文件
#include "pool.hxx"
#include 
"task.hxx"
#include 
<unistd.h>
#include 
<iostream>
#include 
<string>
#include 
<vector>
#include 
<sstream>
#include 
<memory>
#include 
"mutex.hxx"
class WorkTask
  :
public Task
{
public:
  WorkTask(
int level,void *data):Task(std::string(),level)
  
{
    
this->data_=data;
  }

  
~WorkTask(){}
  
virtual void run()
  
{
    std::cout
<<taskName()<<(char*)data_<<std::endl;
    sleep(
2);
    std::cout
<<taskName()<<" ok"<<std::endl;
  }

private:
  
void *data_;
  Mutex mutex;
}
;

int main(void)
{
  ThreadPool pool(
5);
  
char szTemp[]="aaaaaaaaaaaaaaabbbbbbbbbbbccccccccccdddddddddd";
  WorkTask task(
1,szTemp);
  
char buf[20];
  std::vector
<Task*> tasks;
  
for(int i=0;i<10;i++)
    
{
      snprintf(buf,
sizeof(buf),"%s %d","task",i);
      task.setName(buf);
      std::auto_ptr
<Task> t(new WorkTask(task));
      pool.addTask(t.
get());
      tasks.push_back(t.release());
    }

  pool.wait();
  
for(std::vector<Task*>::iterator it=tasks.begin();it!=tasks.end();it++)
    
{
      delete 
*it;
    }

  
return 0;
}





测试的结果


没有注释直接看源码就可解决。。


注:使用本人代码请注明本人的信息

posted on 2010-04-08 15:43 lancey 阅读(7171) 评论(9)  编辑 收藏 引用

Feedback

# re: c++ 线程池的实现(原) 2010-04-08 18:02 oh no

永远不要在析构函数抛出异常。  回复  更多评论   

# re: c++ 线程池的实现(原)[未登录] 2010-04-08 20:52 OnTheWay

#include <pthread.h> 这个文件是系统的吗,还是你写的?  回复  更多评论   

# re: c++ 线程池的实现(原) 2010-04-08 22:58 ChowZenki

@OnTheWay
那是linux的頭文件  回复  更多评论   

# re: c++ 线程池的实现(原) 2010-04-09 02:49 欲三更

两点不同意见:

1.
Mutex::~Mutex()
{
...
throw MutexError("pthread_mutex_destroy error");
...
}

这个绝对不可以有。。。

2.我觉得lock和unlock之间的代码不安全,如果抛出异常可能导致死锁。

  回复  更多评论   

# re: c++ 线程池的实现(原) 2010-04-09 11:39 china_zhli@163.com

@oh no
用代码说明  回复  更多评论   

# re: c++ 线程池的实现(原) 2010-04-09 16:33 lancey

其实我也是不打算让mutex的构造与析构时异常抛出,只是想在连互斥量都无法初始化与销毁时我直接abort或不处理都感觉不好,才这样做,留给调用者处理
  回复  更多评论   

# re: c++ 线程池的实现(原) 2010-04-09 17:31 路过

垃圾实现,去看看ACE或apr实现再来吧,哎  回复  更多评论   

# re: c++ 线程池的实现(原) 2010-04-13 02:06 欲三更

@路过
回访,容我说一句,lz这个东西写的确实不够好,这个不用看高超的实现就能判断。简单的说就是这个实现“不能用”。但是也不能说是垃圾吧?至少lz写的代码整整齐齐,没有注释也很容易看懂,基本方法也对。

“垃圾”这种说法,还是尽量别出现在这个站点比较好。  回复  更多评论   

# re: c++ 线程池的实现(原) 2010-04-13 02:07 欲三更

@路过
ps:apr真的是做得很好:)  回复  更多评论   



只有注册用户登录后才能发表评论。
网站导航: 博客园   IT新闻   BlogJava   知识库   博问   管理


My Links

Blog Stats

常用链接

留言簿(2)

随笔档案

文章分类

文章档案

我的链接

搜索

最新评论

阅读排行榜

评论排行榜