1
#include <thread>
2
// 1.std::thread
3
// 允许接受任意数量的参数,第一个参数为待调用的函数,后续参数是向该函数传递的参数
4
// 使用普通函数创建线程
5
void counter(int id, int numIter) {
6
    for (auto i{0}; i < numIter; ++i) {
7
        cout << "Counter " << id << "has value " << i << endl;
8
    }
9
}
10
thread t1 {counter, 1, 6};
11
thread t2 {counter, 2, 4};
12
t1.join();
13
t2.join();
14

15
// 使用functor创建线程
16
struct Counter {
17
  public:
18
    explicit Counter(int id ,int numIter) : id_(id), max_iter_(numIter) {}
19
    void operator() () const {
20
        for (auto i{0}; i < numIter; ++i)
21
          cout << "Counter " << id << "has value " << i << endl;
22
    }
23
  private:
24
    int id_;
25
    int max_iter_;
26
}
27
thread t3 {Counter{1, 20}};
28
auto c = Counter{2, 12};
29
thread t4 {c};
30

31
// 使用lambda函数创建线程
32
int id = 1;
33
int numIter = 5;
34
auto func = [id, numIter] () {
35
    for (auto i{0}; i < numIter; ++i)
36
          cout << "Counter " << id << "has value " << i << endl;
37
};
38
thread t5 {func};
39

40
// 使用类成员函数创建线程
41
class ClassWithMemberCount {
42
  public:
43
    explicit ClassWithMemberCount(int id ,int numIter) : id_(id), max_iter_(numIter) {}
44
    void count(int id ,int numIter) {
45
        for (auto i{0}; i < numIter; ++i)
46
          cout << "Counter " << id << "has value " << i << endl;
47
    }
48
  private:
49
    int id_;
50
    int max_iter_;
51
};
52
ClassWithMemberCount cls{1, 2};
53
thread t6 {&ClassWithMemberCount::count, &cls};
54

55
// 2.std::jthread
56
// 类似于std::thread,支持上述所有操作.
57
// 但在析构函数中自动调用.join()防止带崩整个程序,且支持协作式取消.
58
// 协作式取消见"线程基础操作".

1
atomic<int> aint{0};
2
int nint{0};  // normal int.
3

4
void process1() {
5
    nint = 1;
6
    aint.store(1, std::memory_order_release);
7
}
8

9
void process2() {
10
    int temp = aint.load(std::memory_order_acquire);
11
    if(temp == 1) assert(nint == 1);  // release-acquire所保证的内容.
12
}
13

14
// 注意:
15
// 无论内存序如何,对atom变量的操作始终具有原子性.
16
// release-acquire不能保证:
17
// 1. acquire指令发生在release之后(这是由线程具体运行进度所决定).
18
// 2. acquire指令先于release运行时,其它变量的有关行为.(如temp == 0时,nint的值存在竞争,可能为0也可能为1.)
19
// release-acqure只能保证:
20
// 当acquire指令确实发生在release之后时,release指令前的所有变量读写操作一定已经完成.

1
#include <atomic>
2
// 1.std::atomic_flag
3
// 1.1.初始化
4
std::atomic_flag flag = ATOMIC_FLAG_INIT;  // 初始化为clear(false)状态.
5
std::atomic_flag flag2;                    // 不定状态(C++20后为clear状态).
6
// 1.2.读写操作
7
flag.clear(mem_order);  // 置false.
8
flag.test(mem_order);  // 返回现有值.
9
flag.test_and_set(mem_order);  // 返回现有值,并置true.
10
// 1.3.同步操作
11
flag.wait(status, mem_order);  // 阻塞到flag的值不等于status,且线程被唤醒(见"条件变量"一节).
12
flag.notify_one();  // 唤醒单个线程(见"条件变量"一节).
13
flag.notify_all();  // 唤醒所有线程(见"条件变量"一节).
14

15
// 2.std::atomic<T>
16
// T必须不含cv限定,可以平凡拷贝,且具备四个移动/复制函数.
17
// 2.1.初始化操作
18
std::atomic<int> aint{0};
19
std::atomic<int> aint1 = 0;
20
// 2.2.读写操作
21
aint = 1;  // 等价于aint.store(1);
22
aint.store(1, mem_order);  // 写入值.
23
aint.load(mem_order);  // 读取值.
24
int b = aint + 1;  // 隐式转换到T.
25
// 2.3.计算操作
26
// 对整型变量,实现了fetch_add, _sub, _and位与, _or位或, _xor位异或, ++, --, +=, -=, &=, ^=, |=
27
// 对浮点变量,实现了fetch_add, _sub
28
int fetched{value.fetch_add(4)};  // fetched == 10, value == 14.
29
// 2.4.同步操作
30
aint.wait(0, mem_order);  // 类似于atomic_flag.
31
aint.notify_one();
32
aint.notify_all();
33
// 2.5.其他操作
34
aint.is_lock_free();  // 检查内部实现是否无锁.
35
aint.exchange(1, mem_order);  // 将1赋给aint,返回aint现有值.全过程为原子操作.
36
aint.compare_exchange_weak(T&expected, T desired, mem_order);    // 类似于_strong函数,性能更高,但有时aint == expected时,仍会返回false+更新expected.
37
aint.compare_exchange_strong(T&expected, T desired, mem_order);  // 原子CAS(compare and swap).
38
// aint.compare_exchange_strong与如下原子逻辑等效:
39
if (*this == expected) {
40
    *this = desired;
41
    return true;
42
} else {
43
    expected = *this;
44
    return false;
45
}
46
// 2.6.别名
47
std::atomic_int  // atomic_*对所有默认scalar量都成立.
48
// 2.7.原子引用atomic_ref
49
// 将非原子变量临时包装为原子变量,性能较差.
50
// 期间如果直接操作原变量,则原子性无法保证.
51
int inc{0};
52
atomic_ref<int> atomicCounter {counter};
53

54
// 3.自旋锁
55
// C++不提供标准实现,需自己写.
56
// 忙碌等待,只适合很短时间.
57
atomic_flag spinlock = ATOMIC_FLAG_INIT;
58
void lock() {
59
    while(spinlock.test_and_set(std::memory_order_acquire));
60
}
61
void unlock() {
62
    spinlock.clear(std::memory_order_release);
63
}
64

65
// 4.mutex/shared_mutex/timed_mutex
66
// 4.1.初始化
67
std::mutex mut;
68
// 4.2.加解锁
69
mut.lock();  // 阻塞加锁.
70
mut.try_lock();  // 非阻塞加锁,返回是否成功获得锁.
71
mut.unlock();  // 释放锁.
72
// 4.3.shared_mutex
73
// 又被称为readerwriter锁,适合读多写少的情况.
74
std::shared_mutex smut;
75
// 写锁相关:独占,只有没有任何线程持有读锁/写锁时才能获得.
76
smut.lock();
77
smut.try_lock();
78
smut.unlock();
79
// 读锁相关:非独占,多个线程可同时持有读锁.
80
smut.lock_shared();
81
smut.try_lock_shared();
82
smut.unlock_shared();
83
// 4.4.recursive_mutex
84
// 持有锁的线程可以多次加锁.
85
// 注意:unlock()次数必须与lock()一致才能解开.
86
std::recursive_mutex rmut;
87
rmut.lock();
88
rmut.try_lock();
89
rmut.unlock();
90
// 4.5.timed_mutex
91
std::timed_mutex tmut;
92
std::recursive_timed_mutex rtmut;
93
std::shared_timed_mutex stmut;
94
tmut.try_lock_for(rel_time);    // 在相对时间内尝试获得锁,返回是否成功的bool.
95
tmut.try_lock_until(abs_time);  // 在绝对时间点前尝试获得锁.

1
// 1.std::lock()与std::try_lock()
2
// 适用于需要同时获得多个锁的情况,避免死锁.
3
std::mutex mut1, mut2;
4
std::lock(mut1, mut2, ...);  // 不按指定顺序锁定互斥体对象;如果其中一个抛出异常,释放所有已获得的锁.
5
std::try_lock(mut1, mut2, ...);  // 全成功:返回-1; 失败:释放所有已获得的锁,返回未成功获得锁的序号(0开始).
6
// 2.std::scoped_lock
7
std::scoped_lock slock(mut1, mut2, ...);  // 使用std::lock()获取多个锁,在其生命周期结束自动释放.
8
// 3.lockguard
9
std::mutex mut, mut2;
10
std::lock_guard lock(mut);  // 阻塞获得锁,在作用域生命周期结束时释放锁.
11
mut2.lock();
12
std::lock_guard lock2(mut2, std::adopt_lock);  // 将已经获得的mut2交给lock2管理.
13
// 4.unique_lock
14
std::timed_mutex tmut;
15
std::unique_lock ulock(mut);  // 初始化并阻塞获得锁,在作用域结束释放锁.
16
std::unique_lock ulock(mut, std::defer_lock);   // 初始化但不取得锁.
17
std::unique_lock ulock(mut, std::try_to_lock);  // 初始化并获得锁(非阻塞),可能不会成功获得锁.
18
std::unique_lock ulock(mut, std::adopt_lock);   // 将已经获得的mut交由ulock管理.
19
std::unique_lock ulock(tmut, rel_time);         // 必须使用支持超时的锁对象.
20
std::unique_lock ulock(tmut, abs_time);
21
bool is_obtained = ulock;  // 判断锁是否成功获得.
22
ulock.release();           // 释放对内部锁的所有权(不对其进行解锁).
23
// 支持.lock(), .try_lock(), .unlock()等方法.
24
// 在作用域结束会自动解锁.
25
// 5.shared_lock
26
// 类似于unique_lock,除了获取的是shared_mutex的读锁.

1
std::once_flag flag;
2

3
void process() {
4
    std::call_once(flag, InitFunction);  // 在多个线程同时运行process,只会运行一次InitFunction.
5
}

1
#include <syncstream>
2
// 同步流中的内容在刷新缓冲区/析构时统一输出.
3
// 每个线程有独立缓冲区,不会出现覆盖现象.
4
std::osyncstream syncedCout {std::cout};  // 用于char
5
std::wosyncstream wsyncedCout;            // 用于wchar

1
// 1.用于atomic变量的条件变量 (C++20)
2
// x.wait(oldValue)
3
// 在X等于oldValue时,将持续阻塞；
4
// 其它线程使用notify_one/notify_all/发生唤醒时,将检测X是否等于oldValue;
5
// 若等于则继续阻塞,否则结束阻塞,线程继续执行.
6
int main() {
7
    std::atomic_int aint{0};
8
    std::thread job { [&aint] () {
9
    std::cout << "Job thread waiting." << std::endl;
10
    aint.wait(0);
11
    std::cout << aint << std::endl;
12
  }};
13

14
std::this_thread::sleep_for(std::chrono::seconds(1));
15
std::cout << "Notify1" << std::endl;
16
aint.notify_one();  // 子线程被唤醒,但aint仍为0,因此重新挂起,继续等待.
17
std::this_thread::sleep_for(std::chrono::milliseconds(10)); // statement 1
18
aint.store(1);
19
std::this_thread::sleep_for(std::chrono::milliseconds(10));  // statement 2
20
aint.notify_all();  // 子线程被唤醒,但aint仍为0,因此重新挂起,继续等待.
21
std::cout << "Notify2" << std::endl;
22
job.detach();
23
}
24
// 运行结果
25
// Job thread waiting.
26
// Notify1
27
// Notify2 (与下一行顺序不定)
28
// 1
29

30
// 2.用于mutex的环境变量.
31
std::condition_variable cv;       // 只能等待std::mutex的条件变量.
32
std::condition_variable_any cva;  // 可以等待多种锁类型(如shared_lock),但效率较低.
33
cv.wait(mut);  // 注意:调用前,该线程应该已经持有mut;将释放mut并在原地阻塞.
34
cv.wait_until(mut, abs_time);
35
cv.wait_for(mut, rel_time);
36
cv.wait(mut, pred);  // 带一个谓词的版本,当唤醒时,先获得锁,再检查谓词;如果为假,释放锁并再次进入等待.
37
cv.notify_one();
38
cv.notify_all();
39
// 对于以下函数,lck应该已经获得.
40
// 线程退出时将自动调用lck.unlock(); cond.notify_all();
41
// 其它线程上的cond在等待的互斥量必须与lck.mutex()一致,否则UB.
42
std::notify_all_at_thread_exit(cond, lck);
43

44
void producer() {
45
    // 准备数据...
46
    {
47
        std::unique_lock<std::mutex> lock{mut};
48
        queue.push_back(new_data);
49
    }
50
    cout << "data ready.";
51
    cv.notify_one();
52
}
53

54
void consumer() {
55
    std::unique_lock<std::mutex> lock{mut};
56
    for(;;) {
57
    cv.wait(lock, [this]{return !queue.empty();});
58
        // 处理数据...
59
    }
60
}

1
#include <latch>
2
std::latch latch{n};
3
latch.arrive_and_wait();  // 递减计数器,阻塞当前线程.
4
latch.wait();        // 不递减计数器,阻塞线程.
5
latch.try_wait();      // 检测计数器是否已经为0.
6
latch.countdown();      // 减少计数器(不阻塞).
7

8
// 示例程序
9
std::latch latch{1};
10
vector<jthread> threads;
11
for(auto i = 0; i < 10; ++i) {
12
    threads.push_back(jthread{[&latch](){
13
        // 运行slave线程初始化操作...
14
        latch.wait();  // 统一阻塞在latch处.
15
        // 正式的处理过程...
16
    }});
17
}
18
// master线程加载数据...
19
latch.countdown();  // 递减latch计数器,开始运行slave线程.

1
#include <barrier>
2
void completionFunction() noexcept {...}
3
int main() {
4
    const size_t kNThreads = 4u;
5
    barrier barrierPoint{kThreads, conpletionFunction};
6
    vector<jthread> threads;
7

8
    for(int i{0}; i < kNThreads; ++i) {
9
        threads.push_back(jthread{
10
            [&barrierPoint](std::stop_token token){
11
                while(!token.stop_requested()) {
12
                    // 进行计算...
13
                    barrierPoint.arrive_and_wait();
14
                }
15
            }
16
        });
17
  }
18
}

1
#include <semaphore>
2
std::counting_semaphore cs{4};  // 多槽信号量
3
std::binary_semaphore bs;
4

5
cs.acquire();  // 递减计数器,若计数器为0时阻塞,直到有其他线程释放.
6
cs.try_acquire();  // 非阻塞获取信号量,未获得时返回false.
7
cs.try_acquire_for();  // 在一定时间内尝试获取.
8
cs.try_acquire_until();
9
cs.release();  // 释放信号量.

1
// 1. promise-future组合
2
// promise不可复制,但可移动.
3
promise<T> p;
4
p.get_future();  // 获取对应的future,只允许调用一次.
5
p.set_value();  // 设定返回值类型,只允许调用一次.
6
p.set_value_at_thread_exit();
7
p.set_exception();
8
p.set_exception_at_thread_exit();
9

10
auto f{p.get_future()};  // f类型为future<T>
11
f.get();  // 从future获取返回值,只能调用一次,否则抛异常.如果结果不可用会阻塞.
12
f.valid();  // 检查future是否为共享态,只有共享态才能调用get();调用get()后会退出共享态.
13
f.wait();  // 阻塞到结果可用.
14
f.wait_for();  // 阻塞到结果可用/达到给定时间.
15
f.wait_until();
16

17
void WorkFunction(promise<int> p) {
18
    // 实际计算...
19
    p.set_value(42);
20
}
21

22
int main() {
23
    promise<int> p;
24
    auto f {p.get_future()};
25
    jthread t{workFunction, move(p)};
26

27
    while(f.wait_for(0)) {  // 检查结果是否计算完成.
28
        auto res {f.get()};
29
        cout << res;
30
    } else {
31
        // 执行结果未完成时的内容...
32
    }
33
}
34

35
// 2. packaged_task
36
// 进一步封装,无需手动传递promise.
37
// 工作函数结束时,自动调用promise.set_value(return_of_work_function).
38
// 同样不可复制,只能移动.
39
int main() {
40
    pacakaged_task<int(void)> pt {[](){return 42;}};
41
    auto f {pt.get_future()};
42
    jthread t{move(pt)};
43

44
    while(f.wait_for(0)) {  // 检查结果是否计算完成.
45
        auto res {f.get()};
46
        cout << res;
47
    } else {
48
        // 执行结果未完成时的内容...
49
    }
50
}
51

52
// 3. async
53
// 进一步封装,无需手动创建进程.
54
// 注意:未捕获async返回值时,其返回的future临时对象将立即析构,
55
// 因此会使父线程阻塞直到子线程完成.
56
int workFunction() {return 42};
57
int main() {
58
    // launch::async:强制创建新线程运行
59
    // launch::deferred:强制在get()时使用当前进程运行
60
    // launch::async | launch::deferred:自动选择(默认行为)
61
  auto f {async(launch::async, workFunction)};
62

63
  while(f.wait_for(0)) {  // 检查结果是否计算完成.
64
        auto res {f.get()};
65
        cout << res;
66
    } else {
67
        // 执行结果未完成时的内容...
68
    }
69
}
70

71
// 4. shared_future
72
// 允许多个进程同时阻塞在get()函数.
73
// 返回类型T必须可以复制.
74
promise<T> p;
75
shared_future<T> sf{p.get_future()};  // 该初始化只接受右值对象.
76
sf.valid();  // 在get()后不会进入非共享态,因此结果可以由多个线程取得.
77

78
// 可起到类似于条件变量的作用.
79
int main()
80
{
81
    std::promise<void> ready_promise, t1_ready_promise, t2_ready_promise;
82
    std::shared_future<void> ready_future(ready_promise.get_future());
83

84
    std::chrono::time_point<std::chrono::high_resolution_clock> start;
85

86
    auto fun1 = [&, ready_future]() -> std::chrono::duration<double, std::milli>
87
    {
88
        t1_ready_promise.set_value();
89
        ready_future.wait(); // waits for the signal from main()
90
        return std::chrono::high_resolution_clock::now() - start;
91
    };
92

93

94
    auto fun2 = [&, ready_future]() -> std::chrono::duration<double, std::milli>
95
    {
96
        t2_ready_promise.set_value();
97
        ready_future.wait(); // waits for the signal from main()
98
        return std::chrono::high_resolution_clock::now() - start;
99
    };
100

101
    auto fut1 = t1_ready_promise.get_future();
102
    auto fut2 = t2_ready_promise.get_future();
103

104
    auto result1 = std::async(std::launch::async, fun1);
105
    auto result2 = std::async(std::launch::async, fun2);
106

107
    // wait for the threads to become ready
108
    fut1.wait();
109
    fut2.wait();
110

111
    // the threads are ready, start the clock
112
    start = std::chrono::high_resolution_clock::now();
113

114
    // signal the threads to go
115
    ready_promise.set_value();
116

117
    std::cout << "Thread 1 received the signal "
118
              << result1.get().count() << " ms after start\n"
119
              << "Thread 2 received the signal "
120
              << result2.get().count() << " ms after start\n";
121
}