博客
关于我
强烈建议你试试无所不能的chatGPT,快点击我
c++11 实现半同步半异步线程池
阅读量:6083 次
发布时间:2019-06-20

本文共 4834 字,大约阅读时间需要 16 分钟。

感受

随着深入学习,现代c++给我带来越来越多的惊喜…

c++真的变强大了。

半同步半异步线程池:

事实上非常好理解。分为三层

同步层:通过IO复用或者其它多线程多进程等不断的将待处理事件加入到队列中。这个过程是同步进行的。
队列层:全部待处理事件都会放到这里。

上一层事件放到这里。下一层从这里获取事件

异步层:事先创建好线程,让线程不断的去处理队列层的任务,上层不关心这些。它仅仅负责把任务放到队列里,所以对上层来说这里是异步的。


补充下思路:

主要是后两层

队列层:c++11 通过std::function能够将函数封装为对象。那么我们一个函数也就是一个任务。通过vector或list等容器来存储这些”任务”来供后面存取。由于会出现竞争资源的问题,所以我们要加锁,而且通过条件变量的条件来唤醒其它堵塞在锁上的线程。当然你想避免线程堵塞浪费资源能够用带时间的锁std::time_mutex。

异步层:c++11 将线程也封装为了对象,那么我们创建一个容器保存线程对象,让他们去队列层取任务并运行,运行完并不结束该线程而是归还给容器(线程池)。


看张图:

这里写图片描写叙述


假设你不熟悉c++11的内容

下面文章仅供參考

代码

同步队列:

#include 
#include
#include
#include
#include
template
class SynQueue{ public: SynQueue(int maxsize): m_maxSize(maxsize), m_needStop(false) { } //加入事件,左值拷贝和右值移动 void Put(const T&x) { //调用private内部接口Add Add(x); } void Put(T &&x) { Add(x); } //从队列中取事件,取全部事件 void Take(std::list
&list) { //有wait方法必须用unique_lock //unique_lock有定时等待等功能,lock_guard就仅仅是RAII手法的相互排斥锁 //但unique_lock的性能稍低于lock_guard 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(); t(); } //停止全部线程在同步队列中的读取 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(); } private: //往队列里加入事件,事件是范型的。c++11我们能够把函数通过std::function封装为对象。 template
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(); } //队列未满 bool NotFull() const { bool full = m_queue.size() >= m_maxSize; if(full) std::cout << "缓冲区满了...请等待" << std::endl; return !full; } //队列不为空 bool NotEmpty() const { bool empty = m_queue.empty(); if(empty) { std::cout << "缓冲区空了...请等待" << std::endl; std::cout << "线程ID:" << std::this_thread::get_id() << std::endl; } return !empty; } private: std::mutex m_mutex; //相互排斥锁 std::list
m_queue; //队列,存放任务 std::condition_variable m_notEmpty; //队列不为空的条件变量 std::condition_variable m_notFull; //队列不为满的条件变量 int m_maxSize; //任务队列最大长度 bool m_needStop; //终止标识};

线程池:

#include 
#include
#include
#include
#include "SynQueue.h"#include
#include
#include
#include
const int MaxTaskCount = 100;class ThreadPool{ public: //规定任务类型为void(),我们能够通过c++11 不定參数模板来实现一个可接受不论什么函数的范型函数模板,这样就是一个能够接受不论什么任务的任务队列了。 using Task = std::function
; //hardware_concurrency检測硬件性能。给出默认线程数 ThreadPool(int numThreads = std::thread::hardware_concurrency()): m_queue(MaxTaskCount) { //初始化线程,并通过shared_ptr来管理 Start(numThreads); } //销毁线程池 ~ThreadPool(void) { Stop(); } //终止全部线程,call_once保证函数仅仅调用一次 void Stop() { std::call_once(m_flag, [this] { StopThreadGroup(); }); } //加入任务。普通版本号和右值引用版本号 void AddTask(const Task& task) { m_queue.Put(task); } void AddTask(Task && task) { m_queue.Put(std::forward
(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) { //智能指针管理,并给出构建线程的參数,线程调用函数和參数 std::cout << "Init create thread pool" << std::endl; m_threadgroup.push_back(std::make_shared
(&ThreadPool::RunInThread, this)); } } //一次取出队列中全部事件 void RunInThread_list() { while(m_running) { std::list
list; std::cout << "take " << std::endl; m_queue.Take(list); for(auto &task : list) { if(!m_running) return; task(); } } } //一次仅仅取一个事件 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
> m_threadgroup; //任务队列 SynQueue
m_queue; //原子布尔值 std::atomic_bool m_running; //辅助变量->call_once std::once_flag m_flag;};int main(int argc, char *argv[]){ ThreadPool pool(2); //创建线程向任务队列加入任务 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;}

这里写图片描写叙述

參考书籍:

深入应用c++11

你可能感兴趣的文章
扩展器必须,SAS 2.0未必(SAS挺进中端存储系统之三)
查看>>
Eclipse遇到Initializing Java Tooling解决办法
查看>>
while((ch = getchar()) != '\n')
查看>>
好程序员web前端分享JS检查浏览器类型和版本
查看>>
Oracle DG 逻辑Standby数据同步性能优化
查看>>
exchange 2010 队列删除
查看>>
「翻译」逐步替换Sass
查看>>
H5实现全屏与F11全屏
查看>>
处理excel表的列
查看>>
C#数据采集类
查看>>
quicksort
查看>>
【BZOJ2019】nim
查看>>
四部曲
查看>>
LINUX内核调试过程
查看>>
【HDOJ】3553 Just a String
查看>>
Java 集合深入理解(7):ArrayList
查看>>
2019年春季学期第四周作业
查看>>
linux环境配置
查看>>
ASP.NET MVC中从前台页面视图(View)传递数据到后台控制器(Controller)方式
查看>>
一个想法(续二):换个角度思考如何解决IT企业招聘难的问题!
查看>>