muduo网络库源码学习———

muduo库里面的线程池是固定线程池，即创建的线程池里面的线程个数是一定的，不是动态的。线程池里面一般要包含线程队列还有任务队列，外部程序将任务存放到线程池的任务队列中，线程池中的线程队列执行任务，也是一种生产者和消费者模型。muduo库中的线程池源码如下：

线程池头文件ThreadPool.h

//线程池

// Use of this source code is governed by a BSD-style license

// that can be found in the License file.

//

// Author: Shuo Chen (chenshuo at chenshuo dot com)

#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();

//启动线程池

  void start(int numThreads);

//关闭线程池

  void stop();

//运行任务，往线程池当中的任务队列添加任务

  void run(const Task& f);

 private:

    //线程池当中的线程要执行的函数

  void runInThread();

    //获取任务

  Task take();

  MutexLock mutex_;//和条件变量配合使用的互斥锁

  Condition cond_;//条件变量用来唤醒线程池中的线程队列来执行任务

  string name_;//线程池名称

  boost::ptr_vector<muduo::Thread> threads_;//存放线程指针

  std::deque<Task> queue_;//任务队列

  bool running_;//线程池是否处于运行的状态

};

}

#endif

线程池实现文件ThreadPool.cc

// Use of this source code is governed by a BSD-style license

// that can be found in the License file.

//

// Author: Shuo Chen (chenshuo at chenshuo dot com)

#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)

  {//for循环创建线程

    char id[32];

    //线程号

    snprintf(id, sizeof id, "%d", i);

    //创建线程并存放线程指针，绑定的函数为runInThread

    threads_.push_back(new muduo::Thread(boost::bind(&ThreadPool::runInThread, this), name_+id));

    threads_[i].start();//启动线程，即runInThread函数执行

  }

}

//关闭线程池

void ThreadPool::stop()

{

  {

  MutexLockGuard lock(mutex_);

  running_ = false;//running置为false

  cond_.notifyAll();//通知所有线程

  }

//等待线程退出

  for_each(threads_.begin(),threads_.end(),boost::bind(&muduo::Thread::join, _1));

}

//添加任务

void ThreadPool::run(const Task& task)

{//将任务添加到线程池当中的任务队列

  if (threads_.empty())//如果线程池当中的线程是空的

  {

    task();//直接执行任务

  }

  else//否则添加

  {

    MutexLockGuard lock(mutex_);

    queue_.push_back(task);

    cond_.notify();//通知队列当中有任务了

  }

}

//获取任务函数

ThreadPool::Task ThreadPool::take()

{//加锁保护

  MutexLockGuard lock(mutex_);

  // always use a while-loop, due to spurious wakeup

  //如果队列为空并且处于运行的状态

  while (queue_.empty() && running_)

  {

    cond_.wait();//等待

  }

  Task task;//定义任务变量，Task是一个函数类型

  if(!queue_.empty())//有任务到来

  {

    task = queue_.front();//取出任务

    queue_.pop_front();//弹出任务

  }

  return task;//返回任务

}

void ThreadPool::runInThread()

{

  try//可能发生异常

  {

    while (running_)

    {//获取任务

      Task task(take());

      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

  }

}

下面是测试代码：

ThreadPool_test.cc

//线程池测试代码

#include <muduo/base/ThreadPool.h>

#include <muduo/base/CountDownLatch.h>

#include <muduo/base/CurrentThread.h>

#include <boost/bind.hpp>

#include <stdio.h>

void print()

{//简单地打印tid

  printf("tid=%d\n", muduo::CurrentThread::tid());

}

void printString(const std::string& str)

{

  printf("tid=%d, str=%s\n", muduo::CurrentThread::tid(), str.c_str());

}

int main()

{//创建一个线程池

  muduo::ThreadPool pool("MainThreadPool");

  //5个线程的线程池

  pool.start(5);

  //添加了2个任务运行print

  pool.run(print);

  pool.run(print);

  //添加了100个任务

  for (int i = 0; i < 100; ++i)

  {

    char buf[32];

    snprintf(buf, sizeof buf, "task %d", i);

    //绑定的函数是带参数的

    pool.run(boost::bind(printString, std::string(buf)));

  }

  //创建CountDownLatch对象，计数值count =1，只需执行一个countDown

  muduo::CountDownLatch latch(1);

  //添加一个任务

  pool.run(boost::bind(&muduo::CountDownLatch::countDown, &latch));

  //count不为0的时候一直等待

  latch.wait();

  //关闭线程池

  pool.stop();

}

执行结果如下：

巴特西

muduo网络库源码学习————线程池实现

最新文章

热门文章