muduo源码分析：ThreadPool 线程池的实现

傷城~ 2022-04-22 02:34 196阅读 0赞

源码:

https://github.com/chenshuo/muduo/blob/master/muduo/base/ThreadPool.h
    
    https://github.com/chenshuo/muduo/blob/master/muduo/base/ThreadPool.cc

线程池ThreadPool用到了前面分析的Thread、MutexLock、Condition。ThreadPool可以设置工作线程的数量，并向任务队列放入任务。放入到任务队列中的任务将由某个工作线程执行。

--------------------

### ThreadPool.h ###

public:
      typedef boost::function<void ()> Task;
    
      explicit ThreadPool(const string& nameArg = string("ThreadPool"));
      ~ThreadPool();
    
      // Must be called before start().
      void setMaxQueueSize(int maxSize) { maxQueueSize_ = maxSize; }	//设置线程池线程最大数目大小
      void setThreadInitCallback(const Task& cb)				//设置线程执行前的回调函数
      { threadInitCallback_ = cb; }
    
      void start(int numThreads);        //启动线程池,numThreads是线程池的容量
      void stop();                       //终止线程池
    
      const string& name() const
      { return name_; }
    
      size_t queueSize() const;
    
      // Could block if maxQueueSize > 0
      void run(const Task& f);
    #ifdef __GXX_EXPERIMENTAL_CXX0X__
      void run(Task&& f);
    #endif

**成员变量**

private:
      bool isFull() const REQUIRES(mutex_);			//判满
      void runInThread();					//线程池的线程运行函数
      Task take();						//取任务函数
    
      mutable MutexLock mutex_;				//mutable表示在const函数也可以改变它
      Condition notEmpty_ GUARDED_BY(mutex_);		//任务队列queue_不为空了，有任务可以执行了，进而唤醒等待的线程。
      Condition notFull_ GUARDED_BY(mutex_);		//任务队列queue_不满了，有空间可以使用了，进而唤醒等待的线程。
      string name_;
      Task threadInitCallback_;				//线程初始化回调函数
      boost::ptr_vector<muduo::Thread> threads_;	        //工作线程容器(线程数组)
      std::deque<Task> queue_ GUARDED_BY(mutex_);	        //任务队列
      size_t maxQueueSize_;				        //队列最大大小
      bool running_;					//线程池运行标志

使用[boost::ptr\_vector][boost_ptr_vector]存放Thead。

每个Task都是`typedef boost::function<void ()> Task; 所有任务都放到`queue\_`中。需要使用条件变量来维护线程将的同步，比如：通知其他线程有任务到来了，可以向任务队列放任务了等等。`

--------------------

**ThreadPool::ThreadPool()**

ThreadPool::ThreadPool(const string& nameArg)
      : mutex_(),
        notEmpty_(mutex_),
        notFull_(mutex_),
        name_(nameArg),
        maxQueueSize_(0),
        running_(false)
    {
    }

构造函数对成员变量进行初始化（使用初始化列表）。

**ThreadPool::~ThreadPool()**

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

析构函数会调用stop， 唤醒所有休眠的线程，然后等待所有线程处理完。

**ThreadPool::stop()**

void ThreadPool::stop()    //终止线程池
    {
      { // new scope
      MutexLockGuard lock(mutex_); // ctor of MutexLockGuard will lock mutex_
      running_ = false;
      notEmpty_.notifyAll();     // 唤醒所有休眠的工作线程
      } // dtor of MutexLockGuard will unlock mutex_
      for_each(threads_.begin(),
               threads_.end(),
               boost::bind(&muduo::Thread::join, _1)); // 等待所有工作线程结束
    }

--------------------

### ThreadPool::start() ###

void ThreadPool::start(int numThreads)    //参数为线程数量,会创建相应数量的线程,线程函数为ThreadPool::runInThread
    {
      assert(threads_.empty());
      running_ = true;              //启动标志
      threads_.reserve(numThreads); // 保证threads_容量至少为numThreads
      for (int i = 0; i < numThreads; ++i)
      {
        char id[32];
        snprintf(id, sizeof id, "%d", i+1);
    
    //创建工作线程并加入线程数组,构造Thread(this.runInThread,name+id)并加入线程数组。线程函数是ThreadPool::runInThread 
        threads_.push_back(new muduo::Thread(
              boost::bind(&ThreadPool::runInThread, this), name_+id));
        threads_[i].start();    //启动每个线程，但是由于线程运行的函数是runInThread，所以会阻塞
      }
      if (numThreads == 0 && threadInitCallback_)
      {
        threadInitCallback_();
      }
    }

参数为线程数量，会创建相应数量的线程，执行体为**ThreadPool::runInThread**。

void ThreadPool::runInThread()    //线程函数
    {
      try
      {
        if (threadInitCallback_)    //如果设置了就执行，进行一些初始化设置
        {
          threadInitCallback_();
        }
        while (running_)        //当线程池启动之后,就在while循环中不停地取任务执行
        {
          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
      }
    }

--------------------

### 获取一个task ###

ThreadPool::Task ThreadPool::take()	//获取一个task
    {
      MutexLockGuard lock(mutex_);
      // always use a while-loop, due to spurious wakeup
      while (queue_.empty() && running_)
      {
        notEmpty_.wait();	//没有任务，则等待（利用条件变量）
      }
      Task task;
      if (!queue_.empty())	//有任务了，就返回一个任务
      {
        task = queue_.front();	//获取任务
        queue_.pop_front();
        if (maxQueueSize_ > 0)
        {
          notFull_.notify();	//当解决了一个任务之后,任务队列肯定不是满的,通知某个等待向队列放入task线程。
        }
      }
      return task;			//返回任务
    }

条件变量的wait操作使用while包裹，预防“虚假唤醒”（如被其他线程抢占了）。

--------------------

### 向线程池添加task ###

void ThreadPool::run(const Task& task)	//向线程池添加task
    {
      if (threads_.empty()
      {
        task();			//如果没有子线程，就在主线程中执行该task
      }
      else
      {
        MutexLockGuard lock(mutex_);
        while (isFull())	    //如果task队列queue_满了，就等待
        {
          notFull_.wait();
        }
        assert(!isFull());
    
        queue_.push_back(task);	//将任务加入队列
        notEmpty_.notify();		//当添加了某个任务之后,任务队列肯定不是空的,通知某个等待从queue_中取task的线程
      }
    }

### 使用示例 ###

struct Foo {
    public:
        void DoWork() {
            std::cout << "run member function in thread:" << CurrentThread::tid() << std::endl;
        }
        void operator() (){
            std::cout << "run functor in thread:" << CurrentThread::tid() << std::endl;
        }
    };
    
    void Task1()
    {
        std::cout << "function run in thread:"  << CurrentThread::tid() << std::endl;
    }
    
    
    int main()
    {
        ThreadPool tp("TestThreadPool");
        tp.setMaxQueueSize(10);
    
        tp.start(4); // 启动4个工作线程，启动之后，由于任务队列queue_为空，所以所有工作线程都休眠了
        tp.run(Task1); // 放入一个task，会唤醒某个工作线程
    
        Foo f;
        tp.run(boost::bind(&Foo::DoWork, &f));
        tp.run(f);
    
        tp.run( [](){ std::cout << "lambda function run in thread:" << CurrentThread::tid() << std::endl; });
    
        typedef void(*pFunc)();
        pFunc pf = Task1;
        tp.run(pf);
    }

可以看到，ThreadPool可以很方便的将某个task放到任务队列中，该task会由某个线程执行。task使用boost::function表示，可以方便地将函数指针、普通函数、成员函数（结合boost::bind）、lambda、重载了函数调用运算符‘()’的类的对象（这些统称为可调用对象）放入到任务队列当中，非常方便。

[boost_ptr_vector]: http://www.boost.org/doc/libs/1_61_0/libs/ptr_container/doc/ptr_container.html