C++11半同步半异步线程池

• 线程池为何产生
• 半同步半异步线程池
• 线程池实现
• • 同步队列
  • 线程池

在处理大量并发任务的时候，如果按照传统的方法一个请求一个线程来处理请求任务，线程的创建和销毁将消耗过多的系统资源，还增加了线程上下文切换的开销，线程池技术可以很好的解决这些问题，线程池技术通过在系统中预先创建一定数量的线程，当任务请求到来的时候从线程池中分配一个预先创建的线程去处理任务，线程在处理完任务以后还可以重用，通过线程池能避免大量的线程创建和摧毁动作

线程池分为半同步半异步线程池和领导者追随者线程池，本文主要通过C++11实现一个半同步半异步的线程池
半同步半异步的线程池分为3层：
第一层是同步服务层，它处理来自上层的任务请求，上层的请求可能是并发的，这些请求不是马上就会被处理，而是将这些任务放到一个同步排队层中，等待处理，第二层是同步排队层，来自上层的任务请求都会排队层中等待处理。第三层是异步服务层，这一层中会有多个线程同时处理排队层中的任务，异步服务层从同步排队层中取出任务并行处理

在实现时，排队层就是一个同步队列，允许多个线程同时去添加或者取出任务，并且要保证操作过程是安全的。一开始线程池会启动一定数量的线程，这些线程属于异步层，主要用来并行处理排队层中的任务，如果排队层中的任务数为空，则等待，如果发现排队层中有任务了，线程池则会从等待的这些线程中唤醒一个来处理新任务，需要对排队队列的任务数量进行上限控制，避免任务过多导致内存暴涨的问题

使用了C++11的锁，条件变量，右值引用，std::move,以及std::forward
代码如下：

#include
#include
#include
#include
#include
using namespace std;
template
class SyncQueue
{
public:
  SyncQueue(int maxSize):m_maxSize(maxSize),m_needStop(false)
  {
  }
  void Put(const T& x)
  {
    Add(x);
  }
  void Put(T&& x)
  {
    Add(std::forward(x));
  }

  void Take(std::list& list)
  {
    std::unique_lock locker(m_mutex);
    m_notEmpty.wait(locker,[this]{return m_needStop||NotEmpty();});
    if(m_needStop)
      return;
    list=std::move(m_queue);
    m_notFull.notify_one();
  }
  void Take(T& t)
  {
    std::unique_lock locker(m_mutex);
    m_notEmpty.wait(locker,[this]{return m_needStop||NotEmpty();});
    if(m_needStop)
      return;
    t=m_queue.front();
    m_queue.pop_front();
    m_notFull.notify_one();
  }
  void Stop()
  {
    {
      std::lock_guard locker(m_mutex);
      m_needStop=true;
    }
    m_notFull.notify_all();
    m_notEmpty.notify_all();
  }
  bool Empty()
  {
    std::lock_guard locker(m_mutex);
    return m_queue.empty();
  }
  bool Full()
  {
    std::lock_guard locker(m_mutex);
    return m_queue.size()==m_maxSize;
  }
  size_t Size()
  {
    std::lock_guard locker(m_mutex);
    return m_queue.size();

  }
  int Count()
  {
    return m_queue.size();
  }
private:
  bool NotFull() const
  {
    bool full=m_queue.size() >=m_maxSize;
    if(full)
      cout<<"缓冲区满了，需要等待..."<
    bool empty=m_queue.empty();
    if(empty)
      cout<<"缓冲区空了，需要等待...,异步层的线程ID： "<
  void Add(F&& x)
  {
    std::unique_lock locker(m_mutex);
    m_notFull.wait(locker,[this]{return m_needStop||NotFull();});
    if(m_needStop)
      return;
    m_queue.push_back(std::forward(x));
    m_notEmpty.notify_one();
  }
private:
  std::list m_queue;
  std::mutex m_mutex;
  std::condition_variable m_notEmpty;
  std::condition_variable m_notFull;
  int m_maxSize;
  bool m_needStop;
};

Take函数：创建一个unique_lock获取mutex,然后再通过条件变量m_notEmpty来等待判断式，判断式由两个条件组成，一个是停止的标志，另一个是不为空的条件，当不满足任何一个条件时，条件变量会释放mutex并将线程置于waiting状态，等待其他线程调用notify_one/notify_all唤醒时，条件变量会先重新获取mutex,然后在检查条件是否满足，如果满足，则往下执行，如果不满足，则释放mutex继续等待.

Add函数：Add的过程和Take的过程是类似的

Stop函数：Stop函数先获取mutex,然后将停止标志为true，为了保证线程安全，这里需要先获取mutex,在将其标志为true之后，再唤醒所有等待的线程，由于线程在m_needStop为true时会退出，所以所有的等待线程会相继退出

代码如下：

#include"SyncQueue.hpp"
#include
#include
#include
#include
#include 
const int MaxTaskCount=100;
class ThreadPool
{
public:
  using Task=std::function;
  ThreadPool(int numThreads):m_queue(MaxTaskCount)
  {
    Start(numThreads);
  }
  ~ThreadPool(void)
  {
    Stop();
  }
  void Stop()
  {
    std::call_once(m_flag,[this]{StopThreadGroup();});
  }
  void AddTask(Task&& task)
  {
    m_queue.Put(std::forward(task));

  }
  void AddTask(const Task& task)
  {
    m_queue.Put(task);
  }
  void Start(int numThreads)
  {
    m_running=true;
    for(int i=0;i
      m_threadgroup.push_back(std::make_shared(&ThreadPool::RunInThread,this));
    }
  }
  void RunInThread()
  {
    while(m_running)
    {
      std::list list;
      m_queue.Take(list);
      for(auto& task:list)
      {
        if(!m_running)
          return;
        task();
      }

    }
  }
  void StopThreadGroup()
  {
    m_queue.Stop();
    m_running=false;
    for(auto thread:m_threadgroup)
    {
      if(thread)
        thread->join();
    }
    m_threadgroup.clear();
  }
private:
  std::list> m_threadgroup;
  SyncQueue m_queue;
  atomic_bool m_running;
  std::once_flag m_flag;
};
int main()
{
  ThreadPool pool(2);
  this_thread::sleep_for(std::chrono::seconds(1));
  std::thread thd1([&pool]{
        for(int i=0;i<10;i++)
        {
          auto thdId=this_thread::get_id();
          pool.AddTask([thdId]{
                cout<<"同步层线程1的线程ID："<
        for(int i=0;i<10;i++)
        {
          auto thdId=this_thread::get_id();
          pool.AddTask([thdId]{
                cout<<"同步层线程2的线程ID: "< 
ThreadPool有3个成员变量，一个是线程组，这个线程组中的线程是预先创建的，应该创建多少个线程由外面传入，另外一个成员变量是同步队列，它不仅用来做线程同步，还用于限制同步队列的上限，第三个变量用于停止线程池的，为了保证线程安全，用到了原子变量stomic_bool
 
main函数里面写了一个应用实例，在这个例子，线程池将初始创建2个线程，然后外部线程将不断地向线程中添加新任务，线程池内部的线程将会并行处理同步队列中的任务
 
如图所示：
 

 由测试结果可知，线程池初始创建了两个内部的线程，由于初始时，线程池中的同步队列是空的，两个线程将进入等待状态，直到有数据才处理，满了之后也会等待，不会允许无限制地添加任务。