C++11实现半同步半异步的线程池
Posted Jqivin
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了C++11实现半同步半异步的线程池相关的知识，希望对你有一定的参考价值。
一、线程池的简单介绍
目的：减少线程创建和销毁的时候的系统资源的开销。减少线程上下文切换的开销。
引入的目的
在处理大量的并发任务的时候，如果按照传统的方式，一个任务开辟一个线程来进行处理，大量的线程创建和销毁将消耗过多的系统资源，而且增加了上下文切换的时间，所以通过引用线程池的概念来解决这些问题。
工作的原理
线程池技术通过在系统中预先创建一定数量的线程，当任务请求到来的时候，从线程池中分配一个预先创建号的线程去处理任务，线程在处理完成任务之后，并不会销毁，而是等待下次的任务的到来。
作用
避免大量的线程创建和销毁动作，从而节省系统资源，这样做的一个好处是，对于多核处理器，由于线程会被分配到多个CPU，会提高并行处理的效率。另一个好处是每个线程独立阻塞，可以防止主线程被阻塞而使主流程被阻塞，导致其他的请求得不到响应的问题。
二、半同步半异步线程池
1.基本结构
同步服务层:负责处理来自上层的请求，上层的请求可能是并发的，这些请求并不会直接进行处理，而是添加到同步队列中。
同步排队层 :来自上层的处理都会添加到排队层的同步队列中等待处理。
异步服务层:在这层会有多个线程同时处理排队层中的任务，异步服务层从同步队列中取出任务并进行处理。
这种三层的结构可以最大程度处理上层的并发请求。对于上层来说只要将任务丢到同步队列中就行了，至于谁去处理，什么时候处理都不用关心，主线程也不会阻塞，还能继续发起新的请求。至于任务具体怎么处理，这些细节都是靠异步服务层的多线程异步并行来完成的，这些线程是一开始就创建的，不会因为大量的任务到来而创建新的线程，避免了频繁创建和销毁线程导致的系统开销，而且通过多核处理能大幅提高处理效率。
2.工作原理
从活动图中可以看到线程池的活动过程，一开始线程池会启动一定数量的线程，这些线程属于异步层，主要用来并行处理排队层中的任务，如果排队层中的任务数为空，则这些线程等待任务的到来，如果发现排队层中有任务了，线程池则会从等待的这些线程中唤醒一个来处理新任务。同步服务层则会不断地将新的任务添加到同步排队层中，这里有个问题值得注意，有可能上层的任务非常多，而任务又是非常耗时的，这时，异步层中的线程处理不过来，则同步排队层中的任务会不断增加，如果同步排队层不加上限控制，则可能会导致排队层中的任务过多，内存暴涨的问题。因此，排队层需要加上限的控制，当排队层中的任务数达到上限时，就不让上层的任务添加进来，起到限制和保护的作用。
三、结构实现
1.同步队列
作用：保证队列中共享数据的安全性（执行操作时要加锁），为上一层同步服务层提供添加任务的接口，为下一层异步服务层提供提取任务的接口。
代码：
template <class T>
class SyncQueue
{
private:
std::list<T> m_queue;
std::mutex m_mutex;
std::condition_variable m_notEmpty;
std::condition_variable m_notFull;
int max_size;
bool m_needStop;
public:
SyncQueue(int size):max_size(size),m_needStop(false){}
~SyncQueue(){}
//这里使用两个Put函数，因为传参的时候可能是左值，也可能是右值
void Put(const T& t)
{
Add(t); //实现代码复用，下面的还有一个函数
}
void Put(T&& t)
{
Add(std::forward<T>(t));
}
void Take( T& t) //取走队列中的一个任务
{
std::unique_lock<mutex> locker(m_mutex);
while (!m_needStop && !NotEmpty())
{
m_notEmpty.wait(locker);
}
if (m_needStop)
return;
t = m_queue.front();
m_queue.pop_front();
m_notFull.notify_one(); //挪走资源了，所以通知添加任务的线程开始工作，这个notfull是由于满了而阻塞的队列
}
void Take(list<T>& list) //取走队列中的全部
{
std::unique_lock<mutex> locker(m_mutex);
while (!m_needStop && !NotEmpty())
{
m_notEmpty.wait(locker);
}
if (m_needStop)
return;
list = std::move(m_queue);
m_notFull.notify_one();
}
void Stop()
{
{
std::lock_guard<std::mutex> locker(m_mutex);
m_needStop = true;
}
m_notEmpty.notify_all();
m_notFull.notify_all();
}
bool IsEmpty()
{
std::unique_lock<std::mutex> locker(m_mutex);
return m_queue.empty();
}
bool IsFull()
{
std::unique_lock<std::mutex> locker(m_mutex);
return m_queue.size() >= max_size;
}
size_t Size() const
{
std::lock_guard<std::mutex> locker(m_mutex);
return m_queue.size();
}
private:
bool NotEmpty()
{
//std::lock_guard<std::mutex> locker(m_mutex); 不要加锁，这个函数是被Add调用的，Add中已经加锁了
bool res = m_queue.empty();
if (res)
cout << "同步队列为空，需要等待，异步线程的id为：" << this_thread::get_id() << endl;
return !res;
}
bool NotFull()
{
bool res = m_queue.size() >= max_size;
if (res)
cout << "同步队列满了，需要等待，同步线程的id为：" << this_thread::get_id() << endl;
return !res;
}
template<class F>
void Add(F&& f) //右值引用，尽可能减少内存的拷贝构造
{
std::unique_lock<mutex> locker(m_mutex);
while (!m_needStop && !NotFull()) //同步队列不需要停（在工作），并且同步队列满了
{
m_notFull.wait(locker);
}
if (m_needStop) //同步队列不工作了
{
return;
}
m_queue.push_back(f);
m_notEmpty.notify_one();
}
};
结构：
2.线程池
介绍：
一个完整的线程池包括三层：同步服务层，排队层，异步服务层。是一种生产者消费者模型，同步层不停地向同步队列中添加任务，异步服务层不停地从队列中取出任务并通过线程池中的线程组中的线程来执行。排队层是一个同步队列，他的内部保证了上下两层对共享数据（任务）的安全访问（通过互斥锁和条件变量），同时还要保证队列不能不做限制的添加任务导致内存暴涨。除此之外，线程池还要提供一个停止的接口，让用户能够在需要的时候停止线程池的工作。
代码：
int max_task = 10;
class ThreadPool
{
using Task = std::function<void()>;
public:
ThreadPool(int numsThread = std::thread::hardware_concurrency()):m_queue(max_task)
{
Start(numsThread);
}
~ThreadPool()
{
//如果没有停止，调用停止
Stop();
}
void Stop()
{
std::call_once(m_flag, [this] {StopThreadGroup(); }); //只能调动一次
}
void AddTask(Task&& task)
{
m_queue.Put(task);
}
void AddTask(const Task& task)
{
m_queue.Put(task);
}
private:
void Start(int nums)
{
m_running = true;
//创建线程组
for (int i = 0; i < nums; i++)
{
m_threadGroup.push_back(std::make_shared<thread>(&ThreadPool::RunInThread, this));
}
}
void RunInThread() //线程的特定函数
{
//这种写法有的时候会报错
/*
while (m_running)
{
Task task;
if ( !m_queue.IsEmpty() ) //error
{
m_queue.Take(task);
if (!m_running)
return;
task();
}
}
*/
while (m_running)
{
list<Task> list;
m_queue.Take(list);
for (auto& task : list)
{
if (!m_running)
return;
task();
}
}
}
void StopThreadGroup()
{
m_queue.Stop(); //让同步队列中停止工作（Put,Take停止工作）
m_running = false; //置为false。让内部的线程跳出循环并退出，（RunINThread）
for (auto thread : m_threadGroup)
{
if (thread)
{
thread->join(); //join之后在调用thread类析构函数就可以了，如果不join会报错
}
}
m_threadGroup.clear();
}
private:
std::list<std::shared_ptr<thread>> m_threadGroup; //线程组
SyncQueue<Task> m_queue; //同步队列
atomic<bool> m_running; //是否停止的标志
std::once_flag m_flag;
};
在上面的例子中，ThreadPool有3个成员变量。
第一个是`线程组`，这个线程组中的线程是预先创建的，应该创建多少个线程由外面
传入，一般建议创建CPU核数的线程以达到最优的效率，线程组循环从同步队列中取出任务并执行，如果线程池为空，线程组将处于等待状态，等待任务的到来。
第二个成员变量是`同步队列`，它不仅用来做线程同步，还用来限制同步队列的上限，这个上限也是由使用者设置的。
第三个成员变量是用来`停止线程池`的，为了保证线程安全，我们用到了原子变量atomic_bool。
关于RunInThread函数
void RunInThread() //线程的特定函数
{
//这种写法有的时候会报错
/*
while (m_running)
{
Task task;
if ( !m_queue.IsEmpty() ) //error
{
m_queue.Take(task);
if (!m_running)
return;
task();
}
}
*/
while (m_running)
{
list<Task> list;
m_queue.Take(list);
for (auto& task : list)
{
if (!m_running)
return;
task();
}
}
}
第一种写法是错误的，因为if语句可能判断结果不为空，但是进去之后可能切换到其他异步执行线程，别人把最后一个任务取走了，这时候如果再切换回来，就会出错了。
第二种情况是每个线程直接将此时刻队列中的任务都取出来，由这个线程循环处理。这样会不会造成只有一个线程工作呢？当然是不会的，因为我们创建的默认的线程都是轮询线程，不可能一次将6个任务都添加到队列中，所以我们只是每个线程取了当前时刻的任务（可能只有一个，两个或者没有），后序还会添加。
解决方案：
（1）加锁
（2）执行task之前进行判断，如果不为空就执行（这时候可以将if语句去掉）
if (task)
{
task();
}
四、应用实例
线程程池将初始创建两个线程，然后外部线程将不停地向线程中添加新任务，线程池内部的线程将会并行处理同步队列中的任务。下面来看看这个例子。
代码
//任务
void task1()
{
cout << "task1 " << this_thread::get_id() << endl;
}
void task2()
{
cout << "task2 " << this_thread::get_id() << endl;
}
//使用两个线程往同步队列中添加任务
void fun1(ThreadPool& pool)
{
for (int i = 0; i < 3; i++)
{
auto theId = this_thread::get_id();
cout << "同步线程1的id:" << theId << endl;
pool.AddTask(task1);
}
}
void fun2(ThreadPool& pool)
{
for (int i = 0; i < 3; i++)
{
auto theId = this_thread::get_id();
cout << "同步线程2的id:" << theId << endl;
pool.AddTask(task2);
}
}
int main()
{
ThreadPool pool(2);
thread sycnth1(fun1, std::ref(pool));
thread sycnth2(fun2, std::ref(pool));
this_thread::sleep_for(std::chrono::seconds(1));
sycnth1.join();
sycnth2.join();
return 0;
}
首先，创建一个线程池，让其产生两个工作线程（异步层的执行线程）。然后创建两个同步线程sycnth1，sycnth2，分别对应执行函数fun1，fun2,。这两个线程负责添加任务。
执行结果: 
由结果可以看出，两个异步线程：25888,14508。在初始时，由于任务队列里没有任务，所以，两个线程进入等待状态，直到队列中有数据的时候才开始处理数据。两个同步线程：10108，11680。这两个线程不断地向线程池中添加任务。最终的结果是，异步服务层的线程交替的处理来自上层的任务。
参考资料：《深化应用C++11代码优化与工程级应用》
以上是关于C++11实现半同步半异步的线程池的主要内容，如果未能解决你的问题，请参考以下文章