1. 執行緒池

執行緒池的問題本質上也是生產者消費者模型問題

1. 生產者生產產品的過程，實際上就是由程式設計師向任務佇列中新增任務的過程(需要程式設計師控制)，實現程式碼見下：

  1. print函式是程式設計師自己手動定義的任務函式
  2. run(Task task)介面的功能是將task新增到任務佇列中，並喚醒等待任務的執行緒
  pool.run(print); //該句話表示：程式設計師手動將print任務新增到任務佇列
  
  for (int i = 0; i < 100; ++i)
  {
    char buf[32];
    snprintf
 
(buf, sizeof buf, "task %d", i);
    pool.run(std::bind(printString, std::string(buf)));
  }

2. 消費者是執行緒池中的執行緒，(不需要程式設計師控制)，當任務佇列中有任務時，將會喚醒執行緒池中的空閒執行緒
   

2. muduo執行緒池程式碼詳解：ThreadPool

// ThreadPool.h檔案
class ThreadPool : noncopyable
{
 public:
  typedef std::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);
  void stop
 
();

  const string& name() const
  { return name_; }

  size_t queueSize() const;

  // Could block if maxQueueSize > 0
  void run(Task f);

 private:
  bool isFull(); //任務佇列滿
  void runInThread();
  Task take();

  mutable MutexLock mutex_;
  Condition notEmpty_;
  Condition notFull_;
  string name_;
  
  Task threadInitCallback_;
  
  std::vector<std::unique_ptr<muduo::Thread>> threads_;
  
  std::deque<Task> queue_;
  size_t maxQueueSize_; //任務佇列的容量
  
  bool running_;
};

// ThreadPool.cc檔案
#include <muduo/base/ThreadPool.h>

#include <muduo/base/Exception.h>

#include <assert.h>
#include <stdio.h>

using namespace muduo;

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

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

// 1. 建立numThreads個執行緒，並放入執行緒佇列
// 2. 呼叫threads_[i]->start()，使每個執行緒都啟動執行緒函式runInThread()
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+1);
    threads_.emplace_back(new muduo::Thread(
          std::bind(&ThreadPool::runInThread, this), name_+id));
    threads_[i]->start();
  }
  if (numThreads == 0 && threadInitCallback_)
  {
    threadInitCallback_();
  }
}

void ThreadPool::runInThread()
{
  try
  {
    if (threadInitCallback_)
    {
      threadInitCallback_();
    }
    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::Task ThreadPool::take()
{
  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();
    }
  }
  return task;
}

void ThreadPool::stop()
{
  {
    MutexLockGuard lock(mutex_);
    running_ = false;
    notEmpty_.notifyAll();//通知所有的等待執行緒
  }
  
  for (auto& thr : threads_) //等待所有的執行緒執行完任務後再退出
  {
    thr->join();
  }
  //等價於  for_each(thread_.begin(),thread_.end(),
		//      boost::bind(&muduo::Thread::join,_1));
}

size_t ThreadPool::queueSize() const
{
  MutexLockGuard lock(mutex_);
  return queue_.size();
}

//
void ThreadPool::run(Task task)
{
  if (threads_.empty()) //建立了0個執行緒，即沒有執行緒池，只有一個主執行緒
  {
    task(); //主執行緒強制執行該任務
  }
  else// 如果執行緒池中有空閒執行緒，就將任務新增到任務佇列
  {
    MutexLockGuard lock(mutex_);
    while (isFull())
    {
      notFull_.wait();
    }
    assert(!isFull());

    queue_.push_back(std::move(task));
    notEmpty_.notify();
  }
}

bool ThreadPool::isFull() const //任務佇列滿了
{
  mutex_.assertLocked();
  return maxQueueSize_ > 0 && queue_.size() >= maxQueueSize_;
}

3. 執行緒池測試示例

#include <muduo/base/ThreadPool.h>
#include <muduo/base/CountDownLatch.h>
#include <muduo/base/CurrentThread.h>
#include <muduo/base/Logging.h>

#include <stdio.h>
#include <unistd.h>  // usleep

void print()
{
  printf("tid=%d\n", muduo::CurrentThread::tid());
}

void printString(const std::string& str)
{
  LOG_INFO << str;
  usleep(100*1000);
}

void test(int maxSize)
{
  LOG_WARN << "Test ThreadPool with max queue size = " << maxSize;
  muduo::ThreadPool pool("MainThreadPool");
  pool.setMaxQueueSize(maxSize); //設定任務佇列的大小
  pool.start(5); //建立5個執行緒，並啟動執行緒

  LOG_WARN << "Adding";
  //向執行緒池中新增自定義的無參的print任務，並喚醒執行緒池中的空閒執行緒
  pool.run(print); 
  pool.run(print);
  for (int i = 0; i < 100; ++i)
  {
    char buf[32];
    snprintf(buf, sizeof buf, "task %d", i);
    //向執行緒池中新增自定義的有引數的printString任務，並喚醒執行緒池中的空閒執行緒
    pool.run(std::bind(printString, std::string(buf)));
  }
  LOG_WARN << "Done";

  //向執行緒池中新增CountDownLatch類中的成員函式countDown()，並喚醒執行緒池中的空閒執行緒
  muduo::CountDownLatch latch(1);
  pool.run(std::bind(&muduo::CountDownLatch::countDown, &latch));
  latch.wait();
  
  pool.stop();
}

int main()
{
  test(0);
  test(1);
  test(5);
  test(10);
  test(50);
}