base7. ThreadPool线程池类——生产者与消费者模型

ThreadPool类图

数据成员：

MutexLock mutex_：一个MutexLock类型的互斥变量mutex_
Condition cond_：一个Condition类型的条件变量cond_
string name_：线程池的名称
boost::ptr_vector<muduo::Thread>threads_：消费者线程队列threads_
std::deque<task. >queue_：任务队列queue_，其中task为一个执行任务的函数，生产者线程在queue_添加任务，所有threads_共享queue_
bool running_：running_表示线程池是否处于运行状态

typedef

typedef boost::function<void ()> Task：基于对象编程的回调函数

成员函数：

explicit ThreadPool(const string& name = string())：构造函数
~ThreadPool()：析构函数
void start(int numThreads)：启动线程池，其中有numThreads个线程
void stop()：关闭线程池
void run(const Task& f)：将任务task添加到线程中的任务队列queue_
void runInThread()：线程池中的线程要执行的函数
Task take()：获取任务

ThreadPool.h

//线程池本质上也是一个生产者/消费者模型
#ifndef MUDUO_BASE_THREADPOOL_H
#define MUDUO_BASE_THREADPOOL_H

#include <muduo/base/Condition.h>
#include <muduo/base/Mutex.h>
#include <muduo/base/Thread.h>
#include <muduo/base/Types.h>

#include <boost/function.hpp>
#include <boost/noncopyable.hpp>
#include <boost/ptr_container/ptr_vector.hpp>

#include <deque>

//固定的线程池，线程池中的线程个数是固定的，不可以自动伸缩
namespace muduo
{

class ThreadPool : boost::noncopyable
{
 public:
  typedef boost::function<void ()> Task;

  explicit ThreadPool(const string& name = string());
  ~ThreadPool();
  //启动线程池，其中有numThreads个线程
  void start(int numThreads);
  //关闭线程池
  void stop();
  //在任务队列queue_中添加任务
  void run(const Task& f);

 private:
  //线程池中的线程要执行的函数
  void runInThread();
  //获取任务
  Task take();
  //一个MutexLock类型的互斥变量mutex_
  MutexLock mutex_;
  //一个Condition类型的条件变量cond_
  Condition cond_;
  //线程池的名称
  string name_;
  //消费者线程队列threads_
  boost::ptr_vector<muduo::Thread> threads_;
  //任务队列queue_，其中task为一个执行任务的函数，生产者线程在queue_添加任务
  std::deque<Task> queue_;
  //running_表示线程池是否处于运行状态
  bool running_;
};

}

#endif

ThreadPool.cc

#include <muduo/base/ThreadPool.h>

#include <muduo/base/Exception.h>

#include <boost/bind.hpp>
#include <assert.h>
#include <stdio.h>

using namespace muduo;

ThreadPool::ThreadPool(const string& name)
  : mutex_(),
    cond_(mutex_),
    name_(name),
    running_(false)
{
}

ThreadPool::~ThreadPool()
{
  if (running_)
  {
    stop();
  }
}

void ThreadPool::start(int numThreads)
{
  assert(threads_.empty());
  running_ = true;
  threads_.reserve(numThreads);
  //添加消费者线程
  for (int i = 0; i < numThreads; ++i)
  {
    char id[32];
    snprintf(id, sizeof id, "%d", i);
    threads_.push_back(new muduo::Thread(
          boost::bind(&ThreadPool::runInThread, this), name_+id));
    //启动runInThread函数
	threads_[i].start();
  }
}

void ThreadPool::stop()
{
  {
  MutexLockGuard lock(mutex_);
  running_ = false;
  //通知所有等待线程
  cond_.notifyAll();
  }
  for_each(threads_.begin(),
           threads_.end(),
           boost::bind(&muduo::Thread::join, _1));
}

//将任务task添加到线程中的任务队列queue_
void ThreadPool::run(const Task& task)
{
  //消费者线程队列threads_为空
  if (threads_.empty())
  {
	//直接执行
    task();
  }
  else
  {
    MutexLockGuard lock(mutex_);
    queue_.push_back(task);
	//通知take()函数，有任务task可以获取
    cond_.notify();
  }
}

ThreadPool::Task ThreadPool::take()
{
  MutexLockGuard lock(mutex_);
  // always use a while-loop, due to spurious wakeup
  while (queue_.empty() && running_)
  {
	//等待函数run()添加任务
    cond_.wait();
  }
  Task task;
  if(!queue_.empty())
  {
    task = queue_.front();
    queue_.pop_front();
  }
  return task;
}

//runInThread()函数调用take()获取任务task
void ThreadPool::runInThread()
{
  try
  {
    while (running_)
    {
	  //阻塞在此函数，消费者线程中的runInThread()函数调用take()获取任务task
      Task task(take());
      //任务task不为空，而执行任务task
	  if (task)
      {
        task();
      }
    }
  }
  catch (const Exception& ex)
  {
    fprintf(stderr, "exception caught in ThreadPool %s\n", name_.c_str());
    fprintf(stderr, "reason: %s\n", ex.what());
    fprintf(stderr, "stack trace: %s\n", ex.stackTrace());
    abort();
  }
  catch (const std::exception& ex)
  {
    fprintf(stderr, "exception caught in ThreadPool %s\n", name_.c_str());
    fprintf(stderr, "reason: %s\n", ex.what());
    abort();
  }
  catch (...)
  {
    fprintf(stderr, "unknown exception caught in ThreadPool %s\n", name_.c_str());
    throw; // rethrow
  }
}