并发实战

约 9039 字大约 30 分钟

2025-03-22

c++并发编程实战

线程相关的知识

线程的操作

c++11背景下面的线程api

std::thread

#include<thread>
std::thread my_thread(do_some);
//参数传入可调用参数类型

class back_ground
{
public:
void operator() ()const
{
do_something();
}
}

std::thread my_thread((back_ground()))//通过双重括号避免声明为函数的问题
std::thread my_thread{back_ground()}//可以通过大括号形式避免该问题

//lambda表达式也能解决这个问题
[]{}

detach()、join、joinable

struct func;
void f()
{
int loacl_state=0;
func my_func(local_state);
std::thread t(my_func);
try
{
do_something_in_current_thread();
	}
ctach(...)
{
t.join();
throw;
}
t.join();
}
//保证主线程即使出错，也能等待子线程结束再结束

//可以使用RAII的思想在创建一个类封装一个线程，在析构函数中释放线程，因为不管有没有抛出异常，栈上的类对象也是可以在推出抛出后销毁的
~threaad_guard(){
if(t.joinable())
{
t.join();
}
}

detach之后成为守护进程：一般守护进程是一个完全独立的个体，比如多窗口下的单独窗口的设计，将启动一个窗口作为一个函数，每次点开窗口就创建这个线程并且把线程分离。

对于不可拷贝的类型可以使用std::move作为参数进行传递。

**所有权转移：**线程允许移动，一个线程只能同时拥有一个单位，如果一个线程已经有一个正在运行的线程，此时再通过移动给其赋值，会出现std::terminate()终止程序，并不会抛出异常。

thread的一些探究： scoped_thread:可以把线程作为参数，并确保线程是可以释放的线程、后续有std::jthread

自动化创造线程的可以使用vector存储thread，然后emplace_back()可以传入线程创建需要仿函数以及参数，就可以直接在vector上面构造

线程数量的确定：

std::thread::hard_concurrency()：可以返回并发线程额数量，如果无法返回的时候就会返回0

线程的标识：stdid 有两种方式检索，stdthread没有关联数据就会返回std::thread::type的默认构造函数，这个返回值表示“无线程”。第二种，当前线程std::this_thread::get_id()

std::thread::id可以对比，作为例如set< stdid >的键值

共享数据

恶性竞争 不变量遭到修改，并发的修改就会有恶性竞争。

方法一：保证只有修改线程才能看到不变量的中间状态（保护机制，锁机制和条件变量）

方法二：保证数据结构和不变量能完成一系列不可分割的变化，就是无锁编程。（开销大）

方法三：利用事务的方式进行处理更新，所需的操作保存在日志之中，让后将操作进行合并统一一次性修改。(STM)软件事务内存，如果在数据结构被另一个线程修改之后，或者已经重启的情况下，提交无法进行。

互斥量：

std::mutex具有成员函数lock(),更好的有如下一些函数

#include<mutex>
//RAII的思想管理互斥量
std::lock_gurard<std::mutex> lock(some_mutex);//并不支持手动释放
//c++17提供了一些优化语法，当互斥量的类型简单时候可以直接std::lockguard lock(some_mutex),这是一种模版推导机制
std::unique_lock<std::mutex> lock(some_mutex);//可以手动释放可以自动释放

互斥量并非绝对安全的，通过指针或者引用可以让保护的内容被修改，所以尽量不要在有锁的时候调用其他函数。

即使有互斥量，条件竞争也是存在的比如多线程下的栈结构判空和size()函数这是基于接口的条件竞争。

对于接口的比如pop()和top()的条件竞争，可以使用合二为一的方法解决，但是此时pop()需要进行拷贝，并且是深拷贝，可能存在空间大小不够抛出异常std::bad_alloc

解决办法：

1、将变量引用作为参数，传入pop()获取弹出值：缺点资源消耗，且受到类型定义的限制，有些类型不支持赋值

2、采用无异常抛出的拷贝构造或移动构造函数：缺点std::is_nothrow_copy_constructible和std::is_nothrow_move_constructible局限太大，因为往往有些类型需要抛出异常，这样的栈结构就不能具有泛用性。

3、返回指向弹出值的指针，对弹出数据进行处理放到堆上并用shared_ptr进行托管，通过enable_shaerd_from_this进行返回，这种方案的开销巨大。

4、采用上述几种方式组合的情况。

#include<exception>
#include<memory>
#include<mutex>
#Include<stack> 
//继承自std::exception 并重写了exception的what()函数并且声明为noexcept即不抛出错误类型
struct empty_stack :std::exception
{
  cosnt char* what() const noexcept override
  {
      return "empty stack!\n"
  }
}

templa <typename T> 
class threadsafe_stack
    {
      private:
        std::stack<T> data;
        mutable std::mutex m;
              public:
        //无参构造
        threadsafe_stack()
            :data(std::stack<t> ()){}
        
        //拷贝构造函数，需要对被拷贝对象加锁
        threadsafe_stack(const threadsafe_stack& other)
        {
            std::lock_guard<std::mutex> lock(other.m);
            data = other.data;
      }
      
       //删除对象的赋值
      threadsafe_stack & operator =(const threadsafe_stack&) =delete;
       
      //放入栈
      void push(T new_vale)
       {
          std::lock_guard<std::mutex> lock(m);
           data.push(new_value);
       }
       
        std::shared_ptr<T> pop()
        {
          std:lock_gurard<std::mutex> lock(m);
           if(data.empty()) throw empty_stack();
          std::shared_ptr<T> const res(std::make_shared<T>(data.top()));
          data.pop();
           return res;
       }
      
      void pop(T& value)
      {
          std::lock_guard<std::mutex> lock(m);
          if(data.empty()) throw rmpty_stack();
          
           value = data.top();
            data.pop();
        }
       
        bool empty() const 
        {
            std::lock<std::mutex> lock(m);
            return data.empty();
        }
    };

大名鼎鼎的死锁，也就是哲学家就餐问题：

成因：互斥、占有、不可抢占、循环访问

一些解决锁的互斥的解决办法：

std::lock(std::mutex，std::mutex);
std::lock_guard<std::mutex>(std::mutex,std::adopt_lock);
//c++17
std::scoped_lock<std::mutex,std::mutex> //c++17允许的一种类型推导方法，支持对传递给的互斥量同时上锁，并且在析构函数钟释放，这是一种RAII的模式，可以替代上面两步做法的叠加
  
   //避免死锁的升级：避免使用嵌套锁，比如不要依次获得锁，如果非要如此，优先通过lock去锁住之后再进行
   //在持有锁的时候避免调用函数和外部代码
	//通过保证锁创建的顺序一致性，比如双端链表，为了减少锁的粒度，通过对前后本节点进行加锁来操作，但是要保证避免的顺序一致性，避免访问每个节点访问的锁的顺序不一致
   //使用层次锁的结构，加锁前检查每个互斥量对应的层数，以及每个线程使用的互斥量，检查枷锁操作是否可以进行。
   //层次锁的示例
   hierarchical_mutex high_level_mutex(10000);//1
	hierachical_mutex low_level_mutex(5000);//2
	hierarchical_mutex other_mutex(6000);//3

	int do_low_level_stuff();

int low_level_func()
	{
	std::lock_guard<hierarchical_mutex> lk(low_level_mutex);//4
   return do_low_level_stuff();
	}

	void high_level_stuff(int some_param);

	void high_level_func()
{
   std::lock_guard<hierarchial_mutex>lk(high_level_mutex);//6
    high_level_stuff(low_level_func());//5
	}

	void thread_a ()//7
	{
   high_level_func();
	}

	void do_other_stuff();

	void do_other_stuff()
	{
    high_level_func();//10
    do_other_stuff();	
	}

	void thread_b()//8
	{
   std::lock_guard<hierarchical_mutex> lk(other_mutex);//9
   other_stuff();
	}
//hieraachical_mutex 创建了三个实例，通过逐层递减进行构造，如果hierachical_mutex 上锁那么只能获取低层锁。

//意思是如果在high_level_func 调用 low_level_func 是可以允许通过的因为他锁的层次更加的高，如果层级一样不能同时持有多个锁。

//这hierarchical_mutex 实现在上锁的时候都会进行层级的判断

std::unique_lock

这个代价相对于lock_guard 的开销更大,但是允许示例不带互斥量。

stdadopt_lock

锁的粒度：

粗粒度的锁性能不好，但是细粒度的锁需要考虑的是并发访问状态下值在读取后的一瞬间就被修改的情况。

一种存粹保护共享数据初始化的机制：

延迟初始化：在初始化之前进行判断，转换为多线程代码的时候要注意：

 std::shared_ptr<some_resource> resource_ptr;
void foo()
 {
 if(!resource_ptr)//1
  {

   std::unique_lock<std::mutex> lk(resource_mutex);//所有线程在这里进行序列化
      if(!resource_ptr)
     {
   resource_ptr.reset(new some_resource);//当是多线程代码的时候需要对这个地方进行保护
 		}
     //引入互斥量

}
 resource_ptr->do_something();
 }
 //但是这个模式有问题，问题在于存在潜在的条件竞争，因为在1处的时候可以，此时的状态可能被其他些写的线程影响。

引入了stdcall_once(可以安全的检测指针是否被其他线程修改，这样进行初始化)，使用std::call_once 相较于显式使用互斥量开销更加小，特别是初始化完成之后。

std::shared_ptr<some_resource> resource_ptr;
std::once_flag resource_flag; //1
void init_resource()
{
 resource_ptr.reset(new some_resource);
}
void foo()
{
std::call_once(resource_flag,init_resource);//可以完整的一次初始化  resource_ptr->do_something();
}

初次调用send_data和data_packet 完成初始化过程，静态成员变量也会村寨这种问题在c++11之后，这问题被解决，因为c++11规定所有的初始化以及定义完全发生在一个线程中，；并没有其他线程可以在初始化完成之前对其进行处理。

保护不常更新的数据结构，采用读写锁（比如DNS表的更新处理）

c++17中支持了两非常好的互斥量----std::shared_mutex和std::shared_timed_mutex,c++14只提供了std::shared_timed_mutex,而c++11就需要采用boost库提供的互斥量，shared_timed_mutex的操作方式更多，但是std::shared_mutex性能更好。

只要有一个非独占锁被锁住，那么所有的请求独占锁的线程就会被阻塞，但是非独占锁的请求不会受到限制，等所有其他的线程释放了非独占锁之后，才能拥有独占锁，拥有独占锁其他所有线程不能拿到任何锁，只有独占锁才能允许写操作。

非更新操作使用stdunique_lock<>()/std::lock_guard<>()

std::recursive_mutex类可以对同一个实例多次上锁，使用嵌套锁可以用在并发访问的类，使用互斥量保护其成员数据

内存模型以及原子操作

简单介绍

主要内容包括：

内存模型
标准库里面的原子类型
如何使用原子类型
使用原子操作同步线程

c++标准中有一个重要特性就是多线程感知内存模型。

原子操作以及原子类型

一方面是内存布局
一方面是并发

原子操作

原子操作并没有指定访问的顺序，反而是将程序拉回定义行为的区域内。

原子操作是一个不可分割的操作。

标准原子类型：

标准原子类型定义在<atomic>,其中类型的操作都是原子的。（标准的实现可能是拥有一份is_lock_free()函数，然后这个函数可以让用户直接查询原子类型的操作是直接用的原子指令（x.is_lock_free()返回true,还是内部使用了一个锁结构（x.is_lock_free()返回false））.

原子类型要可以替代互斥量，就最好不用互斥量实现，因为用互斥量实现不可能有任何的提升，标准库提供了一组宏用来在编译的时候对各种整形原子操作是否无锁进行判断。c++17里面有一个static constexpr 成员变量，如果相应的硬件上的原子类型是无锁类型，那么X::is_always_lock_free将返回true。

通常标准院子类型不支持拷贝和复制，他们没有拷贝构造函数以及拷贝赋值操作符。但是可以隐式转化，所以这些类型支持赋值，通过调用load()、store()、exchange()、compare_exchange_weak()、compare_exchange_strong()。他们都支持使用复合运算符:+=、 -= 、*= 、|=等等在指针以及整型的特化类型下面支持++和--(对应了函数fetch_add()、fetch_or()等等。

以上的哪些函数都有一种叫做内存序参数的变量，这个参数可以用来指定存储的顺序。

重要

store: memory_order_relaxed、memory_order_relaxed、memory_order_seq_cst；

load : memory_order_relaxed 、memory_order_consume、memory_order_acquire、memory_order_seq_cst

Read-modify-write(读-改-写)：memory_order_relaxed、memory_order_consume、memory_order_acquire、

memory_order_release、memory_order_acq_rel、memory_order_seq_cst

并发实战

c++并发编程实战

线程相关的知识

线程的操作

共享数据

同步操作

等待事件或者条件

future

promise & packaged_task

限定等待时间

内存模型以及原子操作

原子操作以及原子类型