C++11并发库

前情提要：学习C++11的并发库必须要是对Linux的多线程了解，这里我们并不是从零开始讲解并发库，不会涉及到线程知识的讲解。

1. thread库

🧵线程库的文档资料

🗨️thread库是对各个系统底层的库进行了封装，（Linux --> pthread, Windows --> Thread），所以**c++11**提供的线程库是跨平台的支持线程使用的库。同时继承了**C++11**的面向对象特性，使其使用方式更加的便捷。

  • 在线程库中，构造函数有4种，日常最实用的有两种，它支持传入一个可调用对象和所需的参数即可，相比于pthread_create而言这里不再局限与传递函数指针，其次传递参数的方法也更加的方便

  • 另外也可以使用第一个和第四个可以配合起来创建线程我们可以使用右值引用对象移动赋值或者是构造给另一个线程对象。

  • 第3个现在已经删除，所以可以看作是，线程对象不支持拷贝。

  • join是主线程结束前强制阻塞等待创建的从线程，否则主线程结束，进程就结束了，从线程可能在运行就会被强行终止！

  • clas thread::id 是一个thread的内部类用来表示线程id, 支持比较大小，流插入和提取，比如通过特化hash仿函数做unordered_map and unordered_set的id等。底层的角度来看还是C++11的线程库进行了各个平台的移植性处理，所以只能使用一个类来进行封装。线程对象可以使用get_id来获取线程id，在执行体种通过this_thread:get_id()来获取线程id。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
default (1)
thread() noexcept;

initialization(2)
template < class Fn, class... Args>
explicit thread(Fn && fn, Args&&... args);

copy[deleted](3)
thread(const thread&) = delete;
copy[deleted](3)
thread & operator= (const thread&) = delete;

move(4)
thread(thread && x) noexcept;
move(4)
thread & operator= (thread && rhs) noexcept;

// pthread库 
int pthread_create(pthread_t * tidp, const pthread_attr_t * attr, void*
		(*start_rtn)(void*), void* arg);
// windows线程创建API 
HANDLE CreateThread(
		LPSECURITY_ATTRIBUTES lpThreadAttributes,//SD
		SIZE_T dwStackSize,//initialstacksize
		LPTHREAD_START_ROUTINE lpStartAddress,//threadfunction
		LPVOID lpParameter,//threadargument
		DWORD dwCreationFlags,//creationoption
		LPDWORD lpThreadId//threadidentifier
	)
void join();

🌈这里我们展示一下简单的使用<thread>库来创建线程：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
#include <iostream>
#include <thread>
#include <mutex>
#include <vector>
using namespace std;

void Print(int n, int i){
	for (; i < n; i++){
		cout << this_thread::get_id() << ":" << i << endl;
	}
	cout << endl;
}

int main(){
	thread t1(Print, 10, 0);
	thread t2(Print, 20, 10);
	
	// 获取线程id
	// cout << t1.get_id() << endl;
	// cout << t2.get_id() << endl;
	
	// 接下来还需要阻塞等待线程运行完成！
	t1.join();
	t2.join();
	
	return 0;
}

2. this_thread

参考资料：

this_thread是一个命名空间，主要封装了4个全局接口函数：

  • get_id: 是执行当前线程的id;

  • yeild是主动让出当前线程的执行权1，让其他线程先执行。这个函数依赖于与实现，特别是取决于使用中的OS调度机制和系统状态，例如先进先出实时调度器（linux的SCHED_FIFO), 会先挂起当前线程并将其放到准备运行的同优先级现成的队列为，五其他线程就没有效果。

  • sleep_for阻塞当前线程执行，至少经过指定的sleep_duration, 因为资源调度的竞争延迟这个函数的阻塞时间可能长于指定的 **sleep_duration.**

  • sleep_until, 阻塞当前线程的执行, 直到抵达指定的sleep_point, 通过杨因为资源调度的问题，这个函数的阻塞时间也是就可能会晚于这个**sleep_point;**

⌛ **chorono**

🕝 **duration**

用来管一个相对时间的类

⏰ **time_point**

用发来管路一个绝对时间点的类

1
2
3
4
5
template <class Clock, class Duration>
void sleep_until(const choron::time_point<Clock, Duration>& abs_time);

template<class Rep, class Period>
void sleep_for(const choron::duration<Req, Period>& rel_time);

下面是三个展示时间库使用的案例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
#define _CRT_SECURE_NO_WARNINGS 1

#include <iostream> // std::cout, std::endl
#include <thread> // std::this_thread::sleep_for
#include <chrono> // std::chrono::seconds
int m24ain()
{
	std::cout << "countdown:n";
	for (int i = 10; i > 0; --i) {
		std::cout << i << std::endl;
		std::this_thread::sleep_for(std::chrono::seconds(1));
	}
	std::cout << "Lift off!n";
	return 0;
}
// this_thread::sleep_for example
#include <iostream> // std::cout
#include <iomanip> // std::put_time
#include <thread> // std::this_thread::sleep_until
#include <chrono> // std::chrono::system_clock
#include <ctime> // std::time_t, std::tm, std::localtime, std::mktime
int m34ain()
{
	using std::chrono::system_clock;
	std::time_t tt = system_clock::to_time_t(system_clock::now());

	struct std::tm* ptm = std::localtime(&tt);
	std::cout << "Current time: " << std::put_time(ptm, "%X") << &#x27;n&#x27;;

	std::cout << "Waiting for the next minute to begin...n";
	++ptm->tm_min; ptm->tm_sec = 0;
	std::this_thread::sleep_until(system_clock::from_time_t(mktime(ptm)));

	std::cout << std::put_time(ptm, "%X") << " reached!n";

	return 0;
}
int main()
{
	using namespace std::chrono;

	duration<int, std::ratio<60 * 60 * 24> > one_day(1);

	system_clock::time_point today = system_clock::now();
	system_clock::time_point tomorrow = today + one_day;
	time_t tt;

	tt = system_clock::to_time_t(today);
	std::cout << "today is: " << ctime(&tt);

	tt = system_clock::to_time_t(tomorrow);
	std::cout << "tomorrow will be: " << ctime(&tt);

	return 0;
}

3. mutex

🌐 学习网站：

mutex是封装的互斥锁类，用于保护临界区域的共享数据。mutex， 主要提供lock 和 unlock 两个函数。

  1. **mutex**提供排他性非递归语义：

  • 递归方线程从他调用成功lock或者是try_lock开始，到他们调用unlock为止占有mutex

  • 当一个线程已经占有mutex时，其他的县城如果视图要求mutex的所有权，那么就会阻塞（就是对于LOCK的调用，对于try_lock的调用则就是返回false。

  1. 如果mutex在正在被占有的情况下被销毁，或者是在占有mutex的时候被销毁那么这样的行为就是未定义行为。

  2. 先看下面的代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
#include <iostream>
#include <chrono> 
#include <thread>
#include <mutex>
using namespace std;
void Print(int n, int& rx, mutex& rmtx)
{
	rmtx.lock();
	for (int i = 0; i < n; i++)
	{
		// t1 t								++rx;
	}
	rmtx.unlock();
}
int main()
{
	int x = 0;
	mutex mtx;
	// 这⾥必须要⽤ref()传参，现成中拿到的才是x和mtx的引⽤，具体原因需要看下⾯thread源码中的分析
// https://legacy.cplusplus.com/reference/functional/ref/?kw=ref
	thread t1(Print, 1000000, ref(x), ref(mtx));
	thread t2(Print, 2000000, ref(x), ref(mtx));
	t1.join();
	t2.join();
	cout << x << endl;
	return 0;
}

为什么在给线程对象传递可调用对象的参数包时，需要使用std::ref 呢？

其实是thread本质还是系统库的提供的磕掉哟个的线程API的封装，thread构造取到参数包以后，就调用创建线程的API， 还是需他将参数包打包成一个结构体传递过去，那么打包成结构体时。就需要将参考包对象拷贝给结构体对象，使用**ref****传递参数，会让结构体中的对应从参数成员类型推到引用，**这样才能真正的实现引用传值。

  1. recursive_mutex跟mutex完全类似， recursive_mutex提供了排他性所有权语义：

调用方线程在他成功调用lock或try_lock开始使其就占有recursive_mutex。在此期间，线程可以进行对**lock**或者是**try_lock**的附加调用。所有权时期线程进行的匹配次数的**unlock**调用时结束

  1. 线程占用recursive_mutex时，若其他的线程试图占用recursive_mutex的所有权，则它们将阻塞（对于调用lock) 或者是收到 false对于调用try_lock

time_mutex和mutex其实完全类似，只是额外的提供了try_lock_for和try_lock_untile的接口，这两个接口和try_lock相似，只是他不会马上返回，而是直接进入阻塞状态，直到时间条件到了或者是解锁了就会唤醒视图获取锁资源。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
// timed_mutex::try_lock_for example
#include <iostream> // std::cout
#include <chrono> // std::chrono::milliseconds
#include <thread> // std::thread
#include <mutex> // std::timed_mutex
std::timed_mutex mtx;
void fireworks(int i)
{
	//std::cout << i;
		// waiting to get a lock: each thread prints "-" every 200ms:
	while (!mtx.try_lock_for(std::chrono::milliseconds(1000)))
	{
		std::cout << "-";
	}
	std::cout << i;
	// got a lock! - wait for 1s, then this thread prints "*"
	std::this_thread::sleep_for(std::chrono::milliseconds(5000));
	std::cout << "*n";
	mtx.unlock();
}
#define n 10
int main()
{
	std::thread threads[n];
	// 利⽤移动赋值的⽅式，将创建的临时对象(右值对象)移动赋值给创建好的空线程对象 
	for (int i = 0; i < n; ++i)
		threads[i] = std::thread(fireworks, i);
	for (auto& th : threads)
		th.join();
	return 0;
}

4. lock_guard

  • 💂🏻lock_guard是C++11提供的支持RAII的方式管理互斥锁资源的类，这样可以更加有效的防止异常等原因导致死锁问题。他们的大致原理如下面的代码：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
// ⽰例1 
#include <iostream>
#include <chrono> 
#include <thread>
#include <mutex>
using namespace std;
template < class Mutex>
class LockGuard
{
public:
	LockGuard(Mutex& mtx)
		: _mtx(mtx)
	{
		_mtx.lock();
	}
	~LockGuard()
	{
		_mtx.unlock();
	}
private:
	Mutex& _mtx;
};
int main()
{
	int x = 0;
	mutex mtx;
	auto Print = [&x, &mtx](size_t n) {
		//lock_guard<mutex> lock(mtx);
		LockGuard<mutex> lock(mtx);
		//mtx.lock();
		for (size_t i = 0; i < n; i++)
		{
			++x;
		}
		//mtx.unlock();
		};
	thread t1(Print, 1000000);
	thread t2(Print, 2000000);
	t1.join();
	t2.join();
	cout << x << endl;
	return 0;
}
  • lock_guard的功能简单纯粹，仅仅支持RAII的方式来管理锁对象。也可以在构造的时候通过传参adopt_lock_t的adopt_lock的对象管理已经lock的锁对象。**lock_guard**类不支持拷贝
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
/*locking(1)
explicit lock_guard(mutex_type& m);
adopting(2)
lock_guard(mutex_type& m, adopt_lock_t tag);
copy[deleted](3)
lock_guard(const lock_guard&) = delete;*/

#include <iostream>
#include <chrono> 
#include <thread>
#include <mutex>
using namespace std;
std::mutex mtx; // mutex for critical section
void print_thread_id(int id) {
	mtx.lock();
	std::lock_guard<std::mutex> lck(mtx, std::adopt_lock);
	std::cout << "thread #" << id << &#x27;n&#x27;;
}
int main()
{
	std::thread threads[10];
	// spawn 10 threads:
	for (int i = 0; i < 10; ++i)
		threads[i] = std::thread(print_thread_id, i + 1);
	for (auto& th : threads) th.join();
	return 0;
}

5. unique_lock

unique_lock特性

  • unique_lock是在C++11中提供RAII的方式管理的互斥锁资源类，相比于lock_guard他的功能更加的丰富和复杂， 网站资源

  • unique_lock, 在构造的时候就支持传入不同的tag，所以支持在够早的时候按照不同的方式来处理锁资源。

标签 特点
没有标签 构造锁对象的时候，直接使用互斥量的lock方法（阻塞式获取锁），构造完成时
try_to_lock 构造锁对象的时候调用的是try_lock的方法（非阻塞式获取锁），尝试获取锁但是不会等待；若是失败，锁对象不会持有锁
defer_lock 构造的时候不执行任何操作，同时假设互斥量当前没有被当前线程锁定，后续需要手动调方法来锁定
adopt_lock 构造时“接管”已经被当前县城接管的互斥量（假设当前的互斥量已经被当前线程接管），构造时不会再调用lock()函数 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
#include <iostream>
#include <mutex>
#include <unique_lock>

int main() {
    std::mutex mtx; // 定义一个互斥量

    // 1. 【无tag（默认）】：构造时直接阻塞式锁定
    {
        // 构造时自动调用mtx.lock()（阻塞直到拿到锁）
        std::unique_lock<std::mutex> lock1(mtx);
        std::cout << "1. 无tag：已锁定互斥量，执行操作n";
        // 出这个代码块时，lock1自动调用mtx.unlock()
    }

    // 2. 【try_to_lock】：构造时非阻塞尝试锁定
    {
        // 构造时调用mtx.try_lock()（不等待，立即返回结果）
        std::unique_lock<std::mutex> lock2(mtx, std::try_to_lock);
        if (lock2.owns_lock()) {
            std::cout << "2. try_to_lock：成功锁定互斥量n";
        } else {
            std::cout << "2. try_to_lock：锁定失败（互斥量被占用）n";
        }
        // 出块自动解锁（若成功锁定过）
    }

    // 3. 【defer_lock】：构造时不锁定，后续手动操作
    {
        // 构造时不做任何锁操作（需保证mtx当前未被本线程锁定）
        std::unique_lock<std::mutex> lock3(mtx, std::defer_lock);
        std::cout << "3. defer_lock：构造后未锁定，准备手动锁定...n";
        lock3.lock(); // 手动调用锁定（也可结合std::lock等工具）
        std::cout << "3. defer_lock：已手动锁定，执行操作n";
        // 出块自动解锁
    }

    // 4. 【adopt_lock】：接管已被本线程锁定的互斥量
    {
        mtx.lock(); // 先手动锁定互斥量（必须确保是本线程锁的）
        // 构造时接管这个已锁定的互斥量，不再调用mtx.lock()
        std::unique_lock<std::mutex> lock4(mtx, std::adopt_lock);
        std::cout << "4. adopt_lock：已接管锁定的互斥量，执行操作n";
        // 出块自动解锁
    }

    return 0;
}

6.lock和try_lock

  • lock是一个函数模板，可以支持对多个锁对象进行锁定，如果其中一个锁对象没有锁住，**lock**函数会把已经锁定的对象解锁而进入阻塞状态，知道锁定所有的对象。

  • try_lock是一个函数模板，尝试对所多个锁对象进行同时尝试锁定，如果所有所对象都已经锁定，返回-1， 如果某一个锁对象尝试锁定失败，那么已经锁定的锁对象解锁，并且返回那个对象的下标（第一个参数对象，下标从0开始）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
// std::lock example
#include <iostream> // std::cout
#include <thread> // std::thread
#include <mutex> // std::mutex, std::try_lock
std::mutex foo, bar;
void task_a() {
	bar.lock();
	std::cout << "task an";
	foo.lock();
	// ...
	foo.unlock();
	bar.unlock();
}
void task_b() {
	int x = try_lock(bar, foo);
	if (x == -1) {
		std::cout << "task bn";
		// ...
		bar.unlock();
		foo.unlock();
	}
	else {
		std::cout << "[task b failed: mutex " << (x ? "foo" : "bar")
			<< " locked]n";

		std::cout << "x : " << x << std::endl;
	}
}
int main()
{
	std::thread th1(task_a);
	std::thread th2(task_b);
	th1.join();
	th2.join();
	return 0;
}

7. call_once

  • 多线程执行的时候，让第一个线程执行fn一次，其他线程不再执行Fn

他需要配合std::once_flag（标记函数是否已经执行的共享状态）进行使用

1
std::call_once(once_flag对象, 目标函数, 函数参数...)

  • 第一个调用的线程执行目标函数，其他的线程阻塞等待执行完成后直接返回

  • 若目标函数抛出异常，本次执行就直接判定为失效，后续的线程会有一个重新尝试执行

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
#include <iostream>
#include <thread>
#include <mutex>
#include <vector>

// 共享的once_flag（标记初始化状态）
std::once_flag init_flag;
// 待初始化的全局资源
int global_config = 0;

// 仅执行一次的初始化函数
void init_global_config() {
    std::cout << "线程" << std::this_thread::get_id() << "正在初始化资源...n";
    global_config = 2026; // 模拟加载配置/初始化的操作
}

// 线程任务：所有线程都会尝试触发初始化
void thread_work() {
    // 关键：call_once保证init_global_config仅执行一次
    std::call_once(init_flag, init_global_config);
    
    // 所有线程都能使用已初始化的资源
    std::cout << "线程" << std::this_thread::get_id() << "读取到的配置：" << global_config << "n";
}

int main() {
    // 创建5个线程同时执行任务
    std::vector<std::thread> threads;
    for (int i = 0; i < 5; ++i) {
        threads.emplace_back(thread_work);
    }

    // 等待所有线程结束
    for (auto& t : threads) {
        t.join();
    }

    return 0;
}

8. atomic

  • atomic是一个模板的实例化和全特化均定义的原子类型，它可以保证对一个原子对象的操作是线程安全的

  • atomic的构造参数对于T类型的要求可用任何满足复制构造（CopyConstructible）以及可复制赋值（CopyAssignable）的可平凡复制（TriviallyCopyable）类型T来实例化，T 类型用以下几个判断时，如果一个返回的是 false，则用于atomic的操作不是线程安全的（不是原子操作）；

1
2
3
4
5
6
std::is_trivially_copyable<T>::value
std::is_copy_constructible<T>::value
std::is_move_constructible<T>::value
std::is_copy_assignable<T>::value
std::is_move_assignable<T>::value
std::is_same<T, typename std::remove_cv<T>::type>::value
  • atomic 对于简单的整形和指针支持基本的加减和位运算

  • load和store可以原子的读取和修改atomic封装的T对象

  • atomic原理主要是硬件层面的支持，现代处理器提供了原子指令来支持原子操作！！

这个就会引出一些更加有趣的概念和知识以及他具体原理，到了后面我们会详细讲解 atomic的详细知识。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
// import std; C++20
#include <vector>
#include <thread>
#include <atomic>

std::atomic<int> cnt(0); // 原子变量，初始值0

void increment() {
    // 原子写：store默认memory_order_seq_cst（强一致性）
    cnt.store(cnt.load() + 1);
}

int main() {
    std::vector<std::thread> tv;
    for (int i = 0; i < 10; i++) {
        tv.emplace_back(increment);
    }

    for (auto& t : tv) t.join();

    std::cout << cnt.load() << std::endl; // 必然输出10, 不存在竞态
    return 0;
}

9. condition_variable

condition_variable需要配合互斥锁系列进行使用，主要提供wait和notifye的系统接口

  • wait的需要传入一个unique_lock<mutex>类型的互斥锁， wait 会阻塞当前线程直到被notify 。在进入阻塞的一瞬间，会解开互斥锁，方便其他的线程取锁，访问条件变量。当notify唤醒时，他会同时获取到锁，再继续往下运行！

  • notify_alle会胡那行当前条件变量上等待的一个其中一个线程，使用他时需要使用互斥锁保护，如果没有现成的阻塞等待，他什么都不会干； notify_all是唤醒当前所有在条件变量上等待的线程。

  • condition_variable_anye类是std::condition_variable的泛化， 相对于只在std::unique_lock<std::mutex> 上工作的std::condition_variable， condition_variable_any可以在任何满足基本锁定（BasicLockable）要求的锁上工作！

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
#include <iostream> // std::cout
#include <thread> // std::thread
#include <mutex> // std::mutex, std::unique_lock
#include <condition_variable> // std::condition_variable
std::mutex mtx;
std::condition_variable cv;
bool ready = false;
void print_id(int id) {
	std::unique_lock<std::mutex> lck(mtx);
	while (!ready)
		cv.wait(lck);
	// ...
	std::cout << "thread " << id << &#x27;n&#x27;;
}
void go() {
	std::unique_lock<std::mutex> lck(mtx);
	ready = true;

	// 通知所有阻塞在条件变量上的线程 
	cv.notify_all();
}
int main()
{
	std::thread threads[10];
	// spawn 10 threads:
	for (int i = 0; i < 10; ++i)
		threads[i] = std::thread(print_id, i);
	std::cout << "10 threads ready to race...n";
	std::this_thread::sleep_for(std::chrono::milliseconds(100));
	go(); // go!
	for (auto& th : threads)
		th.join();
	return 0;
}

好啦C++11 的并发库学习就先告一段落，后面有机会还会补充相关的知识，看到这里的小伙伴都很棒呀. 🌈 😋