c++并发
1. thread
只能join()/detach()一次
#include<thread>
void some_func(int param) {
}
int main () {
std::thread thread{somfunc, 10};
// std::join(); 主线程（main）等待子线程执行完再继续执行
// std::detach(); 主线程和子线程彻底分离， 子线程交给c++ runtime去回收资源
}
2. this_thread命名空间
#include<thread>
std::this_thread::get_id();
std::this_thread::sleep_for(std::chrono::seconds(1));
auto time_point = std::chrono::steady_clock::now() + std::chrono::seconds(3);
std::this_thread::sleep_until(time_point);
3. 互斥
1. mutex
#include<mutex>
#include<iostream>
std::mutex mtx;
void out_count(int count) {
mtx.lock();
std::cout << count << std::endl;
mtx.unglock();
}
2. 符合RAII标准的锁： lock_guard
其实就是锁， 但是为了符合RAII标准，用lock_guard把std::mutex转换成符合RAII标准的锁
lock_guard在构造时自动上锁
析构时自动解锁， 所以在退出作用域时自动解锁
换句话说， std::lock_guard就是在声明的时候上锁， 退出作用域自动解锁, 完全可以理解为std::mutex
#include<mutex>
std::mutex mtx;
// func1 == func2 !
void func1() {
std::lock_guard lock(mtx);
// do_sth
}
void func2() {
mtx.lock();
// do_sth
mtx.unlock();
}
// 自己限定作用域
#include<mutex>
std::mutex mtx;
void func() {
// area
{
// 锁的作用范围
std::lock_guard lock(mtx);
}
// 不锁了
}
3. 符合RAII标准并且更自由：unique_lock
lock_guard只能在退出作用域的时候解锁， 不能自己解锁， 就很沙比
unique_lock就可以理解为可以手动解锁的lock_guard
std::mutex mtx;
void func() {
std::unique_lock ulock(mtx);
// do_sth ...
ulock.unlock();
// 还可以重新上锁
ulock.lock();
}
// 最后退出作用域不管之前怎么解锁上锁， 最后还是会检查一遍要不要解锁
4. 死锁
1. 死锁的预防： 破坏请求和保持条件：一次性上多个锁
1. std::lock()
std::mutex mtx1, mtx2;
void func() {
std::lock(mtx1, mtx2); // 同时上两个锁
mtx1.unlock();
mtx2.unlock();
// 可以分别解锁
}
2. RAII版本： std::scoped_lock()
不解释， 还是退出作用域自动解锁
2. 单个线程也会死锁——使用递归锁避免： std::recursive_mutex
死锁举例
void other() {
mtx.lock();
// do_sth
mtx.unlock();
}
std::mutex mtx;
void func() {
mtx.lock();
other();
mtx.unlock();
}
在一个线程里锁两次，就死锁了， 把这里的锁替换成recursive_mutex， 这玩意里面自带计数器， 就不会死锁了
5. 读写锁： shared_mutex
特点： 读共享， 写互斥（操作系统的读写模型）
能允许的事情：
- n个人读， 没人写
- 一个人写， 没人读
- 没人读没人写
接口：
std::shared_mutex smtx;
void func() {
// 互斥
smtx.lock();
smtx.unlock();
// 共享锁
smtx.lock_shared();
smtx.unlock_shared();
}
符合RAII：
std::shared_lock 根之前一样， 不多说
4. 硬件支持最大线程数
unsigned int max_thread_num = std::thread::hardware_concurrency();
lscpu # 看看和输出是不是一样滴
5. cmake中引入pthread
set(THREADS_PREFER_PTHREAD_FLAG ON)
find_package(Threads REQUIRED)
target_link_libraries(test PUBLIC Threads::Threads)
6. 同步： std::condition_variable
类似信号量， 根据条件阻塞和解锁
举例
#include<mutex>
#include<condition_variable>
std::mutex mtx;
std::condition_variable cv;
int i = 0;
void fun() {
std::unique_lock ulock(mtx);
while(i > 0) {
// 当前线程进入等待
cv.wait(ulock);
}
}
void fun2() {
// 唤醒陷入等待的线程
cv.notify_one();
}
增加条件
cv.wait(lock, condition)
condition是一个lambda表达式， 当返回值为真才能唤醒
std::mutex mtx;
std::condition_variable cv;
bool ready = false;
void fun() {
std::unique_lock ulock(mtx);
cv.wait(ulock, [&] { return ready; });
}
void fun2() {
cv.notify_one(); // 不能唤醒fun()
ready = true;
cv.notify_one(); // fun被唤醒
cv.notify_all(); // 全部唤醒（如果有多个线程阻塞在cv）
}
notif
7. 线程池
并发中一般把任务作为一个类， 在任务中实现线程安全
把线程也实现一个类， 用容器装线程有关的信息
然后用线程池去创建线程， 任务对象方法线程的入口函数
#ifndef _THREADPOOL_H
#define _THREADPOOL_H
#include <vector>
#include <queue>
#include <thread>
#include <iostream>
#include <stdexcept>
#include <condition_variable>
#include <memory> //unique_ptr
#include<assert.h>
const int MAX_THREADS = 1000; //最大线程数目
template <typename T>
class threadPool
{
public:
threadPool(int number = 1);//默认开一个线程
~threadPool();
std::queue<T *> tasks_queue; //任务队列
bool append(T *request);//往请求队列＜task_queue＞中添加任务<T *>
private:
//工作线程需要运行的函数,不断的从任务队列中取出并执行
static void *worker(void *arg);
void run();
private:
std::vector<std::thread> work_threads; //工作线程
std::mutex queue_mutex;
std::condition_variable condition; //必须与unique_lock配合使用
bool stop;
};//end class
//构造函数，创建线程
template <typename T>
threadPool<T>::threadPool(int number) : stop(false)
{
if (number <= 0 || number > MAX_THREADS)
throw std::exception();
for (int i = 0; i < number; i++)
{
std::cout << "created Thread num is : " << i <<std::endl;
work_threads.emplace_back(worker, this);//添加线程
//直接在容器尾部创建这个元素，省去了拷贝或移动元素的过程。
}
}
template <typename T>
inline threadPool<T>::~threadPool()
{
std::unique_lock<std::mutex> lock(queue_mutex);
stop = true;
condition.notify_all();
for (auto &ww : work_threads)
ww.join();//可以在析构函数中join
}
//添加任务
template <typename T>
bool threadPool<T>::append(T *request)
{
/*操作工作队列时一定要加锁，因为他被所有线程共享*/
queue_mutex.lock();//同一个类的锁
tasks_queue.push(request);
queue_mutex.unlock();
condition.notify_one(); //线程池添加进去了任务，自然要通知等待的线程
return true;
}
//单个线程
template <typename T>
void *threadPool<T>::worker(void *arg)
{
threadPool *pool = (threadPool *)arg;
pool->run();//线程运行
return pool;
}
template <typename T>
void threadPool<T>::run()
{
while (!stop)
{
std::unique_lock<std::mutex> lk(this->queue_mutex);
/* unique_lock() 出作用域会自动解锁 */
this->condition.wait(lk, [this] { return !this->tasks_queue.empty(); });
//如果任务为空，则wait，就停下来等待唤醒
//需要有任务，才启动该线程，不然就休眠
if (this->tasks_queue.empty())//任务为空，双重保障
{
assert(0&&"断了");//实际上不会运行到这一步，因为任务为空，wait就休眠了。
continue;
}
else
{
T *request = tasks_queue.front();
tasks_queue.pop();
if (request)//来任务了，开始执行
request->process();
}
}
}
#endif
8. atomic
为什么要使用atomic？
在cpu指令层面实现线程安全的变量， 就是在cpu指令层面实现的锁，比互斥锁开销更小，并发度更高
经典例子
#include<thread>
#include<atomic>
#include<iostream>
#include<vector>
int count = 0;
void add() {
for(int i = 0; i < 1000; i++) {
count++;
}
}
int main() {
std::vector<std::thread> thread_pool;
for(int i = 0; i < 4; i++) {
thread_pool.push_back(std::thread{[&] {
add();
}});
}
std::cout << cout << std::endl; // 最后加出来结果小于等于4000 因为count++ 翻译成汇编不止一条指令，多个线程并发指令会乱（读后写，写后读）
return 0;
}
// 解决方法
// 1. 上锁（std::mutex）
// 2. 使用atomic
std::atomic<int> count = 0;
std::atomic_int count = 0;
// 两种定义都可以
原子整数
不支持浮点数和自定义类型
#include<atomic>
std::atomic<int> count{100};
不可拷贝
底层原理
轻量级的锁， cpu指令提供支持，对原子变量的操作的指令不可分， 这样就保证了封闭性和可再现性
接口
返回load()
存入store()
9. 时间
#include<chrono>
// 时间段类型
std::chrono::seconds(30); // 30s
std::chrono::minutes(30); // 30min
std::chrono::milliseconds(30); // 30ms
std::chorono::microseconds(30) // 30us
// 时间点类型
std::chrono::steady_clock::time_point tp// 精度较高的cpu时间（还有操作系统时间，这里不给出）
运算：
时间点 + 时间段 = 时间点
时间点 - 时间点 = 时间段
auto t0 = std::chrono::steady_clock::now();
auto t1 = t0 + std::chrono::seconds(30);
auto time_duration = t1 - t0;
与this_thread配合使用见上面
10. 异步
同步&异步
在并发中：
- 同步： 完全按代码顺序执行
- 异步： 由中断或者信号驱动， 不一定按照顺序执行
举例
#include<async>
#include<future>
void do_this() {
...
}
void do_other_things() {
...
}
int main() {
std::future<int> res = std::async{[&] {
return do_this();
}};
// future就是保证以后会有这个变量过来， 但现在没有
// async是先挂起，用另一个线程偷偷在后台干这个函数，main就可以干自己的
do_other_things();
res.get(); // 这时候再获得挂起的函数的返回值， 如果此时在后台还没执行完， 就会等待函数执行完， 再得到函数的返回值
}
