C++并发与多线程(五)互斥量，atomic、与线程池

文章目录

互斥量
std::atomic
std::atomic续谈
std::async深入理解
atomic的原子操作
std::async和std::thread()区别：
浅谈线程池

本文系列大部分来自c++11并发与多线程视频课程的学习笔记，系列文章有(不定期更新维护)：

C++并发与多线程(一)线程传参
C++并发与多线程(二) 创建多个线程、数据共享问题分析、案例代码
C++并发与多线程(三)单例设计模式与共享数据分析、call_once、condition_variable使用
C++并发与多线程(四)async、future、packaged_task、promise、shared_future
C++并发与多线程(五)互斥量，atomic、与线程池

互斥量

互斥量：多线程编程中用于保护共享数据：先锁住，操作共享数据，解锁。有两个线程，对一个变量进行操作，一个线程读这个变量的值，一个线程往这个变量中写值。即使是一个简单变量的读取和写入操作，如果不加锁，也有可能会导致读写值混乱（一条C语句会被拆成3、4条汇编语句来执行，所以仍然有可能混乱）。

#include <iostream>
#include <thread>
using namespace std;
int g_count = 0;void mythread1() {for (int i = 0; i < 1000000; i++) {g_count++;}
}int main() {std::thread t1(mythread1);std::thread t2(mythread1);t1.join();t2.join();cout << "正常情况下结果应该是200 0000次，实际是" << g_count << endl;
}

出现上述情况的问题在于g_count++;这步操作被打断了。我们可以使用mutex解决这个问题：

#include <iostream>
#include <thread>
#include <mutex>
using namespace std;
int g_count = 0;
std::mutex mymutex;void mythread1() {for (int i = 0; i < 1000000; i++) {std::unique_lock<std::mutex> u1(mymutex);g_count++;}
}int main() {std::thread t1(mythread1);std::thread t2(mythread1);t1.join();t2.join();cout << "正常情况下结果应该是200 0000次，实际是" << g_count << endl;
}

std::atomic

大家可以把原子操作理解成一种：不需要用到互斥量加锁（无锁）技术的多线程并发编程方式。原子操作：在多线程中不会被打断的程序执行片段。从效率上来说，原子操作要比互斥量的方式效率要高。互斥量的加锁一般是针对一个代码段，而原子操作针对的一般都是一个变量。原子操作，一般都是指“不可分割的操作”；也就是说这种操作状态要么是完成的，要么是没完成的，不可能出现半完成状态。std::atomic来代表原子操作，是个类模板。其实std::atomic是用来封装某个类型的值的。需要添加#include头文件。

#include <iostream>
#include <thread>
#include <atomic>
using namespace std;
std::atomic<int> g_count = 0; //封装了一个类型为int的 对象（值）void mythread1() {for (int i = 0; i < 1000000; i++) {g_count++;}
}int main() {std::thread t1(mythread1);std::thread t2(mythread1);t1.join();t2.join();cout << "正常情况下结果应该是200 0000次，实际是" << g_count << endl;
}

输出结果为：

#include <iostream>
#include <thread>
#include <atomic>
using namespace std;
std::atomic<bool> g_ifEnd = false; //封装了一个类型为bool的 对象（值）void mythread() {std::chrono::milliseconds dura(1000);while (g_ifEnd == false) {cout << "thread id = " << std::this_thread::get_id() << "运行中" << endl;std::this_thread::sleep_for(dura);}cout << "thread id = " << std::this_thread::get_id() << "运行结束" << endl;
}int main() {std::thread t1(mythread);std::thread t2(mythread);std::chrono::milliseconds dura(5000);std::this_thread::sleep_for(dura);g_ifEnd = true;cout << "程序执行完毕" << endl;t1.join();t2.join();
}

程序输出结果为：

原子操作一般用于计数或者统计（如累计发送多少个数据包，累计接收到了多少个数据包），多个线程一起统计，这种情况如果不使用原子操作会导致统计发生混乱。

std::atomic续谈

#include <iostream>
#include <mutex>
#include <thread>using namespace std;
std::atomic<int> g_count;void mythread1() {for (int i = 0; i < 1000000; i++) {//g_count += 1; // 正确g_count = g_count + 1; // 错误：虽然g_count使用了原子操作模板，但是这种写法既读又写，}
}
int main() {std::thread t1(mythread1);std::thread t2(mythread1);t1.join();t2.join();cout << "正常情况下结果应该是200 0000次，实际是" << g_count << endl;
}

一般atomic原子操作，针对++，–，+=，-=，&=，|=，^=是支持的，其他操作不一定支持。

std::async深入理解

std::async参数详述，async用来创建一个异步任务。延迟调用参数 std::launch::deferred【延迟调用】，std::launch::async【强制创建一个线程】。std::async()我们一般不叫创建线程（他能够创建线程），我们一般叫它创建一个异步任务。

#include <iostream>
#include <mutex>
#include <thread>
#include <future>using namespace std;
std::atomic<int> g_count;int mythread() {cout << "thread start thread id is: " << std::this_thread::get_id() << endl;return 10;
}
int main() {cout << "main start thread id is: " << std::this_thread::get_id() << endl;std::future<int> res = std::async(mythread);cout << "res.get() is : " << res.get() << endl;
}

程序输出结果为：

main start thread id is: 0x1000e3d40
res.get() is : thread start thread id is: 0x16fe87000
10

std::async和std::thread最明显的不同，就是async有时候并不创建新线程。

如果用std::launch::deferred来调用async？

#include <iostream>
#include <mutex>
#include <thread>
#include <future>using namespace std;
std::atomic<int> g_count;int mythread() {cout << "thread start thread id is: " << std::this_thread::get_id() << endl;return 10;
}
int main() {cout << "main start thread id is: " << std::this_thread::get_id() << endl;std::future<int> res = std::async(std::launch::deferred,mythread);cout << "res.get() is : " << res.get() << endl;
}

程序输出结果为：

main start thread id is: 0x1000e3d40
res.get() is : thread start thread id is: 0x1000e3d40
10

延迟到调用get()或者wait()时执行，如果不调用就不会执行。并且没有创建新线程。

如果用std::launch::async来调用async？

#include <iostream>
#include <mutex>
#include <thread>
#include <future>using namespace std;
std::atomic<int> g_count;int mythread() {cout << "thread start thread id is: " << std::this_thread::get_id() << endl;return 10;
}
int main() {cout << "main start thread id is: " << std::this_thread::get_id() << endl;std::future<int> res = std::async(std::launch::async,mythread);cout << "res.get() is : " << res.get() << endl;
}

程序输出结果为：

main start thread id is: 0x1000e3d40
res.get() is : thread start thread id is: 0x16fe87000
10

强制这个异步任务在新线程上执行，这意味着，系统必须要创建出新线程来运行入口函数。

如果同时用std::launch::async | std::launch::deferred

#include <iostream>
#include <mutex>
#include <thread>
#include <future>using namespace std;
std::atomic<int> g_count;int mythread() {cout << "thread start thread id is: " << std::this_thread::get_id() << endl;return 10;
}
int main() {cout << "main start thread id is: " << std::this_thread::get_id() << endl;std::future<int> res = std::async(std::launch::async | std::launch::deferred,mythread);cout << "res.get() is : " << res.get() << endl;
}

这里这个或者关系意味着async的行为可能是std::launch::async创建新线程立即执行，也可能是 std::launch::deferred没有创建新线程并且延迟到调用get()执行，由系统根据实际情况来决定采取哪种方案。

不带额外参数std::async(mythread)，只给async一个入口函数名，此时的系统给的默认值是 std::launch::async | std::launch::deferred和3.一样，有系统自行决定异步还是同步运行。

async的这种不确定性问题，不确定是否会创建出新线程的话，如何来解决呢。不加额外参数的async调用时让系统自行决定，是否创建新线程。

std::future result = std::async(mythread);

这个问题焦点在于，上述这种写法，任务到底有没有被推迟执行。我们可以通过wait_for返回状态来判断：

#include <iostream>
#include <mutex>
#include <thread>
#include <future>using namespace std;
std::atomic<int> g_count;int mythread() {cout << "thread start thread id is: " << std::this_thread::get_id() << endl;return 10;
}
int main() {cout << "main start thread id is: " << std::this_thread::get_id() << endl;std::future<int> res = std::async(mythread);std::future_status status = res.wait_for(std::chrono::seconds(0s));//std::future_status status = result.wait_for(6s);if (status == std::future_status::timeout) {//超时：表示线程还没有执行完cout << "超时了，线程还没有执行完" << endl;}else if (status == std::future_status::ready) {//表示线程成功放回cout << "线程执行成功，返回" << endl;cout << res.get() << endl;}else if (status == std::future_status::deferred) {cout << "线程延迟执行" << endl;cout << res.get() << endl; // 这个时候才去调用了mythread}
}

程序输出结果为：

main start thread id is: 0x1000e7d40
超时了，线程还没有执行完
thread start thread id is: 0x16fe87000
Program ended with exit code: 0

atomic的原子操作

来看如下代码：

#include <iostream>
#include <thread>
#include <list>
#include <mutex>
using namespace std;class A {public:A(){atm = 0;}void inMsgRecvQueue(){for(int i = 0; i < 10; ++i){++atm;}}void outMsgRecvQueue(){while(true){cout << "automic is: " << atm << endl;}}
private:std::atomic<int> atm;list<int> msgRecvQueue;mutex myMutex;std::condition_variable cond;
};int main()
{A myobja;thread myOutMsgObj(&A::outMsgRecvQueue, &myobja);thread myInMsgObj(&A::inMsgRecvQueue, &myobja);myOutMsgObj.join();myInMsgObj.join();return 0;
}

上述代码中的cout << atm << endl;并不是一个原子操作。因为只有读取atm是原子操作，但是cout输出的时候，有可能atm的值已经被改变掉了，导致最终显示在屏幕上的值是一个“曾经值”。

如果在拷贝构造函数中，调用赋值语句的话，我们可以得到如下代码：

std::atomic<int> atm = 0;
auto atm2 = atm; //不可以

但是上述这种代码用来初始化是不可以的，会报错。但是可以通过load()函数来以原子方式读atomic对象的值。

atomic<int> atm2(atm.load());

store()以原子方式写入内容：

atm2.store(12);

原子操作实质上是：不允许在进行原子对象操作时进行CPU的上下文切换。

std::async和std::thread()区别：

std::thread()如果系统资源紧张可能出现创建线程失败的情况，如果创建线程失败那么程序就可能崩溃，并且不容易拿到函数返回值（不是拿不到，通过设置全局变量可以拿到）。std::async()创建异步任务。可能创建线程也可能不创建线程，并且容易拿到线程入口函数的返回值。由于系统资源限制：

如果用std::thread创建的线程太多，则可能创建失败，系统报告异常，崩溃。
如果用std::async，一般就不会报异常。因为如果系统资源紧张，无法创建新线程的时候，async不加额外参数的调用方式就不会创建新线程。而是在后续调用get()请求结果时执行在这个调用get()的线程上。如果你强制async一定要创建新线程就要使用std::launch::async标记。承受的代价是，系统资源紧张时可能崩溃。
根据经验，一个程序中线程数量不宜超过100~200。

浅谈线程池

假设我们有如下场景，场景设想：服务器程序，每来一个客户端，就创建一个新线程为这个客户提供服务。我们需要考虑如下问题：

2万个玩家，不可能给每个玩家创建一个新线程，此程序写法在这种场景下不通。
程序稳定性问题：编写代码中，“时不时地突然”创建一个线程，这种写法，一般情况下不会出错，但是不稳定的；

线程池：把一堆线程弄到一起，统一管理。这种统一管理调度，循环利用的方式，就叫做线程池。用的时候抓一个过来用，用完了之后把它放回线程池中去，也不释放。

实现方式：程序启动时，一次性创建好一定数量的线程。这种方式让人更放心，觉得程序代码更稳定。

线程创建数量谈

线程创建的数量极限的问题：一般来讲，2000个线程基本就是极限；再创建就会崩溃。
线程创建数量建议：采用某些计数开发程序提供的建议，遵照建议和指示来确保程序高效执行。
创建多线程完成业务；考虑可能被阻塞的线程数量，创建多余最大被阻塞线程数量的线程，如100个线程被阻塞再充值业务，开110个线程就是很合适的。
线程创建数量尽量不要超过500个，尽量控制在200个之内；