MIT 2012分布式课程基础源码解析-线程池实现

主要内容

ScopedLock

队列实现

线程池实现

在正式讲解线程池实现之前，先讲解两个有用的工具类：

ScopedLock

fifo队列

ScopedLock:

ScopedLock是局域锁的实现(我也不知道叫什么，姑且这么说吧)，它使用了C++中RAII（Resource acquisition is initialization资源获取即初始化），这种技巧实现的锁可在代码块开始处初始化锁，在代码块结束处释放锁，可省去try catch这样的语句，具体实现如下：

struct ScopedLock {
private:
pthread_mutex_t *m_;
public:
ScopedLock(pthread_mutex_t *m): m_(m) {
VERIFY(pthread_mutex_lock(m_)==0);
}
~ScopedLock() {
VERIFY(pthread_mutex_unlock(m_)==0);
}
};

其中宏VERIFY定义如下，这时贯穿于整个项目并最常用的一个宏：

#ifdef NDEBUG
#define VERIFY(expr) do { if (!(expr)) abort(); } while (0)
#else
#define VERIFY(expr) assert(expr)
#endif

ScopedLock的使用方法如下：

void func() {
......
{
ScopedLock lock(&m_);
......
}
......
}

在代码块{}定义一个lock变量即可，在}处便能自动调用析构函数，从而自动释放锁，这种技巧在《c++必知必会》中也是强烈推荐的一种技巧。

fifo队列实现

代码中的fifo队列实现了简单生产者消费者模型，提供了阻塞非阻塞选项，实现代码在fifo.h文件下，我们首先看看类定义：

//线程执行方法，while循环中获取队列中的工作，因为队列默认是阻塞队列
//线程在没获取到工作时，将阻塞在相应的条件变量上
static void *
do_worker(void *arg)
{
ThrPool *tp = (ThrPool *)arg; //将this转换为ThrPool指针
while (1) {
ThrPool::job_t j;
if (!tp->takeJob(&j))
break; //die

(void)(j.f)(j.a); //执行工作
}
pthread_exit(NULL);
}

//if blocking, then addJob() blocks when queue is full
//otherwise, addJob() simply returns false when queue is full
ThrPool::ThrPool(int sz, bool blocking)
: nthreads_(sz),blockadd_(blocking),jobq_(100*sz)
{
pthread_attr_init(&attr_);
pthread_attr_setstacksize(&attr_, 128<<10);

for (int i = 0; i < sz; i++) {
pthread_t t;
//注意这里函数是do_worker，添加的参数为this，这样在do_worker函数中方便取出更多的信息
VERIFY(pthread_create(&t, &attr_, do_worker, (void *)this) ==0);
th_.push_back(t);
}
}

//IMPORTANT: this function can be called only when no external thread
//will ever use this thread pool again or is currently blocking on it
ThrPool::~ThrPool()
{
for (int i = 0; i < nthreads_; i++) {
job_t j;
j.f = (void *(*)(void *))NULL; //poison pill to tell worker threads to exit
jobq_.enq(j);
}

for (int i = 0; i < nthreads_; i++) {
VERIFY(pthread_join(th_[i], NULL)==0);
}

VERIFY(pthread_attr_destroy(&attr_)==0);
}

//添加工作的私有类，初始化job_t类，并添加到队列中
bool
ThrPool::addJob(void *(*f)(void *), void *a)
{
job_t j;
j.f = f;
j.a = a;

return jobq_.enq(j,blockadd_);
}

//获取队列中的工作回调，注意传入的是指针
bool
ThrPool::takeJob(job_t *j)
{
jobq_.deq(j);
return (j->f!=NULL);
}

View Code
该线程池的实现确认令人咋舌，很巧妙的将回调类转换成了内部的job_t类，也不失为一个很好的c++学习案例。

使用该线程池很简单，只需定义好相应的事件回调类，然后初始化线程池，再将回调类添加(addObjJob)到线程池中即可