复习C++标准库多线程的基本使用

淩亂°似流年 2023-01-19 04:39 12阅读 0赞

参考书籍：《C++并发编程实战》第一版&&第二版

参考文档：[https://zh.cppreference.com/w/cpp/thread][https_zh.cppreference.com_w_cpp_thread]

# 第一章：你好，C++并发世界 #

（略）

# 第二章：管理线程 #

（2021-05-07 笔记）

类 std::thread 表示单个执行线程，thread 对象允许 move 但不能 copy 。 构造函数：

template< class Function, class... Args >
    explicit thread( Function&& f, Args&&... args );
    构造 std::thread 实例并将它关联至新创建的执行线程。复制或者移动 func 和 args 的每个元素到内部存储中，并持续在新的执行线程的整个声明周期。在新的线程上执行 invoke。
    移动或按值复制线程函数的参数。若需要传递引用参数给线程函数，则必须包装它（例如用 std::ref 或 std::cref ）。
    忽略来自函数的任何返回值。若函数抛异常，则调用 std::terminate 。为将返回值或异常传递回调用方线程，可使用 std::promise 或 std::async 。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dvbmdqaWFuYm8xOTky_size_16_color_FFFFFF_t_70][]

基本操作：

bool joinable() const noexcept; 
    检查 std::thread 对象是否标识为活跃的执行线程。具体而言，若 get_id()!=std::thread::id() 则返回 true 。故默认构造的 thread 不可 join。
    结束执行代码，但仍未 join 的线程仍被当作活跃的执行线程，从而可 join。
    调用 join() 或者 detach() 后也为 false 。
    
    std::thread::id get_id() const noexcept;
    返回标识与 *this 关联的线程的 std::thread::id 。若无关联的线程，则返回默认构造的 std::thread::id 。
    
    static unsigned int hardware_concurrency() noexcept;
    返回实现支持的并发线程数。应该只把该值当做提示。
    
    void join();
    阻塞当前线程直至 *this 所标识的线程结束其执行。
    
    void detach();
    从 thread 对象分离执行线程，允许执行独立地持续。一旦该线程退出，则释放任何分配的资源。
    调用 detach 后 *this 不再占有任何线程。
    
    void swap( std::thread& other ) noexcept;
    交换两个 thread 对象的底层 handle 。

练习：

#include <iostream>
    #include <thread>
    
    void func1(int a, int& b){
    	b = a;
    }
    
    //守护类，释放时自动join
    class scoped_thread
    {
    public:
    	explicit scoped_thread(std::thread& t)
    		: th(std::move(t)){
    	}
    	~scoped_thread(){
    		if (th.joinable())
    			th.join();
    	}
    	scoped_thread(const scoped_thread&) = delete;
    	scoped_thread& operator = (const scoped_thread&) = delete;
    
    private:
    	std::thread th;
    };
    
    int main()
    {
    	std::thread t1([] { });
    	if (t1.joinable()) {
    		//等待线程结束
    		t1.join();
    	}
    
    	std::thread t2([] { });
    	if (t2.joinable()) {
    		//分离不等待，不会在释放时调用std::terminate
    		t2.detach();
    	}
    
    	//线程函数的参数按值移动或复制。
    	//如果需要将引用参数传递给线程函数，则必须将其包装（例如，使用std::ref或std::cref）。
    	int a = 1, b = 2;
    	std::thread t3(func1, a, std::ref(b));
    	if (t3.joinable())
    		t3.join();
    
    	//std::thread可移动但不可复制
    	//std::thread t4(std::move(t3));
    	scoped_thread scope(t3);
    
    	return 0;
    }

# 第三章：在线程间共享数据 #

（2021-05-16 笔记）

类 std::mutex 是能用于保护共享数据免受从多个线程同时访问的同步原语，即互斥锁、互斥元、互斥体。

std::mutex 基本操作：

void lock();
    锁定互斥。若另一线程已锁定互斥，则到 lock 的调用将阻塞执行，直至获得锁。
    
    bool try_lock();
    尝试锁定互斥。立即返回。成功则锁定并返回 true ，否则返回 false 。
    
    void unlock();
    解锁互斥。互斥必须为当前执行线程所锁定，否则行为未定义。

std::timed\_mutex 多两个接口：

template< class Rep, class Period >
    bool try_lock_for( const std::chrono::duration<Rep,Period>& timeout_duration );
    尝试锁互斥。阻塞直到经过指定的 timeout_duration 或得到锁，取决于何者先到来。成功获得锁时返回 true ， 否则返回 false 。
    
    template< class Clock, class Duration >
    bool try_lock_until( const std::chrono::time_point<Clock,Duration>& timeout_time );
    尝试锁互斥。阻塞直至抵达指定的 timeout_time 或得到锁，取决于何者先到来。成功获得锁时返回 true ，否则返回 false 。

死锁：两个或多个线程互相持有对方所需要的资源，导致互相等待。

初始化：C++11 开始可用 static 或者 std::call\_once 来防范初始化时的多线程竞争。

练习：

#include <iostream>
    #include <string>
    #include <chrono>
    #include <list>
    #include <thread>
    #include <mutex>
    #include <shared_mutex>
    
    //1.自定义多线程操作安全的链表
    template<typename T>
    class MyList
    {
    public:
    	bool pop(T& value) {
    		//给需要多线程操作的接口上锁，解决竞争
    		std::lock_guard<std::mutex> guard(mtx);
    		if (data.size() > 0) {
    			value = data.front();
    			data.pop_front();
    			return true;
    		}
    		return false;
    	}
    
    	void push(const T& value) {
    		std::lock_guard<std::mutex> guard(mtx);
    		data.push_back(value);
    	}
    
    	//同时操作多个mutex时要避免死锁
    	friend void swap(MyList<T>& left, MyList<T>& right) {
    		if (&left == &right)
    			return;
    		// 用 std::lock 获得二个锁，而不用担心死锁
    		std::lock(left.mtx, right.mtx);
    		std::lock_guard<std::mutex> lguard(left.mtx, std::adopt_lock);
    		std::lock_guard<std::mutex> rguard(right.mtx, std::adopt_lock);
    		// 等价代码（若需要 unique_locks ，例如对于条件变量）
            //std::unique_lock<std::mutex> lk1(left.mtx, std::defer_lock);
            //std::unique_lock<std::mutex> lk2(right.mtx, std::defer_lock);
            //std::lock(lk1, lk2);
            // C++17 中可用的较优解法
            //std::scoped_lock lk(left.mtx, right.mtx);
    
    		std::swap(left.data, right.data);
    	}
    
    private:
    	std::list<T> data;
    	//加mutable是为了在const接口中使用
    	mutable std::mutex mtx;
    };
    
    //2.初始化时保护共享数据
    std::once_flag onceflag;
    void init() {
    
    }
    int* instance() {
    	static int i;
    	return &i;
    }
    
    //3.递归锁(C++11) 
    class DemoA {
    public:
    	void func1() {
    		std::lock_guard<std::recursive_mutex> lk(m);
    		//递归锁定了func2中的
    		func2();
    	}
    	void func2() {
    		std::lock_guard<std::recursive_mutex> lk(m);
    	}
    private:
    	std::recursive_mutex m;
    };
    
    //4.读写锁(C++17)
    class ThreadSafeCounter {
    public:
    	// 多个线程/读者能同时读计数器的值。
    	unsigned int get() const {
    		//share共享，只读
    		std::shared_lock<std::shared_mutex> lock(mtx);
    		return value;
    	}
    	// 只有一个线程/写者能增加/写线程的值。
    	void increment() {
    		//unique独占，读写
    		std::unique_lock<std::shared_mutex> lock(mtx);
    		value++;
    	}
    	// 只有一个线程/写者能重置/写线程的值。
    	void reset() {
    		std::unique_lock<std::shared_mutex> lock(mtx);
    		value = 0;
    	}
    
    private:
    	mutable std::shared_mutex mtx;
    	unsigned int value = 0;
    };
    
    //5.时间限定的锁定(C++11)
    std::timed_mutex tmtx;
    void time_lock() {
    	//指定时间段，如10ms后超时，或者unlock
    	//std::chrono::duration
    	//if (tmtx.try_lock_for(std::chrono::minutes(1))) {
    	if (tmtx.try_lock_for(std::chrono::duration<int, std::ratio<60>> (1))) {
    		//...
    		tmtx.unlock();
    	}
    	//指定时间点，如1点钟超时，或者unlock
    	//std::chrono::time_point
    	auto t1 = std::chrono::high_resolution_clock::now();
    	if (tmtx.try_lock_until(t1 + std::chrono::minutes(1))) {
    		//...
    		tmtx.unlock();
    	}
    }
    
    int main()
    {
    	//1.线程安全链表
    	MyList<int> queue;
    	queue.push(10);
    	MyList<int> queue2;
    	queue2.push(20);
    	swap(queue, queue2);
    
    	int value;
    	if (queue.pop(value)) {
    		std::cout << "pop value:" << value << std::endl;
    	}
    
    	//2.初始化时保护共享数据
    	//static 或者 std::call_once
    	{
    		std::call_once(onceflag, init);
    		int *i = instance();
    	}
    
    	system("pause");
    	return 0;
    }

# 第四章：同步并发操作 #

（2021-05-22笔记）

本章主要讲了条件变量和 future 两种同步方式，条件变量可重复的执行任务并等待， future 只针对单次任务。

对于 C++20 新增的信号量、闩 latch 、屏障 barrier 之后再学。

1.std::condition\_variable 条件变量能用于阻塞一个线程，或同时阻塞多个线程，直至另一线程修改共享数据并通知 std::condition\_variable。

*  修改数据的线程持有互斥锁并对数据进行操作，然后 notify\_all 或 notify\_one 唤醒阻塞的线程。
 *  阻塞的线程 wait 到唤醒并获得互斥锁，然后操作数据。

这里面要注意的就是虚假唤醒问题，需要给 wait 加一个谓词来判断当前是否满足唤醒条件而不是因为中断等问题被激活，不满足就继续睡。

2.std::future 提供访问异步操作结果的机制：

*  （通过 std::async 、 std::packaged\_task 或 std::promise 创建的）异步操作能提供一个 std::future 对象给该异步操作的创建者。
 *  然后，异步操作的创建者能用各种方法查询、等待或从 std::future 提取值。若异步操作仍未提供值，则这些方法可能阻塞。
 *  异步操作准备好发送结果给创建者时，它能通过修改链接到创建者的 std::future 的共享状态（例如 std::promise::set\_value ）进行。

注意， std::future 所引用的共享状态不与另一异步返回对象共享（与 std::shared\_future 相反，shared 是可以引用同一状态的）。

这几个类的接口都很简单，直接照着 demo 练习：

#include <iostream>
    #include <string>
    #include <chrono>
    #include <list>
    #include <thread>
    #include <mutex>
    #include <shared_mutex>
    #include <condition_variable>
    #include <future>
    
    //1.condition_variable条件变量
    class TestCV
    {
    public:
    	//处理数据
    	void process()
    	{
    		std::unique_lock<std::mutex> lk(mtx);
    		while (runing)
    		{
    			//wait时不持有mtx，阻塞等待
    			//为了防止虚假唤醒，需要谓词判断
    			//wait死等直到唤醒，唤醒后获得mtx，然后检查谓词
    			//wait_for等到唤醒or等一定时长
    			//wait_until等到唤醒or等到某个时间点
    			cv.wait(lk, [this] {
    				return (!runing || ready);
    			});
    			if (!runing)
    				break;
    			ready = false;
    
    			//开始处理数据
    			//process_datas(datas);
    		}
    	}
    
    	//接收数据
    	void recv()
    	{
    		while (runing) {
    			//获取数据
    			//read_datas();
    			if (!runing)
    				break;
    
    			std::lock_guard<std::mutex> lk(mtx);
    			//set_datas(datas);
    			ready = true;
    			//唤醒某个wait
    			//也可以cv.notify_all()唤醒所有wait
    			cv.notify_one();
    		}
    	}
    
    private:
    	std::mutex mtx;
    	std::condition_variable cv;
    	std::list<int> datas;
    	bool ready;
    	bool runing;
    };
    
    //4.future+promise
    void processP(std::promise<int> ret)
    {
    	std::this_thread::sleep_for(std::chrono::seconds(1));
    	ret.set_value(1992);
    }
    
    int main()
    {
    	//2.future+promise
    	std::promise<int> promise_p;
    	std::future<int> future_p = promise_p.get_future();
    	//std::thread thread_p(processP, std::move(promise_a));
    	std::thread thread_p([&promise_p] {
    		std::this_thread::sleep_for(std::chrono::seconds(1));
    		promise_p.set_value(1992);
    		//线程结束时结果状态就绪，所以得先join再get
    		//promise_a.set_value_at_thread_exit(123);
    	});
    	//可以future.wait等待完成，但是get会等待结果不用先wait
    	//如果是返回void，直接调用wait等完成就行了
    	//也可以在轮询中wait_for,wait_until等一会儿
    	std::cout << "future p result:" << future_p.get() << std::endl;
    	thread_p.join();
    
    	//3.future+packaged_task
    	std::packaged_task<int()> task_t([] {  return 1993; });
    	std::future<int> future_t = task_t.get_future();
    	//在线程上运行task
    	std::thread thread_t(std::move(task_t));
    	thread_t.detach();
    	std::cout << "future t result:" << future_t.get() << std::endl;
    
    	//4.future+async
    	//std::launch::async多线程，异步执行
    	//std::launch::deferred同步，等到获取值时执行
    	std::future<int> future_a = std::async(
    		std::launch::async, []() {
    		return 1994;
    	});
    	std::cout << "future a result:" << future_a.get() << std::endl;
    
    	//5.shared_future
    	//future可move但不能并发访问，同事get只能操作一次
    	//shared_future可move，可复制并引用同一状态，
    	//如果每个线程通过自己的shared_future去访问状态，就是安全的
    	//shared_future可以通过future构造，或者引用其他shared_future
    	//std::shared_future<int> s_future = future_a.share();
    	//std::shared_future<int> s_future(std::move(future_a));
    	//s_future.wait();
    	//std::cout << "shared_future result:" << s_future.get() << std::endl;
    
    	//书上新增的拓展内容，闩latch和屏障barrier已在C++20增加，后面单独写
    
    	system("pause");
    	return 0;
    }

### 补：同步并发操作 C++20 部分 ###

（略）

# 第五章：C++内存模型和原子类型操作 #

（2021-05-29 笔记）

C++11 制定了 6 中内存顺序约束：（很多地方还没明白，以后明白了再来更新下）

memory_order_relaxed
    不会在多线程的内存访问间强加顺序，只保证原子性和修改顺序一致性。其他线程可能读到新值，也可能读到旧值。
    如 std::shared_ptr 的引用计数器，因为这只要求原子性，但不要求顺序或同步（注意 std::shared_ptr 计数器自减要求与析构函数进行获得释放同步）
    
    memory_order_consume
    文档没看明白，而且还不完善不建议使用，可以用代价更高的 memory_order_acquire
    
    memory_order_acquire
    获取操作，一般可与 memory_order_release 配对使用，在该原子变量上施加 release 语义的操作发生之后，acquire 可以保证读到所有在 release 前发生的写入
    
    memory_order_release
    写入操作，一般可与 memory_order_acquire 配对使用，在这之前的读写操作顺序都不会被重排到此操作之后
    当前线程内的所有写操作，对于其他对这个原子变量进行 acquire 的线程可见
    当前线程内的与这块内存有关的所有写操作，对于其他对这个原子变量进行 consume 的线程可见
    
    memory_order_acq_rel
    对读取和写入施加 acquire-release 语义，无法被重排
    可以看见其他线程施加 release 语义的所有写入，同时自己的 release 结束后所有写入对其他施加 acquire 语义的线程可见
    
    memory_order_seq_cst
    如果是读取就是 acquire 语义，如果是写入就是 release 语义，如果是读取+写入就是 acquire-release 语义
    同时会对所有使用此 memory order 的原子操作进行同步，所有线程看到的内存操作的顺序都是一样的，就像单个线程在执行所有线程的指令一样

参照： [https://www.cnblogs.com/lizhanzhe/p/10893016.html][https_www.cnblogs.com_lizhanzhe_p_10893016.html]

类模板 std::atomic 用于表示一个原子类型，对原子对象的访问可以建立线程间同步，并按 std::memory\_order 所对非原子内存访问定序。（原子，即表示操作不可分割，类似数据库实务操作）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dvbmdqaWFuYm8xOTky_size_16_color_FFFFFF_t_70 1][]

基本操作：

T operator=( T desired ) noexcept;
    原子地赋 desired 给值原子变量。等价于 store(desired) 
    原子变量非可复制赋值 (CopyAssignable) 
    
    bool is_lock_free() const noexcept;
    检查此类型所有对象上的原子操作是否免锁
    
    void store( T desired, std::memory_order order = std::memory_order_seq_cst ) noexcept;
    原子地以 desired 替换当前值。按照 order 的值影响内存。
    order 必须是 std::memory_order_relaxed 、 std::memory_order_release 或 std::memory_order_seq_cst 之一。否则行为未定义。
    
    T load( std::memory_order order = std::memory_order_seq_cst ) const noexcept;
    原子地加载并返回原子变量的当前值。按照 order 的值影响内存。
    order 必须是 std::memory_order_relaxed 、 std::memory_order_consume 、 std::memory_order_acquire 或 std::memory_order_seq_cst 之一。否则行为未定义。
    
    operator T() const noexcept;
    原子地加载并返回原子变量的当前值。等价于 load() 
    
    
    T exchange( T desired, std::memory_order order = std::memory_order_seq_cst ) noexcept;
    原子地以 desired 替换底层值。操作为读-修改-写操作。根据 order 的值影响内存。
    
    compare_exchange_weak
    compare_exchange_strong
    原子地比较原子对象与非原子参数的值，若相等则进行交换，若不相等则进行加载
    
    void wait( T old, std::memory_order order = std::memory_order::seq_cst ) const noexcept;
    C++20 原子的等待。
    表现为如同进行下列步骤：
    比较 this->test(order) 的值表示与 old 。old 为要检测 atomic 的对象不再含有的值
    若它们相等，则阻塞直至 *this 被 notify_one() 或 notify_all() 提醒，或线程被虚假地除阻。
    这些函数保证仅若值更改才返回，即使底层实现虚假地除阻。
    
    void notify_one() noexcept; 
    C++20 唤醒一个在该对象上阻塞等待的线程
    void notify_all() noexcept;
    C++20 唤醒所有在该对象上阻塞等待的线程
    
    is_always_lock_free [静态](C++17)
    指示该类型是否始终免锁
    
    fetch_add
    fetch_sub
    fetch_and
    fetch_or
    fetch_xor
    原子地操作对象值，并返回先前保有的值
    
    以及一些加减自增自减等操作

练习：

#include <iostream>
    #include <thread>
    #include <atomic>
    
    //1.atomice_flag 无锁原子变量
    //C++20 开始ATOMIC_FLAG_INIT被废弃，默认以clear初始化
    std::atomic_flag flag = ATOMIC_FLAG_INIT;
    
    void test_flag()
    {
    	//原子的设置为false
    	//void clear( std::memory_order order = std::memory_order_seq_cst ) noexcept;
    	flag.clear();
    	//原子的设置为true并获取其先前值
    	//bool test_and_set(std::memory_order order = std::memory_order_seq_cst) noexcept;
    	bool ret = flag.test_and_set();
    	//C++20新增四个接口
    	//原子地返回标志的值
    	//bool test( std::memory_order order = std::memory_order::seq_cst ) const noexcept;
    	ret = flag.test();
    	//进行原子等待操作。表现为如同进行下列步骤：
    	//比较 this->test(order) 与 old 。
    	//若它们相等，则阻塞直至* this 被 notify_one() 或 notify_all() 提醒，或线程被虚假地除阻。
    	//这些函数保证仅若值更改才返回，即使底层实现虚假地除阻。
    	//void wait( bool old, std::memory_order order = std::memory_order::seq_cst ) const noexcept;
    	flag.wait(true);
    	//唤醒阻塞等待该变量的线程
    	//void notify_one() noexcept;
    	flag.notify_one();
    	//void notify_all() noexcept;
    	flag.notify_all();
    }
    
    int main()
    {
    	system("pause");
    	return 0;
    }

# 第六章：设计基于锁的并发数据结构 #

# 第七章：设计无锁的并发数据结构 #

# 第八章：设计并发代码 #

# 第九章：高级线程管理 #

# 第十章：多线程应用的测试与调试 #

[https_zh.cppreference.com_w_cpp_thread]: https://zh.cppreference.com/w/cpp/thread
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dvbmdqaWFuYm8xOTky_size_16_color_FFFFFF_t_70]: /images/20221021/6ebc2824cdc440d18a8fe01417a711c0.png
[https_www.cnblogs.com_lizhanzhe_p_10893016.html]: https://www.cnblogs.com/lizhanzhe/p/10893016.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dvbmdqaWFuYm8xOTky_size_16_color_FFFFFF_t_70 1]: /images/20221021/87e2f4039a9d4bbd9a17688f20d0d49a.png