最近在为一个小东西作点铺垫，学了点并发编程，搞了个线程池出来，学到了点小东西，浅记录一下
（其实完全是抄的(15 封私信 / 17 条消息) 基于C++11实现线程池 - 知乎
1.并发知识
c++绝大部分并发编程内容都是c++11后出现，包括thread，mutex等等工具，有了这些类之后，最好的就是代码对系统兼容性宽泛许多，不用去查windows的LPTHREAD和linux pthread等等用法
(1)std::thread
最典型的线程类，为相应函数创造一个线程类用以运行该函数，并对其进行操作
(2)std::mutex
互斥锁，用以控制资源访问安全，避免多线程编程中多个线程访问或改变同一资源带来的混乱
当然互斥锁有很多种，比如递归互斥锁等等
日常主要使用两种RAII的自动管理锁工具std::unique_mutex和std::lock_guard
其中unique_mutex可手动解锁，lock_guard不可手动解锁
为了只在一定范围上锁，一般会用{}来规定其作用范围
(3)std::function
function是一个函数类，他能存储封装一类函数。譬如std::function<int(int,bool)>能够存储int fun(int a,bool b)形式的函数，且可随时调用。本来我以为用function加入线程池便可解决一切，后来发现function只能取一类特定的函数，便改用可变参数模板
(4)std::future,std::promise,std::packaged_task
packaged_task很类似于function(其实并不怎么类似)，他能封装一个异步任务，并可存储返回值
而promise可用于线程间传递值，包括传递函数，对象等等等等
future则用于从异步任务中获取所共享或返回的值,通过promise或者packagedtask便可在线程间传递数据。
(5)std::condition_variable
条件变量，用于线程的挂起和唤醒。当线程闲置时，可释放互斥锁并挂起线程，直到满足条件，唤醒线程执行任务，此时mutex重新上锁。条件变量使用时与互斥锁绑定，来控制资源的高效访问和有序修正
2.线程池原理与实现
(1)原理
作为一个池化结构，线程池同样是一次性申请一定数量的线程存储起来，等到有需要的时候再执行任务。
提交任务时，将任务封装后调用空闲线程执行
听着很简单吧
确实如此，线程池的码量远远低于内存池，但细节细化上比较复杂，尤其是勾槽的c++11乱七八糟的语法
不过仍然一步一步来
(2)实现
A.安全队列，MulityQueue类
名字随便起的，无所谓
出于多线程访问的需要，我们需要一个有互斥锁来维护队列资源，只需要在每个操作前给STL基本队列上锁即可
template <typename T>
class MulityQueue{
public:
MulityQueue()=default;
~MulityQueue()=default;
bool pop(){
std::unique_lock<std::mutex> locker(mtx);
if(q.empty())return 0;
q.pop();
return 1;
}
void add(const T& pusher){
std::unique_lock<std::mutex> locker(mtx);
q.emplace(pusher);
}
const T& top(){
std::unique_lock<std::mutex> locker(mtx);
return q.front();
}
bool empty(){
std::unique_lock<std::mutex> locker(mtx);
return q.empty();
}
int size(){
std::unique_lock<std::mutex> locker(mtx);
return q.size();
}
private:
std::mutex mtx;
std::queue<T> q;
};
由于unique_lock的RAII特性，直接在函数体内声明即上锁
B.线程池，ThreadPool类
首先考虑我们需要什么
最基本的，线程数组存储相应的线程，以及待执行的任务队列
接着是互斥量，锁定临界区资源；条件变量，挂起无用线程以及唤醒线程使用，以及监测运行状态的bool变量
除此之外，我们还需要线程执行的主函数，以持续线程的运行
而对外的接口，除了构造和析构，需要一个停止函数和最重要的任务提交函数
到此一个简单的线程池结构就完成了
class ThreadPool{
public:
ThreadPool(int size=10);
~ThreadPool(){stop();};
template<typename T,typename... Args>
auto submit(T &&f,Args &&...args)
->std::future<std::invoke_result_t<T,Args...>>;
void stop();
private:
std::mutex mtx;
bool is_running;
MulityQueue<std::function<void()> > Tasks;
std::vector<std::thread> Threads;
void KeepGetting();
std::condition_variable conv;
};
接下来一个个分析
void KeepGetting()
首先为了实现线程待机，不能让线程线性执行后直接return，否则线程直接关闭
因此我们在线程中持续循环while(1)尝试向任务队列获取任务执行
void ThreadPool::KeepGetting(){
while(1){
std::function<void()> f;
{
std::unique_lock<std::mutex> lock(mtx);
conv.wait(lock,[this]{return !is_running||!Tasks.empty();});
if(!is_running&&Tasks.empty())return;
f=Tasks.top();
Tasks.pop();
}
f();
}
}
好的，这里涉及一些唐诗语法
首先这里多出来的大括号是用来限定互斥锁的作用域，限定锁在这个域内生效，域外不生效。不用考虑为什么mutex不上锁，unique_lock会帮你安排一切的，除非真的有特别需求。
接着是条件变量的wait成员
template< class Predicate > void wait( std::unique_lock<std::mutex>& lock, Predicate pred );
函数第一个参数为互斥锁，后一个是谓词成分
这个函数可以阻塞线程，直到条件变量唤醒（notify_one或notify_all）,再继续运行；而且唤醒后若谓词成分(pred)仍为false，则继续阻塞线程
再想一下我们需要做什么。我们当然可以while(1)检测任务队列，若队列非空则取出队首执行，互斥锁可以保证操作的安全
但是这样的做法未免太过繁琐，请求过于频繁，降低效率
于是我们引入条件变量来控制线程的阻塞，并且检测线程池是否运行以及任务队列是否空作为谓词
（当然谓词部分也可以用while(!pred)wait(lock);来代替，效果相同
但是这里留个小悬念，为什么f一定是void()类型的呢？提交上去的函数不一定都是这个类型
但如果类型不同，如何放入队列调用呢？
所以为了统一，我们会在接下来的submit函数中将任务封装成一个void()函数，且能顺利调用
template<typename T,typename... Args>auto submit(T &&f,Args &&...args)->std::future<std::invoke_result_t<T,Args...>>;
这个是不是看得怪怪的？
先从声明说起
1.现在我们想要传入函数以及函数的参数，然而函数本身的类型以及有多少参数并不是能够确定的，因此我们使用可变参数模板来解决这个问题，c++11中的typename...可以规定一个参数数量可变的类模板，std::thread便用了这种技术传参
2.submit的参数:T &&f与Args &&...args
我们知道，&表示引用，那么&&是什么呢？我们好像只在条件语句中看到过这个“与”运算符。
这里&&用法表示右值引用，传统的&则是左值引用。右值引用能够避免额外的空间开销与可能导致的数据改变
3.auto func()->(type)
c++11的特性尾返回类型推导，将返回类型后置。
平时的函数声明是将函数类型直接声明，然而我们并不知道接受的函数返回类型会是多少(模板中的T类型此时是函数指针)，利用这个能将函数返回类型后置。
4.std::invoke_result_t<T,Args...>
std::future的用法前面已经陈述完毕，即能够获取异步任务中返回的数据，但是仍然是前面的问题，即返回类型怎么获得？
先介绍一下c++11中有一个很方便的工具:decltype。这个关键字与auto很相似，都能够进行编译时类型推导，不同的是，decltype支持推导表达式的值，即decltype(表达式)，而且，更加不同的一点是，decltype推导出的能够作为类型声明，而不是推到初始化表达式。下面是一个简单的用例差别，二者差别在于k的初始值
auto k=4; <->decltype(4) k
所以事实上我们可以写decltype(T(args...))，但是会有一些小问题
AI给出的说法
简单来说，std::invoke_result_t返回值更精确一些，且能够避免函数签名，函数指针引用与成员函数等问题
于是我们使用c++17的std::invoke_result_t来推导返回类型（貌似还有一个工具std::declval可用，可以了解一下）。该类型是由c++11的result_t(由于安全问题，c++14被废止，c++17彻底禁用改为invoke_result_t)
至此，声明部分结束，我们做到能返回正确值以及正确传参，下面是实现部分
template<typename T,typename... Args>
auto ThreadPool::submit(T &&f,Args &&...args)
->std::future<std::invoke_result_t<T,Args...>>{
std::function<std::invoke_result_t<T,Args...>()> F=std::bind(std::forward<T>(f),std::forward<Args>(args)...);
auto task_=std::make_shared<std::packaged_task<std::invoke_result_t<T,Args...>()>>(F);
std::function<void()> wrapped_task=
[task_]{
(*task_)();
};
{
std::unique_lock<std::mutex> lk(mtx);
if(!is_running)throw std::runtime_error("ThreadPool Has Stopped");
Tasks.add(wrapped_task);
}conv.notify_one();
return task_->get_future();
}
该说不说，这些语法实在是太复杂了……一点点说吧
整体思路：一点点封装任务函数为统一结构安全放入队列，呼叫thread运行
1.std::function<std::invoke_result_t<T,Args...>()> F=std::bind(std::forward<T>(f),std::forward<Args>(args)...);
好的，invoke_result_t我们已经说明过了，function导引时也说过了，问题是这个类型是个无参函数，可是f的参数不固定，怎么做到的？
std::bind
简而言之，std::bind能把一个函数的参数预先绑定好，这样函数运行时只需要剩下的参数就可以，可以理解为预设值，bind能返回一个新函数，除了参数数量以外功能与原函数一致
有点抽象，举个例子吧
int func(int a,int b){...}
int main(){
int c=1,d;
std::cin>>d;
std::function<int(int)> newfun=std::bind(func,c);
std::cout<<newfun(d);
}
如上，将func的第一个参数与一个变量绑定，新函数就只需要一个传入参数，这种做法对泛用性强的函数在专用区域的使用很方便
回过头来，我们已经将f利用std::bind封装为一个零参函数
auto task_=std::make_shared<std::packaged_task<std::invoke_result_t<T,Args...>()>>(F);
接下来，用智能指针封装任务
如何封装？我们明显感觉到包括下面这一步进一步的封装是没什么作用的，纯粹是叠洋葱层层封包
其实我们的想法是对的，确实没什么用，单纯增加代码量
但是还记得我们的返回值吗？虽然能自动推导类型，但我们返回的仍是std::future,也因此不可避免地需要std::packaged_task这一异步任务类型来通过get_future()返回
但是，std::packaged_task不是一个可执行的函数，因此我们还需要封装在函数体里，于是用std::make_shared创建指向已封装的函数F的函数指针
当然，为什么需要这个指针呢？还记得我们刚开始说的吗，要将函数类型统一为void()类型，才能方便调用和存储，于是有
std::function<void()> wrapped_task=
[task_]{
(*task_)();
};
lamdba表达式，即匿名函数，简单地创造一个void()无返回值函数，通过[task_]捕捉函数指针在lamdba函数体内执行，这样就能封装为function<void()>类型
到此我们封装完毕，接下来便是插入任务队列，唤醒线程执行
{
std::unique_lock<std::mutex> lk(mtx);
if(!is_running)throw std::runtime_error("ThreadPool Has Stopped");
Tasks.add(wrapped_task);
}conv.notify_one();
return task_->get_future();
那个看起来莫名其妙的大括号是用来限定互斥锁范围，执行后及时释放方便线程使用
此时封装后的wrapped_task便能顺利地插入任务队列，插入队列后notify_one唤醒一个线程来执行任务，最后用get_future()获取返回值
至此，大功告成！
4.总结与使用
使用的话，既然说完了，调用接口就行了
算是慢慢磨（抄）过来的一个小东西吧，我把它用到的webserver上提升性能
该说不说c++一些新标准都挺好用的，可惜越来越没人搞了
前次的memorypool，我没敢用，安全性问题我没把握
好了，完结撒花！
Comments | NOTHING