感受:
隨著深入學(xué)習(xí),,現(xiàn)代c++給我?guī)?lái)越來(lái)越多的驚喜…
c++真的變強(qiáng)大了,。
半同步半異步線程池:
其實(shí)很好理解,分為三層
同步層:通過(guò)IO復(fù)用或者其他多線程多進(jìn)程等不斷的將待處理事件添加到隊(duì)列中,,這個(gè)過(guò)程是同步進(jìn)行的,。
隊(duì)列層:所有待處理事件都會(huì)放到這里。上一層事件放到這里,,下一層從這里獲取事件
異步層:事先創(chuàng)建好線程,,讓線程不斷的去處理隊(duì)列層的任務(wù),上層不關(guān)心這些,,它只負(fù)責(zé)把任務(wù)放到隊(duì)列里,,所以對(duì)上層來(lái)說(shuō)這里是異步的。
補(bǔ)充下思路:
主要是后兩層
隊(duì)列層:c++11 通過(guò)std::function可以將函數(shù)封裝為對(duì)象,,那么我們一個(gè)函數(shù)也就是一個(gè)任務(wù),,通過(guò)vector或list等容器來(lái)存儲(chǔ)這些”任務(wù)”來(lái)供后面存取。因?yàn)闀?huì)出現(xiàn)競(jìng)爭(zhēng)資源的問(wèn)題,,所以我們要加鎖,,并且通過(guò)條件變量的條件來(lái)喚醒其他阻塞在鎖上的線程,,當(dāng)然你想避免線程阻塞浪費(fèi)資源可以用帶時(shí)間的鎖std::time_mutex,。
異步層:c++11 將線程也封裝為了對(duì)象,,那么我們創(chuàng)建一個(gè)容器保存線程對(duì)象,讓他們?nèi)リ?duì)列層取任務(wù)并執(zhí)行,,執(zhí)行完并不結(jié)束該線程而是歸還給容器(線程池),。
看張圖:
如果你不熟悉c++11的內(nèi)容
以下文章僅供參考
c++11 多線程
c++11 智能指針
c++11 對(duì)象移動(dòng)
c++11 lambda,bind,function
代碼:
同步隊(duì)列:
#include <list>
#include <mutex>
#include <thread>
#include <condition_variable>
#include <iostream>
template<typename T>
class SynQueue
{
public:
SynQueue(int maxsize):
m_maxSize(maxsize), m_needStop(false) { }
//添加事件,左值拷貝和右值移動(dòng)
void Put(const T&x)
{
//調(diào)用private內(nèi)部接口Add
Add(x);
}
void Put(T &&x)
{
Add(x);
}
//從隊(duì)列中取事件,取所有事件
void Take(std::list<T> &list)
{
//有wait方法必須用unique_lock
//unique_lock有定時(shí)等待等功能,,lock_guard就僅僅是RAII手法的互斥鎖
//但unique_lock的性能稍低于lock_guard
std::unique_lock<std::mutex> 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();
}
//取一個(gè)事件
void Take(T &t)
{
std::unique_lock<std::mutex> 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();
t();
}
//停止所有線程在同步隊(duì)列中的讀取
void Stop()
{
{
std::lock_guard<std::mutex> locker(m_mutex);
m_needStop = true;
}
m_notFull.notify_all();
m_notEmpty.notify_all();
}
//隊(duì)列為空
bool Empty()
{
std::lock_guard<std::mutex> locker(m_mutex);
return m_queue.empty();
}
//隊(duì)列為滿
bool Full()
{
std::lock_guard<std::mutex> locker(m_mutex);
return m_queue.size() == m_maxSize;
}
//隊(duì)列大小
size_t Size()
{
std::lock_guard<std::mutex> locker(m_mutex);
return m_queue.size();
}
private:
//往隊(duì)列里添加事件,事件是范型的,,c++11我們可以把函數(shù)通過(guò)std::function封裝為對(duì)象,。
template<typename F>
void Add(F &&x)
{
std::unique_lock<std::mutex> locker(m_mutex);
m_notFull.wait(locker, [this] {
return m_needStop || NotFull() ; });
if(m_needStop)
return;
m_queue.push_back(std::forward<F>(x));
m_notEmpty.notify_one();
}
//隊(duì)列未滿
bool NotFull() const
{
bool full = m_queue.size() >= m_maxSize;
if(full)
std::cout << "緩沖區(qū)滿了...請(qǐng)等待" << std::endl;
return !full;
}
//隊(duì)列不為空
bool NotEmpty() const
{
bool empty = m_queue.empty();
if(empty)
{
std::cout << "緩沖區(qū)空了...請(qǐng)等待" << std::endl;
std::cout << "線程ID:" << std::this_thread::get_id() << std::endl;
}
return !empty;
}
private:
std::mutex m_mutex; //互斥鎖
std::list<T> m_queue; //隊(duì)列,存放任務(wù)
std::condition_variable m_notEmpty; //隊(duì)列不為空的條件變量
std::condition_variable m_notFull; //隊(duì)列不為滿的條件變量
int m_maxSize; //任務(wù)隊(duì)列最大長(zhǎng)度
bool m_needStop; //終止標(biāo)識(shí)
};
線程池:
#include <stdio.h>
#include <stdlib.h>
#include <assert.h>
#include <unistd.h>
#include "SynQueue.h"
#include <functional>
#include <thread>
#include <memory>
#include <atomic>
const int MaxTaskCount = 100;
class ThreadPool
{
public:
//規(guī)定任務(wù)類型為void(),,我們可以通過(guò)c++11 不定參數(shù)模板來(lái)實(shí)現(xiàn)一個(gè)可接受任何函數(shù)的范型函數(shù)模板,,這樣就是一個(gè)可以接受任何任務(wù)的任務(wù)隊(duì)列了。
using Task = std::function<void()>;
//hardware_concurrency檢測(cè)硬件性能,,給出默認(rèn)線程數(shù)
ThreadPool(int numThreads = std::thread::hardware_concurrency()):
m_queue(MaxTaskCount)
{
//初始化線程,并通過(guò)shared_ptr來(lái)管理
Start(numThreads);
}
//銷毀線程池
~ThreadPool(void)
{
Stop();
}
//終止所有線程,call_once保證函數(shù)只調(diào)用一次
void Stop()
{
std::call_once(m_flag, [this] { StopThreadGroup(); });
}
//添加任務(wù),,普通版本和右值引用版本
void AddTask(const Task& task)
{
m_queue.Put(task);
}
void AddTask(Task && task)
{
m_queue.Put(std::forward<Task>(task));
}
private:
//停止線程池
void StopThreadGroup()
{
m_queue.Stop();
m_running = false;
for(auto thread : m_threadgroup)
{
if(thread)
thread->join();
}
m_threadgroup.clear();
}
void Start(int numThreads)
{
m_running = true;
for(int i = 0; i < numThreads; ++i)
{
//智能指針管理,并給出構(gòu)建線程的參數(shù),,線程調(diào)用函數(shù)和參數(shù)
std::cout << "Init create thread pool" << std::endl;
m_threadgroup.push_back(std::make_shared<std::thread>(&ThreadPool::RunInThread, this));
}
}
//一次取出隊(duì)列中全部事件
void RunInThread_list()
{
while(m_running)
{
std::list<Task> list;
std::cout << "take " << std::endl;
m_queue.Take(list);
for(auto &task : list)
{
if(!m_running)
return;
task();
}
}
}
//一次只取一個(gè)事件
void RunInThread()
{
std::cout << m_queue.Size() << std::endl;
while(m_running)
{
Task task;
if(!m_running)
return;
m_queue.Take(task);
}
}
private:
//線程池
std::list<std::shared_ptr<std::thread>> m_threadgroup;
//任務(wù)隊(duì)列
SynQueue<Task>m_queue;
//原子布爾值
std::atomic_bool m_running;
//輔助變量->call_once
std::once_flag m_flag;
};
int main(int argc, char *argv[])
{
ThreadPool pool(2);
//創(chuàng)建線程向任務(wù)隊(duì)列添加任務(wù)
std::thread thd1([&pool]{
for(int i = 0; i < 10; i++)
{
auto thdId = std::this_thread::get_id();
pool.AddTask([thdId](){
std::cout << thdId << " thread execute task" << std::endl;
});
}
});
std::this_thread::sleep_for(std::chrono::seconds(2));
pool.Stop();
thd1.join();
return EXIT_SUCCESS;
}
參考書籍:
深入應(yīng)用c++11
完
|