Computer Science

WangMC Blog

C++并发编程核心知识体系

发表于 更新于 分类于
本文字数： 60k 阅读时长 ≈ 55 分钟

如题，结合官方文档与网络帖子，利用AI整理

C++ 并发编程核心知识体系：从技术实现到哲学思想

目录

  1. 并发编程概述

  2. C++ 线程基础：std::thread

  3. 互斥锁机制：std::mutex 家族

  4. 条件变量：线程间通信的艺术

  5. 并发设计模式

  6. 内存模型与原子操作

  7. 异步操作：非阻塞编程范式

  8. 并发编程哲学思想

  9. 最佳实践与设计原则

  10. 常见问题与解决方案

1. 并发编程概述

1.1 什么是并发编程

并发编程是指在同一时间间隔内执行多个任务的编程范式。在 C++ 中，这主要通过多线程来实现，每个线程代表一个独立的执行路径。

1.2 并发编程的核心挑战

  • 数据竞争：多个线程同时访问共享数据

  • 死锁：线程间相互等待对方释放资源

  • 活锁：线程不断改变状态但无法继续执行

  • 内存可见性：CPU 缓存导致的操作不可见问题

  • 指令重排序：编译器和 CPU 的优化导致执行顺序变化

1.3 C++ 并发编程的发展历程

  • C++11：引入了标准的并发编程支持

  • C++14：完善了并发库功能

  • C++17：新增了共享锁等特性

  • C++20：引入了协程等高级特性

2. C++ 线程基础：std::thread

2.1 线程的创建与管理

基本线程创建

1
2
3
4
5
6
7
8
9
10
11
12
#include <iostream>
#include <thread>

void thread_function() {
    std::cout << "Hello from thread!" << std::endl;
}

int main() {
    std::thread t(thread_function);  // 创建线程
    t.join();                        // 等待线程完成
    return 0;
}

传递参数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
#include <iostream>
#include <thread>
#include <string>

void print_message(const std::string& message, int count) {
    for (int i = 0; i < count; ++i) {
        std::cout << message << " " << i << std::endl;
    }
}

int main() {
    std::string msg = "Hello";
    std::thread t(print_message, msg, 5);
    t.join();
    return 0;
}

2.2 线程的生命周期管理

join() vs detach()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
#include <iostream>
#include <thread>
#include <chrono>

void long_running_task() {
    std::this_thread::sleep_for(std::chrono::seconds(3));
    std::cout << "Task completed!" << std::endl;
}

int main() {
    std::thread t(long_running_task);
    
    // 使用join()：主线程等待子线程完成
    // t.join();
    
    // 使用detach()：子线程在后台运行
    t.detach();
    
    std::cout << "Main thread exiting..." << std::endl;
    return 0;
}

2.3 线程属性控制

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
#include <iostream>
#include <thread>
#include <chrono>

void thread_with_high_priority() {
    // 在支持的平台上设置线程优先级
    #ifdef _WIN32
        // Windows平台设置优先级
        SetThreadPriority(GetCurrentThread(), THREAD_PRIORITY_HIGHEST);
    #elif defined(__linux__)
        // Linux平台设置优先级
        struct sched_param param;
        param.sched_priority = 99;
        pthread_setschedparam(pthread_self(), SCHED_FIFO, &param);
    #endif
    
    std::cout << "High priority thread running..." << std::endl;
}

int main() {
    std::thread t(thread_with_high_priority);
    t.join();
    return 0;
}

3. 互斥锁机制：std::mutex 家族

3.1 互斥锁的基本概念

互斥锁（Mutex）是保证线程间互斥访问共享资源的同步原语。

3.2 各种互斥锁类型

std::mutex - 基本互斥锁

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
#include <iostream>
#include <thread>
#include <mutex>
#include <vector>

std::mutex mtx;
int shared_counter = 0;

void increment_counter() {
    for (int i = 0; i < 100000; ++i) {
        mtx.lock();
        shared_counter++;
        mtx.unlock();
    }
}

int main() {
    std::vector<std::thread> threads;
    
    for (int i = 0; i < 10; ++i) {
        threads.emplace_back(increment_counter);
    }
    
    for (auto& t : threads) {
        t.join();
    }
    
    std::cout << "Final counter value: " << shared_counter << std::endl;
    return 0;
}

std::lock_guard - 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
#include <iostream>
#include <thread>
#include <mutex>
#include <vector>

std::mutex mtx;
int shared_counter = 0;

void increment_counter() {
    for (int i = 0; i < 100000; ++i) {
        std::lock_guard<std::mutex> lock(mtx);
        shared_counter++;
        // 锁会在离开作用域时自动释放
    }
}

int main() {
    std::vector<std::thread> threads;
    
    for (int i = 0; i < 10; ++i) {
        threads.emplace_back(increment_counter);
    }
    
    for (auto& t : threads) {
        t.join();
    }
    
    std::cout << "Final counter value: " << shared_counter << std::endl;
    return 0;
}

std::unique_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
#include <iostream>
#include <thread>
#include <mutex>
#include <vector>

std::mutex mtx;
int shared_counter = 0;

void increment_counter() {
    for (int i = 0; i < 100000; ++i) {
        std::unique_lock<std::mutex> lock(mtx);
        shared_counter++;
        
        // 可以手动解锁和重新锁定
        lock.unlock();
        // 执行一些不需要锁的操作
        lock.lock();
        shared_counter++;
    }
}

int main() {
    std::vector<std::thread> threads;
    
    for (int i = 0; i < 10; ++i) {
        threads.emplace_back(increment_counter);
    }
    
    for (auto& t : threads) {
        t.join();
    }
    
    std::cout << "Final counter value: " << shared_counter << std::endl;
    return 0;
}

C++17 共享锁：std::shared_mutex

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
#include <iostream>
#include <thread>
#include <shared_mutex>
#include <vector>

std::shared_mutex rw_mutex;
int shared_data = 0;

// 读操作：可以多个线程同时进行
void reader_thread(int id) {
    std::shared_lock<std::shared_mutex> lock(rw_mutex);
    std::cout << "Reader " << id << " read data: " << shared_data << std::endl;
}

// 写操作：独占访问
void writer_thread(int id) {
    std::unique_lock<std::shared_mutex> lock(rw_mutex);
    shared_data++;
    std::cout << "Writer " << id << " wrote data: " << shared_data << std::endl;
}

int main() {
    std::vector<std::thread> threads;
    
    // 创建多个读者线程
    for (int i = 0; i < 5; ++i) {
        threads.emplace_back(reader_thread, i);
    }
    
    // 创建写者线程
    for (int i = 0; i < 2; ++i) {
        threads.emplace_back(writer_thread, i);
    }
    
    for (auto& t : threads) {
        t.join();
    }
    
    return 0;
}

3.3 避免死锁的技术

1. 统一锁获取顺序

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
#include <iostream>
#include <thread>
#include <mutex>

std::mutex mutex1, mutex2;

void thread1() {
    std::lock(mutex1, mutex2);  // 同时获取多个锁
    std::lock_guard<std::mutex> lock1(mutex1, std::adopt_lock);
    std::lock_guard<std::mutex> lock2(mutex2, std::adopt_lock);
    
    std::cout << "Thread 1 acquired both mutexes" << std::endl;
}

void thread2() {
    std::lock(mutex1, mutex2);  // 保持相同的获取顺序
    std::lock_guard<std::mutex> lock1(mutex1, std::adopt_lock);
    std::lock_guard<std::mutex> lock2(mutex2, std::adopt_lock);
    
    std::cout << "Thread 2 acquired both mutexes" << std::endl;
}

int main() {
    std::thread t1(thread1);
    std::thread t2(thread2);
    
    t1.join();
    t2.join();
    
    return 0;
}

2. 使用 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
32
33
#include <iostream>
#include <thread>
#include <mutex>
#include <chrono>

std::mutex mutex1, mutex2;

void thread1() {
    while (true) {
        if (mutex1.try_lock()) {
            std::cout << "Thread 1 acquired mutex1" << std::endl;
            
            if (mutex2.try_lock()) {
                std::cout << "Thread 1 acquired mutex2" << std::endl;
                // 执行操作
                mutex2.unlock();
                mutex1.unlock();
                break;
            } else {
                mutex1.unlock();
                std::this_thread::sleep_for(std::chrono::milliseconds(10));
            }
        } else {
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
        }
    }
}

int main() {
    std::thread t1(thread1);
    t1.join();
    return 0;
}

4. 条件变量：线程间通信的艺术

4.1 条件变量的基本概念

条件变量（Condition Variable）用于线程间的通信，允许一个线程等待另一个线程满足某个条件。

4.2 std::condition_variable 的使用

生产者 - 消费者模式

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
56
57
58
59
60
61
62
63
64
65
66
#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <queue>
#include <chrono>

std::mutex mtx;
std::condition_variable cv;
std::queue<int> data_queue;
bool stop_flag = false;

// 生产者线程
void producer() {
    for (int i = 0; i < 10; ++i) {
        {
            std::lock_guard<std::mutex> lock(mtx);
            data_queue.push(i);
            std::cout << "Produced: " << i << std::endl;
        }
        cv.notify_one();  // 通知消费者
        std::this_thread::sleep_for(std::chrono::milliseconds(500));
    }
    
    // 通知消费者停止
    {
        std::lock_guard<std::mutex> lock(mtx);
        stop_flag = true;
    }
    cv.notify_one();
}

// 消费者线程
void consumer() {
    while (true) {
        std::unique_lock<std::mutex> lock(mtx);
        
        // 等待条件：队列不为空或停止标志为true
        cv.wait(lock, []{ 
            return !data_queue.empty() || stop_flag; 
        });
        
        // 检查是否需要停止
        if (stop_flag && data_queue.empty()) {
            break;
        }
        
        // 处理数据
        if (!data_queue.empty()) {
            int data = data_queue.front();
            data_queue.pop();
            std::cout << "Consumed: " << data << std::endl;
        }
    }
    std::cout << "Consumer stopped" << std::endl;
}

int main() {
    std::thread prod_thread(producer);
    std::thread cons_thread(consumer);
    
    prod_thread.join();
    cons_thread.join();
    
    return 0;
}

4.3 条件变量的高级用法

多个条件变量的协调

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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <queue>
#include <chrono>

std::mutex mtx;
std::condition_variable cv_producer;
std::condition_variable cv_consumer;
std::queue<int> data_queue;
const int MAX_QUEUE_SIZE = 5;
bool stop_flag = false;

void producer() {
    for (int i = 0; i < 10; ++i) {
        std::unique_lock<std::mutex> lock(mtx);
        
        // 等待队列有空间
        cv_producer.wait(lock, []{
            return data_queue.size() < MAX_QUEUE_SIZE || stop_flag;
        });
        
        if (stop_flag) break;
        
        data_queue.push(i);
        std::cout << "Produced: " << i << ", Queue size: " << data_queue.size() << std::endl;
        
        cv_consumer.notify_one();  // 通知消费者
    }
    
    // 通知消费者停止
    {
        std::lock_guard<std::mutex> lock(mtx);
        stop_flag = true;
    }
    cv_consumer.notify_all();
}

void consumer() {
    while (true) {
        std::unique_lock<std::mutex> lock(mtx);
        
        // 等待队列有数据
        cv_consumer.wait(lock, []{
            return !data_queue.empty() || stop_flag;
        });
        
        if (stop_flag && data_queue.empty()) {
            break;
        }
        
        if (!data_queue.empty()) {
            int data = data_queue.front();
            data_queue.pop();
            std::cout << "Consumed: " << data << ", Queue size: " << data_queue.size() << std::endl;
            
            cv_producer.notify_one();  // 通知生产者队列有空间
        }
    }
}

int main() {
    std::thread prod_thread(producer);
    std::thread cons_thread1(consumer);
    std::thread cons_thread2(consumer);
    
    prod_thread.join();
    cons_thread1.join();
    cons_thread2.join();
    
    return 0;
}

5. 并发设计模式

5.1 Fork-Join 模式

Fork-Join 模式将大任务分解为多个小任务并行执行，然后合并结果。

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
56
57
58
59
#include <iostream>
#include <thread>
#include <vector>
#include <algorithm>
#include <numeric>

template<typename Iterator, typename T>
struct accumulate_block {
    void operator()(Iterator first, Iterator last, T& result) {
        result = std::accumulate(first, last, result);
    }
};

template<typename Iterator, typename T>
T parallel_accumulate(Iterator first, Iterator last, T init) {
    unsigned long const length = std::distance(first, last);
    
    if (!length)
        return init;
    
    unsigned long const min_per_thread = 25;
    unsigned long const max_threads = (length + min_per_thread - 1) / min_per_thread;
    unsigned long const hardware_threads = std::thread::hardware_concurrency();
    unsigned long const num_threads = std::min(hardware_threads != 0 ? hardware_threads : 2, max_threads);
    unsigned long const block_size = length / num_threads;
    
    std::vector<T> results(num_threads);
    std::vector<std::thread> threads(num_threads - 1);
    
    Iterator block_start = first;
    for (unsigned long i = 0; i < (num_threads - 1); ++i) {
        Iterator block_end = block_start;
        std::advance(block_end, block_size);
        threads[i] = std::thread(
            accumulate_block<Iterator, T>(),
            block_start, block_end, std::ref(results[i])
        );
        block_start = block_end;
    }
    
    accumulate_block<Iterator, T>()(block_start, last, results[num_threads - 1]);
    
    for (auto& entry : threads) {
        entry.join();
    }
    
    return std::accumulate(results.begin(), results.end(), init);
}

int main() {
    std::vector<int> numbers(1000000, 1);
    
    int result = parallel_accumulate(numbers.begin(), numbers.end(), 0);
    
    std::cout << "Parallel accumulate result: " << result << std::endl;
    std::cout << "Expected result: " << numbers.size() << std::endl;
    
    return 0;
}

5.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
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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
#include <iostream>
#include <thread>
#include <vector>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <future>
#include <stdexcept>

class ThreadPool {
public:
    ThreadPool(size_t threads);
    
    template<class F, class... Args>
    auto enqueue(F&& f, Args&&... args) 
        -> std::future<typename std::result_of<F(Args...)>::type>;
    
    ~ThreadPool();
    
private:
    std::vector<std::thread> workers;
    std::queue<std::function<void()>> tasks;
    
    std::mutex queue_mutex;
    std::condition_variable condition;
    bool stop;
};

ThreadPool::ThreadPool(size_t threads) 
    : stop(false) {
    
    for (size_t i = 0; i < threads; ++i) {
        workers.emplace_back([this] {
            for (;;) {
                std::function<void()> task;
                
                {
                    std::unique_lock<std::mutex> lock(this->queue_mutex);
                    this->condition.wait(lock, [this] {
                        return this->stop || !this->tasks.empty();
                    });
                    
                    if (this->stop && this->tasks.empty())
                        return;
                    
                    task = std::move(this->tasks.front());
                    this->tasks.pop();
                }
                
                task();
            }
        });
    }
}

template<class F, class... Args>
auto ThreadPool::enqueue(F&& f, Args&&... args) 
    -> std::future<typename std::result_of<F(Args...)>::type> {
    
    using return_type = typename std::result_of<F(Args...)>::type;
    
    auto task = std::make_shared<std::packaged_task<return_type()>>(
        std::bind(std::forward<F>(f), std::forward<Args>(args)...)
    );
    
    std::future<return_type> res = task->get_future();
    
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        
        if (stop)
            throw std::runtime_error("enqueue on stopped ThreadPool");
        
        tasks.emplace([task]() { (*task)(); });
    }
    
    condition.notify_one();
    return res;
}

ThreadPool::~ThreadPool() {
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        stop = true;
    }
    
    condition.notify_all();
    
    for (std::thread& worker : workers)
        worker.join();
}

// 使用示例
int calculate_square(int x) {
    return x * x;
}

int main() {
    ThreadPool pool(4);
    std::vector<std::future<int>> results;
    
    for (int i = 0; i < 8; ++i) {
        results.emplace_back(
            pool.enqueue(calculate_square, i)
        );
    }
    
    for (auto&& result : results) {
        std::cout << result.get() << " ";
    }
    std::cout << std::endl;
    
    return 0;
}

5.3 读写锁模式

读写锁允许多个读者同时访问，但写者需要独占访问。

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
56
57
58
59
60
61
62
63
64
65
#include <iostream>
#include <thread>
#include <shared_mutex>
#include <vector>
#include <chrono>

class SharedData {
private:
    int data_;
    mutable std::shared_mutex mutex_;
    
public:
    SharedData(int initial_data) : data_(initial_data) {}
    
    // 读操作：共享访问
    int read_data() const {
        std::shared_lock<std::shared_mutex> lock(mutex_);
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
        return data_;
    }
    
    // 写操作：独占访问
    void write_data(int new_value) {
        std::unique_lock<std::shared_mutex> lock(mutex_);
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
        data_ = new_value;
    }
};

void reader_thread(SharedData& data, int id) {
    for (int i = 0; i < 5; ++i) {
        int value = data.read_data();
        std::cout << "Reader " << id << " read: " << value << std::endl;
        std::this_thread::sleep_for(std::chrono::milliseconds(50));
    }
}

void writer_thread(SharedData& data, int id, int start_value) {
    for (int i = 0; i < 3; ++i) {
        int new_value = start_value + i;
        data.write_data(new_value);
        std::cout << "Writer " << id << " wrote: " << new_value << std::endl;
        std::this_thread::sleep_for(std::chrono::milliseconds(200));
    }
}

int main() {
    SharedData data(0);
    std::vector<std::thread> threads;
    
    // 创建5个读者线程
    for (int i = 0; i < 5; ++i) {
        threads.emplace_back(reader_thread, std::ref(data), i);
    }
    
    // 创建2个写者线程
    threads.emplace_back(writer_thread, std::ref(data), 0, 100);
    threads.emplace_back(writer_thread, std::ref(data), 1, 200);
    
    for (auto& t : threads) {
        t.join();
    }
    
    return 0;
}

6. 内存模型与原子操作

6.1 C++ 内存模型基础

C++ 内存模型定义了多线程环境下内存操作的可见性和顺序性规则。

6.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
28
29
30
31
#include <iostream>
#include <thread>
#include <atomic>
#include <vector>

std::atomic<int> atomic_counter(0);
int normal_counter = 0;

void increment_counters() {
    for (int i = 0; i < 100000; ++i) {
        atomic_counter++;
        normal_counter++;
    }
}

int main() {
    std::vector<std::thread> threads;
    
    for (int i = 0; i < 10; ++i) {
        threads.emplace_back(increment_counters);
    }
    
    for (auto& t : threads) {
        t.join();
    }
    
    std::cout << "Atomic counter: " << atomic_counter << std::endl;
    std::cout << "Normal counter: " << normal_counter << std::endl;
    
    return 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
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
#include <iostream>
#include <thread>
#include <atomic>
#include <vector>

std::atomic<bool> x(false), y(false);
std::atomic<int> z(0);

void write_x() {
    x.store(true, std::memory_order_release);
}

void write_y() {
    y.store(true, std::memory_order_release);
}

void read_x_then_y() {
    while (!x.load(std::memory_order_acquire));
    if (y.load(std::memory_order_acquire)) {
        z++;
    }
}

void read_y_then_x() {
    while (!y.load(std::memory_order_acquire));
    if (x.load(std::memory_order_acquire)) {
        z++;
    }
}

int main() {
    for (int i = 0; i < 10000; ++i) {
        x = false;
        y = false;
        z = 0;
        
        std::thread a(write_x);
        std::thread b(write_y);
        std::thread c(read_x_then_y);
        std::thread d(read_y_then_x);
        
        a.join();
        b.join();
        c.join();
        d.join();
        
        if (z.load() == 0) {
            std::cout << "z is 0 after " << i + 1 << " iterations" << std::endl;
        }
    }
    
    return 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
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
#include <iostream>
#include <thread>
#include <atomic>
#include <vector>

template<typename T>
class LockFreeStack {
private:
    struct Node {
        T data;
        Node* next;
        
        Node(const T& data) : data(data), next(nullptr) {}
    };
    
    std::atomic<Node*> head;
    
public:
    LockFreeStack() : head(nullptr) {}
    
    ~LockFreeStack() {
        while (Node* old_head = head.load()) {
            head.store(old_head->next);
            delete old_head;
        }
    }
    
    void push(const T& data) {
        Node* new_node = new Node(data);
        new_node->next = head.load();
        
        // 使用compare_exchange_weak处理并发
        while (!head.compare_exchange_weak(new_node->next, new_node)) {
            // 重试直到成功
        }
    }
    
    bool pop(T& result) {
        Node* old_head = head.load();
        
        while (old_head && !head.compare_exchange_weak(old_head, old_head->next)) {
            // 重试直到成功或栈为空
        }
        
        if (!old_head) {
            return false;
        }
        
        result = old_head->data;
        delete old_head;
        return true;
    }
    
    bool empty() const {
        return head.load() == nullptr;
    }
};

void push_to_stack(LockFreeStack<int>& stack, int start, int end) {
    for (int i = start; i < end; ++i) {
        stack.push(i);
    }
}

void pop_from_stack(LockFreeStack<int>& stack, int count, std::vector<int>& results) {
    int value;
    for (int i = 0; i < count; ++i) {
        if (stack.pop(value)) {
            results.push_back(value);
        }
    }
}

int main() {
    LockFreeStack<int> stack;
    std::vector<int> results1, results2;
    
    std::thread pusher1(push_to_stack, std::ref(stack), 0, 1000);
    std::thread pusher2(push_to_stack, std::ref(stack), 1000, 2000);
    std::thread popper1(pop_from_stack, std::ref(stack), 1000, std::ref(results1));
    std::thread popper2(pop_from_stack, std::ref(stack), 1000, std::ref(results2));
    
    pusher1.join();
    pusher2.join();
    popper1.join();
    popper2.join();
    
    std::cout << "Pushed 2000 elements" << std::endl;
    std::cout << "Popped " << results1.size() + results2.size() << " elements" << std::endl;
    
    return 0;
}

7. 异步操作：非阻塞编程范式

7.1 异步编程概述

7.1.1 什么是异步编程

异步编程是一种编程范式，它允许程序在等待操作完成（如网络请求、文件 I/O、数据库查询等）时继续执行其他任务。与同步编程不同，异步操作不会阻塞当前执行流程。

7.1.2 同步 vs 异步的本质区别

特性 同步 (Synchronous) 异步 (Asynchronous)
执行方式 顺序执行，一步接一步 并发执行，无需等待
等待行为 阻塞等待结果返回 非阻塞，立即返回
资源利用 低效（CPU 常闲置等待） 高效（CPU 持续工作）
编程模型 直线式思维，易于理解 事件驱动 / 回调思维
复杂度 简单直接 较复杂（回调地狱风险）
适用场景 CPU 密集型任务、简单脚本 I/O 密集型、高并发服务

7.1.3 生活中的异步哲学

1
2
同步场景：在快餐店排队点餐 - 必须等前一个人完成才能点餐
异步场景：在正餐厅桌边点餐 - 点餐后可以做其他事，服务员会送餐

7.2 C++ 异步编程组件

7.2.1 std::future - 异步结果的占位符

std::future 表示一个异步操作的结果，可以在未来某个时间点获取。

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
#include <iostream>
#include <future>
#include <chrono>

// 模拟耗时计算
int expensive_calculation(int x, int y) {
    std::this_thread::sleep_for(std::chrono::seconds(2));
    return x * y;
}

int main() {
    std::cout << "Main thread started" << std::endl;
    
    // 启动异步任务
    std::future<int> result_future = std::async(expensive_calculation, 10, 20);
    
    // 主线程可以继续执行其他任务
    std::cout << "Main thread is doing other work..." << std::endl;
    for (int i = 0; i < 5; ++i) {
        std::cout << "Main: " << i << std::endl;
        std::this_thread::sleep_for(std::chrono::milliseconds(300));
    }
    
    // 获取异步任务结果（如果未完成会阻塞）
    std::cout << "Waiting for async result..." << std::endl;
    int result = result_future.get();
    std::cout << "Async result: " << result << std::endl;
    
    return 0;
}

7.2.2 std::async - 异步任务启动器

std::async 是启动异步任务的便捷函数，返回一个 std::future 对象。

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
#include <iostream>
#include <future>
#include <vector>

// 并行计算示例
int parallel_sum(const std::vector<int>& data, int start, int end) {
    int sum = 0;
    for (int i = start; i < end; ++i) {
        sum += data[i];
    }
    return sum;
}

int main() {
    std::vector<int> large_data(1000000, 1);
    
    // 将数据分成4个部分并行计算
    int mid1 = large_data.size() / 4;
    int mid2 = large_data.size() / 2;
    int mid3 = 3 * large_data.size() / 4;
    
    std::future<int> f1 = std::async(parallel_sum, std::ref(large_data), 0, mid1);
    std::future<int> f2 = std::async(parallel_sum, std::ref(large_data), mid1, mid2);
    std::future<int> f3 = std::async(parallel_sum, std::ref(large_data), mid2, mid3);
    std::future<int> f4 = std::async(parallel_sum, std::ref(large_data), mid3, large_data.size());
    
    // 汇总结果
    int total = f1.get() + f2.get() + f3.get() + f4.get();
    std::cout << "Total sum: " << total << std::endl;
    
    return 0;
}

7.2.3 std::promise - 主动设置异步结果

std::promise 允许一个线程设置结果，另一个线程通过 std::future 获取结果。

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
#include <iostream>
#include <future>
#include <thread>
#include <chrono>

void worker_thread(std::promise<int>& prom) {
    try {
        std::cout << "Worker thread started" << std::endl;
        std::this_thread::sleep_for(std::chrono::seconds(2));
        
        // 模拟计算
        int result = 42;
        
        // 设置结果
        prom.set_value(result);
        std::cout << "Worker thread finished" << std::endl;
    } catch (...) {
        // 设置异常
        prom.set_exception(std::current_exception());
    }
}

int main() {
    std::promise<int> prom;
    std::future<int> fut = prom.get_future();
    
    // 启动工作线程
    std::thread t(worker_thread, std::ref(prom));
    
    std::cout << "Main thread waiting for result..." << std::endl;
    
    // 获取结果
    try {
        int result = fut.get();
        std::cout << "Result from worker: " << result << std::endl;
    } catch (const std::exception& e) {
        std::cerr << "Exception from worker: " << e.what() << std::endl;
    }
    
    t.join();
    return 0;
}

7.2.4 std::packaged_task - 任务包装器

std::packaged_task 将可调用对象包装起来，方便作为线程函数使用。

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
56
57
58
#include <iostream>
#include <future>
#include <thread>
#include <vector>

// 矩阵乘法的一部分
void matrix_multiply_part(const std::vector<std::vector<int>>& A,
                          const std::vector<std::vector<int>>& B,
                          std::vector<std::vector<int>>& C,
                          int start_row, int end_row) {
    int cols_B = B[0].size();
    int cols_A = A[0].size();
    
    for (int i = start_row; i < end_row; ++i) {
        for (int j = 0; j < cols_B; ++j) {
            C[i][j] = 0;
            for (int k = 0; k < cols_A; ++k) {
                C[i][j] += A[i][k] * B[k][j];
            }
        }
    }
}

int main() {
    const int size = 100;
    std::vector<std::vector<int>> A(size, std::vector<int>(size, 1));
    std::vector<std::vector<int>> B(size, std::vector<int>(size, 2));
    std::vector<std::vector<int>> C(size, std::vector<int>(size, 0));
    
    int num_threads = std::thread::hardware_concurrency();
    int rows_per_thread = size / num_threads;
    
    std::vector<std::future<void>> futures;
    
    for (int i = 0; i < num_threads; ++i) {
        int start = i * rows_per_thread;
        int end = (i == num_threads - 1) ? size : (i + 1) * rows_per_thread;
        
        // 使用packaged_task包装任务
        std::packaged_task<void()> task([&, start, end]() {
            matrix_multiply_part(A, B, C, start, end);
        });
        
        futures.push_back(task.get_future());
        
        // 启动线程执行任务
        std::thread t(std::move(task));
        t.detach();
    }
    
    // 等待所有任务完成
    for (auto& fut : futures) {
        fut.get();
    }
    
    std::cout << "Matrix multiplication completed" << std::endl;
    return 0;
}

7.3 异步编程的设计原理

7.3.1 异步编程的核心原理

1. 事件驱动模型

异步编程基于事件驱动模型，程序通过响应事件来处理任务完成。

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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
#include <iostream>
#include <future>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <functional>

class EventLoop {
private:
    std::queue<std::function<void()>> events_;
    std::mutex mtx_;
    std::condition_variable cv_;
    bool running_;
    
public:
    EventLoop() : running_(true) {}
    
    void post(std::function<void()> event) {
        std::lock_guard<std::mutex> lock(mtx_);
        events_.push(event);
        cv_.notify_one();
    }
    
    void run() {
        while (running_) {
            std::unique_lock<std::mutex> lock(mtx_);
            cv_.wait(lock, [this]{ return !events_.empty() || !running_; });
            
            if (!running_) break;
            
            auto event = events_.front();
            events_.pop();
            lock.unlock();
            
            event();
        }
    }
    
    void stop() {
        std::lock_guard<std::mutex> lock(mtx_);
        running_ = false;
        cv_.notify_one();
    }
};

// 使用事件循环的异步任务
void async_task(EventLoop& loop, int id) {
    std::cout << "Async task " << id << " started" << std::endl;
    
    // 模拟异步操作
    std::thread([&loop, id]() {
        std::this_thread::sleep_for(std::chrono::seconds(1));
        
        // 任务完成后发布事件
        loop.post([id]() {
            std::cout << "Async task " << id << " completed" << std::endl;
        });
    }).detach();
}

int main() {
    EventLoop loop;
    
    // 启动事件循环线程
    std::thread loop_thread([&loop]() {
        loop.run();
    });
    
    // 提交多个异步任务
    for (int i = 0; i < 5; ++i) {
        async_task(loop, i);
        std::this_thread::sleep_for(std::chrono::milliseconds(200));
    }
    
    // 等待所有任务完成
    std::this_thread::sleep_for(std::chrono::seconds(2));
    
    loop.stop();
    loop_thread.join();
    
    return 0;
}

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
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
#include <iostream>
#include <future>
#include <functional>
#include <chrono>

// 异步文件读取模拟
void async_file_read(const std::string& filename, 
                    std::function<void(const std::string&)> callback) {
    std::thread([filename, callback]() {
        std::this_thread::sleep_for(std::chrono::seconds(1));
        
        // 模拟文件内容
        std::string content = "File content of " + filename;
        
        // 调用回调函数
        callback(content);
    }).detach();
}

// 链式异步操作
void process_data_chain() {
    // 第一步：读取文件
    async_file_read("data.txt", [](const std::string& content) {
        std::cout << "Step 1: File read completed" << std::endl;
        
        // 第二步：处理数据
        std::string processed = content + " (processed)";
        
        // 第三步：保存结果
        async_file_read("result.txt", [processed](const std::string&) {
            std::cout << "Step 3: Result saved" << std::endl;
            std::cout << "Final data: " << processed << std::endl;
        });
    });
}

int main() {
    std::cout << "Starting async processing chain..." << std::endl;
    process_data_chain();
    
    // 等待异步操作完成
    std::this_thread::sleep_for(std::chrono::seconds(3));
    
    return 0;
}

7.3.2 异步编程模式

1. 生产者 - 消费者模式的异步实现

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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
#include <iostream>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <thread>
#include <future>
#include <chrono>

template<typename T>
class AsyncQueue {
private:
    std::queue<T> queue_;
    std::mutex mtx_;
    std::condition_variable cv_;
    bool stopped_;
    
public:
    AsyncQueue() : stopped_(false) {}
    
    void push(const T& item) {
        std::lock_guard<std::mutex> lock(mtx_);
        if (!stopped_) {
            queue_.push(item);
            cv_.notify_one();
        }
    }
    
    bool pop(T& item, std::chrono::milliseconds timeout = std::chrono::milliseconds(0)) {
        std::unique_lock<std::mutex> lock(mtx_);
        
        if (timeout.count() == 0) {
            cv_.wait(lock, [this]{ return !queue_.empty() || stopped_; });
        } else {
            if (!cv_.wait_for(lock, timeout, [this]{ return !queue_.empty() || stopped_; })) {
                return false;
            }
        }
        
        if (stopped_ && queue_.empty()) {
            return false;
        }
        
        item = queue_.front();
        queue_.pop();
        return true;
    }
    
    void stop() {
        std::lock_guard<std::mutex> lock(mtx_);
        stopped_ = true;
        cv_.notify_all();
    }
    
    bool empty() const {
        std::lock_guard<std::mutex> lock(mtx_);
        return queue_.empty();
    }
};

// 异步生产者
void producer(AsyncQueue<int>& queue) {
    for (int i = 0; i < 10; ++i) {
        std::cout << "Producing: " << i << std::endl;
        queue.push(i);
        std::this_thread::sleep_for(std::chrono::milliseconds(300));
    }
    queue.stop();
}

// 异步消费者
void consumer(AsyncQueue<int>& queue, const std::string& name) {
    int item;
    while (queue.pop(item)) {
        std::cout << "Consumer " << name << " processing: " << item << std::endl;
        std::this_thread::sleep_for(std::chrono::milliseconds(500));
    }
    std::cout << "Consumer " << name << " stopped" << std::endl;
}

int main() {
    AsyncQueue<int> queue;
    
    // 启动生产者
    std::thread prod_thread(producer, std::ref(queue));
    
    // 启动消费者
    std::thread cons_thread1(consumer, std::ref(queue), "A");
    std::thread cons_thread2(consumer, std::ref(queue), "B");
    
    prod_thread.join();
    cons_thread1.join();
    cons_thread2.join();
    
    return 0;
}

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
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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
#include <iostream>
#include <vector>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <future>
#include <functional>
#include <stdexcept>

class AsyncWorkerPool {
private:
    std::vector<std::thread> workers_;
    std::queue<std::function<void()>> tasks_;
    std::mutex mtx_;
    std::condition_variable cv_;
    bool stop_;
    
public:
    AsyncWorkerPool(size_t num_workers) : stop_(false) {
        for (size_t i = 0; i < num_workers; ++i) {
            workers_.emplace_back([this] {
                while (true) {
                    std::function<void()> task;
                    
                    {
                        std::unique_lock<std::mutex> lock(this->mtx_);
                        this->cv_.wait(lock, [this] {
                            return this->stop_ || !this->tasks_.empty();
                        });
                        
                        if (this->stop_ && this->tasks_.empty()) {
                            return;
                        }
                        
                        task = std::move(this->tasks_.front());
                        this->tasks_.pop();
                    }
                    
                    task();
                }
            });
        }
    }
    
    template<class F, class... Args>
    auto enqueue(F&& f, Args&&... args) 
        -> std::future<typename std::result_of<F(Args...)>::type> {
        
        using return_type = typename std::result_of<F(Args...)>::type;
        
        auto task = std::make_shared<std::packaged_task<return_type()>>(
            std::bind(std::forward<F>(f), std::forward<Args>(args)...)
        );
        
        std::future<return_type> res = task->get_future();
        
        {
            std::unique_lock<std::mutex> lock(mtx_);
            
            if (stop_) {
                throw std::runtime_error("enqueue on stopped AsyncWorkerPool");
            }
            
            tasks_.emplace([task]() { (*task)(); });
        }
        
        cv_.notify_one();
        return res;
    }
    
    ~AsyncWorkerPool() {
        {
            std::unique_lock<std::mutex> lock(mtx_);
            stop_ = true;
        }
        
        cv_.notify_all();
        
        for (std::thread& worker : workers_) {
            worker.join();
        }
    }
};

// 复杂计算任务
int complex_calculation(int x, int y) {
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    return x * y;
}

int main() {
    AsyncWorkerPool pool(4);
    std::vector<std::future<int>> results;
    
    // 提交多个异步任务
    for (int i = 0; i < 16; ++i) {
        results.emplace_back(
            pool.enqueue(complex_calculation, i, i + 1)
        );
    }
    
    // 获取结果
    for (auto&& result : results) {
        std::cout << result.get() << " ";
    }
    std::cout << std::endl;
    
    return 0;
}

7.4 异步编程的哲学思想

7.4.1 异步的时间哲学

异步编程体现了对时间资源的深刻理解和优化利用：

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
// 同步思维：浪费时间等待
void synchronous_thinking() {
    // 等待每个操作完成后再进行下一个
    auto result1 = operation1();  // 等待2秒
    auto result2 = operation2(result1);  // 等待3秒
    auto result3 = operation3(result2);  // 等待1秒
    
    // 总时间：6秒，CPU利用率低
}

// 异步思维：充分利用等待时间
void asynchronous_thinking() {
    // 启动所有操作，在等待时做其他事情
    auto future1 = async_operation1();
    auto future2 = async_operation2();
    auto future3 = async_operation3();
    
    // 在等待结果时处理其他任务
    do_other_work();
    
    // 获取结果
    auto result1 = future1.get();
    auto result2 = future2.get();
    auto result3 = future3.get();
    
    // 总时间：3秒（取决于最慢的操作），CPU利用率高
}

7.4.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
28
29
30
31
32
33
34
35
36
37
38
39
40
41
#include <iostream>
#include <future>
#include <chrono>
#include <vector>

// I/O密集型任务
void io_bound_task(int id) {
    std::cout << "Task " << id << " started (I/O bound)" << std::endl;
    
    // 模拟I/O等待时间
    std::this_thread::sleep_for(std::chrono::seconds(2));
    
    std::cout << "Task " << id << " completed" << std::endl;
}

int main() {
    const int num_tasks = 100;
    std::vector<std::future<void>> futures;
    
    auto start_time = std::chrono::high_resolution_clock::now();
    
    // 异步执行多个I/O密集型任务
    for (int i = 0; i < num_tasks; ++i) {
        futures.emplace_back(std::async(std::launch::async, io_bound_task, i));
    }
    
    // 等待所有任务完成
    for (auto& fut : futures) {
        fut.get();
    }
    
    auto end_time = std::chrono::high_resolution_clock::now();
    auto duration = std::chrono::duration_cast<std::chrono::seconds>(
        end_time - start_time
    );
    
    std::cout << "All " << num_tasks << " tasks completed in " 
              << duration.count() << " seconds" << std::endl;
    
    return 0;
}

7.4.3 异步的责任链哲学

异步编程建立了清晰的责任边界和协作模式：

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
#include <iostream>
#include <future>
#include <functional>
#include <string>

// 异步处理管道
class AsyncPipeline {
public:
    template<typename T, typename Func>
    static auto process(const T& input, Func&& func) {
        return std::async(std::launch::async, std::forward<Func>(func), input);
    }
    
    template<typename Future, typename Func>
    static auto then(Future&& future, Func&& func) {
        return std::async(std::launch::async, [future = std::move(future), func = std::forward<Func>(func)]() mutable {
            return func(future.get());
        });
    }
};

// 数据处理步骤
std::string read_data(const std::string& filename) {
    std::cout << "Step 1: Reading data from " << filename << std::endl;
    return "raw_data_from_" + filename;
}

std::string process_data(const std::string& data) {
    std::cout << "Step 2: Processing data" << std::endl;
    return data + "_processed";
}

void save_data(const std::string& processed_data) {
    std::cout << "Step 3: Saving processed data" << std::endl;
    std::cout << "Final data: " << processed_data << std::endl;
}

int main() {
    // 构建异步处理管道
    auto final_future = AsyncPipeline::then(
        AsyncPipeline::then(
            AsyncPipeline::process("input.txt", read_data),
            process_data
        ),
        save_data
    );
    
    // 等待整个管道完成
    final_future.get();
    
    return 0;
}

7.5 异步编程的最佳实践

7.5.1 避免回调地狱

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
// ❌ 回调地狱：深层嵌套的回调函数
void callback_hell() {
    async_operation1([](Result1 res1) {
        async_operation2(res1, [](Result2 res2) {
            async_operation3(res2, [](Result3 res3) {
                async_operation4(res3, [](Result4 res4) {
                    // 更多嵌套...
                });
            });
        });
    });
}

// ✅ 使用future避免回调地狱
void using_futures() {
    auto future1 = async_operation1();
    auto future2 = future1.then([](Result1 res1) {
        return async_operation2(res1);
    });
    auto future3 = future2.then([](Result2 res2) {
        return async_operation3(res2);
    });
    auto future4 = future3.then([](Result3 res3) {
        return async_operation4(res3);
    });
    
    auto final_result = future4.get();
}

7.5.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
28
29
30
31
32
33
34
35
36
#include <iostream>
#include <future>
#include <stdexcept>
#include <chrono>

void may_throw(bool should_throw) {
    std::this_thread::sleep_for(std::chrono::seconds(1));
    
    if (should_throw) {
        throw std::runtime_error("Something went wrong in async task");
    }
    
    std::cout << "Async task completed successfully" << std::endl;
}

int main() {
    try {
        // 启动可能抛出异常的异步任务
        auto future1 = std::async(std::launch::async, may_throw, true);
        auto future2 = std::async(std::launch::async, may_throw, false);
        
        // 处理异常
        try {
            future1.get();
        } catch (const std::exception& e) {
            std::cerr << "Caught exception from first task: " << e.what() << std::endl;
        }
        
        future2.get();
        
    } catch (const std::exception& e) {
        std::cerr << "Caught unexpected exception: " << e.what() << std::endl;
    }
    
    return 0;
}

7.5.3 合理设置超时

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
#include <iostream>
#include <future>
#include <chrono>
#include <stdexcept>

void long_running_task() {
    std::this_thread::sleep_for(std::chrono::seconds(5));
    std::cout << "Long running task completed" << std::endl;
}

int main() {
    auto future = std::async(std::launch::async, long_running_task);
    
    std::cout << "Waiting for task with timeout..." << std::endl;
    
    // 设置2秒超时
    auto status = future.wait_for(std::chrono::seconds(2));
    
    if (status == std::future_status::ready) {
        std::cout << "Task completed within timeout" << std::endl;
        future.get();
    } else if (status == std::future_status::timeout) {
        std::cout << "Task timed out after 2 seconds" << std::endl;
        // 注意：无法直接取消std::future任务
    } else if (status == std::future_status::deferred) {
        std::cout << "Task is deferred" << std::endl;
    }
    
    return 0;
}

7.5.4 异步任务的取消机制

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
56
57
58
59
60
61
62
63
64
65
66
67
68
#include <iostream>
#include <thread>
#include <atomic>
#include <future>
#include <chrono>

class CancellableTask {
private:
    std::atomic<bool> cancelled_;
    std::future<void> future_;
    
public:
    template<typename Func>
    CancellableTask(Func&& func) {
        cancelled_ = false;
        
        future_ = std::async(std::launch::async, [this, func = std::forward<Func>(func)]() {
            func([this]() { return cancelled_.load(); });
        });
    }
    
    void cancel() {
        cancelled_ = true;
    }
    
    void wait() {
        if (future_.valid()) {
            future_.wait();
        }
    }
    
    bool is_cancelled() const {
        return cancelled_.load();
    }
};

// 可取消的任务函数
void cancellable_operation(const std::function<bool()>& is_cancelled) {
    for (int i = 0; i < 10; ++i) {
        // 检查是否被取消
        if (is_cancelled()) {
            std::cout << "Task cancelled at iteration " << i << std::endl;
            return;
        }
        
        std::cout << "Task iteration " << i << std::endl;
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
    
    std::cout << "Task completed normally" << std::endl;
}

int main() {
    std::cout << "Starting cancellable task..." << std::endl;
    
    CancellableTask task(cancellable_operation);
    
    // 运行3秒后取消任务
    std::this_thread::sleep_for(std::chrono::seconds(3));
    std::cout << "Cancelling task..." << std::endl;
    task.cancel();
    
    task.wait();
    
    std::cout << "Main thread exiting" << std::endl;
    
    return 0;
}

7.6 异步编程的应用场景

7.6.1 I/O 密集型应用

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
#include <iostream>
#include <future>
#include <vector>
#include <string>
#include <chrono>

// 模拟网络请求
std::string fetch_data_from_network(const std::string& url) {
    std::cout << "Fetching data from " << url << std::endl;
    std::this_thread::sleep_for(std::chrono::milliseconds(500));
    return "Data from " + url;
}

int main() {
    std::vector<std::string> urls = {
        "http://example.com/api/data1",
        "http://example.com/api/data2", 
        "http://example.com/api/data3",
        "http://example.com/api/data4"
    };
    
    std::vector<std::future<std::string>> futures;
    
    auto start_time = std::chrono::high_resolution_clock::now();
    
    // 异步获取所有URL数据
    for (const auto& url : urls) {
        futures.emplace_back(
            std::async(std::launch::async, fetch_data_from_network, url)
        );
    }
    
    // 处理结果
    std::vector<std::string> results;
    for (auto& fut : futures) {
        results.push_back(fut.get());
    }
    
    auto end_time = std::chrono::high_resolution_clock::now();
    auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(
        end_time - start_time
    );
    
    std::cout << "All data fetched in " << duration.count() << "ms" << std::endl;
    
    for (const auto& result : results) {
        std::cout << result << std::endl;
    }
    
    return 0;
}

7.6.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
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
#include <iostream>
#include <future>
#include <vector>
#include <numeric>
#include <algorithm>

// 并行求和
template<typename Iterator>
typename Iterator::value_type parallel_sum(Iterator begin, Iterator end) {
    auto distance = std::distance(begin, end);
    
    // 小数据集直接计算
    if (distance <= 1000) {
        return std::accumulate(begin, end, 0);
    }
    
    auto mid = begin;
    std::advance(mid, distance / 2);
    
    // 递归并行计算
    auto future_left = std::async(std::launch::async, parallel_sum<Iterator>, begin, mid);
    auto right_sum = parallel_sum(mid, end);
    auto left_sum = future_left.get();
    
    return left_sum + right_sum;
}

int main() {
    std::vector<int> large_data(10000000, 1);
    
    auto start_time = std::chrono::high_resolution_clock::now();
    
    int total = parallel_sum(large_data.begin(), large_data.end());
    
    auto end_time = std::chrono::high_resolution_clock::now();
    auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(
        end_time - start_time
    );
    
    std::cout << "Total sum: " << total << std::endl;
    std::cout << "Time taken: " << duration.count() << "ms" << std::endl;
    
    return 0;
}

7.6.3 异步事件处理

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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
#include <iostream>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <thread>
#include <functional>
#include <chrono>
#include <random>

class EventDispatcher {
private:
    std::queue<std::function<void()>> events_;
    std::mutex mtx_;
    std::condition_variable cv_;
    std::thread dispatcher_thread_;
    bool running_;
    
public:
    EventDispatcher() : running_(true) {
        dispatcher_thread_ = std::thread([this] {
            dispatch_loop();
        });
    }
    
    ~EventDispatcher() {
        stop();
        if (dispatcher_thread_.joinable()) {
            dispatcher_thread_.join();
        }
    }
    
    template<typename Func>
    void post(Func&& func) {
        std::lock_guard<std::mutex> lock(mtx_);
        if (running_) {
            events_.emplace(std::forward<Func>(func));
            cv_.notify_one();
        }
    }
    
    void stop() {
        std::lock_guard<std::mutex> lock(mtx_);
        running_ = false;
        cv_.notify_one();
    }
    
private:
    void dispatch_loop() {
        while (running_) {
            std::function<void()> event;
            
            {
                std::unique_lock<std::mutex> lock(mtx_);
                cv_.wait(lock, [this] {
                    return !events_.empty() || !running_;
                });
                
                if (!running_ && events_.empty()) {
                    break;
                }
                
                if (!events_.empty()) {
                    event = std::move(events_.front());
                    events_.pop();
                }
            }
            
            if (event) {
                try {
                    event();
                } catch (const std::exception& e) {
                    std::cerr << "Event processing error: " << e.what() << std::endl;
                }
            }
        }
    }
};

// 事件源
class EventSource {
private:
    EventDispatcher& dispatcher_;
    std::thread event_thread_;
    std::atomic<bool> running_;
    std::mt19937 rng_;
    std::uniform_int_distribution<> dist_;
    
public:
    EventSource(EventDispatcher& dispatcher) 
        : dispatcher_(dispatcher), running_(true), rng_(std::random_device{}()), dist_(100, 1000) {
        
        event_thread_ = std::thread([this] {
            generate_events();
        });
    }
    
    ~EventSource() {
        running_ = false;
        if (event_thread_.joinable()) {
            event_thread_.join();
        }
    }
    
private:
    void generate_events() {
        int event_id = 0;
        
        while (running_) {
            int delay = dist_(rng_);
            std::this_thread::sleep_for(std::chrono::milliseconds(delay));
            
            // 发布事件
            dispatcher_.post([event_id, delay] {
                std::cout << "Processing event " << event_id 
                          << " (generated after " << delay << "ms)" << std::endl;
            });
            
            event_id++;
        }
    }
};

int main() {
    EventDispatcher dispatcher;
    EventSource source(dispatcher);
    
    std::cout << "Event system running. Press Enter to exit..." << std::endl;
    std::cin.get();
    
    return 0;
}

8. 并发编程哲学思想

8.1 并发编程的哲学基础

8.1.1 共享内存 vs 消息传递

共享内存模型：

  • 线程通过共享内存进行通信

  • 需要显式同步机制（锁、原子操作）

  • C++ 标准采用的主要模型

消息传递模型：

  • 线程通过发送消息进行通信

  • 数据在传递过程中所有权转移

  • Go 语言的 goroutine+channel 是典型代表

8.1.2 并发的本质：时间与空间的权衡

并发编程本质上是在有限的物理资源（CPU 核心）上，通过时间分片来模拟并行执行。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
// 时间分片的哲学体现
void concurrent_tasks() {
    // 任务A和任务B在同一个CPU核心上交替执行
    std::thread t1([]{
        for (int i = 0; i < 1000; ++i) {
            // 任务A的计算
            std::this_thread::yield();  // 主动让出CPU时间片
        }
    });
    
    std::thread t2([]{
        for (int i = 0; i < 1000; ++i) {
            // 任务B的计算
            std::this_thread::yield();
        }
    });
    
    t1.join();
    t2.join();
}

8.2 并发编程的设计哲学

8.2.1 最小同步原则

只在必要时使用同步机制，减少锁竞争。

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
// 不好的设计：过度同步
class BadDesign {
private:
    std::mutex mtx;
    int a, b, c;
    
public:
    void update_all(int new_a, int new_b, int new_c) {
        std::lock_guard<std::mutex> lock(mtx);
        a = new_a;
        b = new_b;
        c = new_c;
    }
    
    // 即使只需要读取一个变量，也要获取整个对象的锁
    int get_a() const {
        std::lock_guard<std::mutex> lock(mtx);
        return a;
    }
};

// 更好的设计：细粒度同步
class BetterDesign {
private:
    std::mutex mtx_a, mtx_b, mtx_c;
    int a, b, c;
    
public:
    void update_a(int new_a) {
        std::lock_guard<std::mutex> lock(mtx_a);
        a = new_a;
    }
    
    int get_a() const {
        std::lock_guard<std::mutex> lock(mtx_a);
        return a;
    }
    
    // 其他成员的类似方法...
};

8.2.2 不可变性原则

使用不可变对象减少同步需求。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
// 不可变对象：创建后状态不再改变
class ImmutableData {
private:
    const int value_;
    const std::string name_;
    
public:
    ImmutableData(int value, const std::string& name)
        : value_(value), name_(name) {}
    
    int get_value() const { return value_; }
    std::string get_name() const { return name_; }
    
    // 没有修改状态的方法
};

// 不可变对象可以安全地在多个线程间共享
void process_data(const ImmutableData& data) {
    // 读取操作不需要同步
    std::cout << data.get_name() << ": " << data.get_value() << std::endl;
}

8.2.3 单一职责原则在并发中的应用

每个线程应该有明确的职责，避免功能耦合。

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
// 职责分离的设计
class DataProducer {
private:
    std::queue<int> data_queue_;
    std::mutex mtx_;
    std::condition_variable cv_;
    
public:
    void produce_data(int data) {
        std::lock_guard<std::mutex> lock(mtx_);
        data_queue_.push(data);
        cv_.notify_one();
    }
    
    bool consume_data(int& data) {
        std::unique_lock<std::mutex> lock(mtx_);
        if (cv_.wait_for(lock, std::chrono::seconds(1), 
            [this]{ return !data_queue_.empty(); })) {
            data = data_queue_.front();
            data_queue_.pop();
            return true;
        }
        return false;
    }
};

// 专门的生产者线程
void producer_task(DataProducer& producer) {
    for (int i = 0; i < 100; ++i) {
        producer.produce_data(i);
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
    }
}

// 专门的消费者线程
void consumer_task(DataProducer& producer) {
    int data;
    while (producer.consume_data(data)) {
        std::cout << "Consumed: " << data << std::endl;
    }
}

8.3 异步编程的哲学深化

8.3.1 异步的 “等待哲学”

异步编程重新定义了 “等待” 的概念：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
// 同步等待：消极等待，浪费时间
void synchronous_waiting() {
    auto result = blocking_operation();  // 等待时什么都做不了
    process_result(result);
}

// 异步等待：积极等待，充分利用时间
void asynchronous_waiting() {
    auto future = async_operation();
    
    // 在等待结果时做其他有意义的事情
    do_other_useful_work();
    
    auto result = future.get();
    process_result(result);
}

8.3.2 异步的 “责任转移” 哲学

异步操作体现了责任的转移和委托：

1
2
3
4
5
6
7
8
9
10
11
// 同步模式：调用者承担所有责任
Result synchronous_mode() {
    // 调用者必须等待并处理所有细节
    return perform_complex_operation();
}

// 异步模式：责任转移给系统
std::future<Result> asynchronous_mode() {
    // 系统承担执行和通知的责任
    return std::async(perform_complex_operation);
}

8.3.3 异步的 “事件驱动” 哲学

异步编程基于事件驱动的世界观：

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
// 事件驱动的思维模式
class EventDrivenSystem {
private:
    EventDispatcher dispatcher_;
    
public:
    void initialize() {
        // 注册事件处理器
        dispatcher_.register_handler<DataReceivedEvent>(
            [this](const DataReceivedEvent& event) {
                this->handle_data(event.data);
            }
        );
        
        dispatcher_.register_handler<TimeoutEvent>(
            [this](const TimeoutEvent& event) {
                this->handle_timeout(event.timeout);
            }
        );
    }
    
    void start() {
        // 启动事件循环
        dispatcher_.run();
    }
    
private:
    void handle_data(const std::string& data) {
        // 处理收到的数据
        auto processed = process_data(data);
        
        // 触发新事件
        dispatcher_.post_event(DataProcessedEvent{processed});
    }
    
    void handle_timeout(int timeout) {
        // 处理超时事件
        dispatcher_.post_event(TimeoutHandledEvent{timeout});
    }
};

8.4 并发编程的性能哲学

8.4.1 Amdahl 定律：串行部分的限制

Amdahl 定律指出，程序的加速比受限于串行执行部分的比例。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
// 假设程序有80%的并行部分和20%的串行部分
void parallelizable_task() {
    // 80%的计算可以并行执行
    std::vector<std::thread> threads;
    for (int i = 0; i < 4; ++i) {
        threads.emplace_back([]{
            // 并行计算
        });
    }
    
    for (auto& t : threads) {
        t.join();
    }
}

void serial_task() {
    // 20%的计算必须串行执行
}

void main_task() {
    serial_task();          // 串行部分（20%）
    parallelizable_task();  // 并行部分（80%）
}

8.4.2 Gustafson 定律：问题规模的影响

Gustafson 定律表明，随着问题规模的增加，并行计算的收益也会增加。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
// 处理更大规模的数据时，并行的优势更明显
template<typename T>
void process_large_data(const std::vector<T>& data) {
    const size_t num_threads = std::thread::hardware_concurrency();
    const size_t chunk_size = data.size() / num_threads;
    
    std::vector<std::thread> threads;
    
    for (size_t i = 0; i < num_threads; ++i) {
        size_t start = i * chunk_size;
        size_t end = (i == num_threads - 1) ? data.size() : (i + 1) * chunk_size;
        
        threads.emplace_back([start, end, &data]{
            for (size_t j = start; j < end; ++j) {
                // 处理单个数据元素
                process_element(data[j]);
            }
        });
    }
    
    for (auto& t : threads) {
        t.join();
    }
}

9. 最佳实践与设计原则

9.1 线程安全设计原则

9.1.1 优先使用高层抽象

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
// 不好的做法：手动管理线程和共享数据
void bad_example() {
    int result = 0;
    std::mutex mtx;
    
    std::thread t([&]{
        std::lock_guard<std::mutex> lock(mtx);
        result = expensive_calculation();
    });
    
    t.join();
    std::cout << result << std::endl;
}

// 好的做法：使用std::async
void good_example() {
    auto future = std::async(std::launch::async, expensive_calculation);
    std::cout << future.get() << std::endl;
}

9.1.2 始终使用 RAII 管理资源

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
// 不好的做法：手动管理锁
void bad_lock_management() {
    std::mutex mtx;
    mtx.lock();
    
    try {
        // 可能抛出异常的操作
        risky_operation();
        mtx.unlock();
    } catch (...) {
        mtx.unlock();
        throw;
    }
}

// 好的做法：使用RAII锁
void good_lock_management() {
    std::mutex mtx;
    std::lock_guard<std::mutex> lock(mtx);
    
    // 即使抛出异常，锁也会自动释放
    risky_operation();
}

9.1.3 避免共享可变状态

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
// 不好的做法：共享可变状态
class SharedState {
private:
    std::mutex mtx;
    int state_;
    
public:
    void update_state(int new_state) {
        std::lock_guard<std::mutex> lock(mtx);
        state_ = new_state;
    }
};

// 好的做法：使用消息传递
class MessagePassing {
private:
    std::queue<int> messages_;
    std::mutex mtx_;
    std::condition_variable cv_;
    
public:
    void send_message(int message) {
        std::lock_guard<std::mutex> lock(mtx_);
        messages_.push(message);
        cv_.notify_one();
    }
    
    int receive_message() {
        std::unique_lock<std::mutex> lock(mtx_);
        cv_.wait(lock, [this]{ return !messages_.empty(); });
        
        int message = messages_.front();
        messages_.pop();
        return message;
    }
};

9.2 异步编程最佳实践

9.2.1 选择合适的异步模型

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
// 根据场景选择不同的异步模型

// 1. 简单的异步计算：使用std::async
auto async_result = std::async(std::launch::async, []{
    return expensive_calculation();
});

// 2. 需要主动设置结果：使用std::promise
std::promise<int> prom;
std::future<int> fut = prom.get_future();

std::thread([&prom]{
    prom.set_value(calculate_result());
}).detach();

// 3. 复杂的异步工作流：使用事件循环
EventLoop loop;
loop.post([]{
    // 异步任务
});

9.2.2 正确处理异步错误

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
// 异步操作中的异常处理
void async_with_exception_handling() {
    try {
        auto future = std::async(std::launch::async, []{
            try {
                // 可能抛出异常的操作
                return risky_operation();
            } catch (...) {
                // 在异步任务中处理或重新抛出
                std::throw_with_nested(std::runtime_error("Async operation failed"));
            }
        });
        
        auto result = future.get();
    } catch (const std::exception& e) {
        std::cerr << "Async error: " << e.what() << std::endl;
    }
}

9.2.3 避免过度异步化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
// 不是所有操作都需要异步化

// 适合异步的场景：I/O密集型操作
auto io_future = std::async(std::launch::async, []{
    return read_large_file();  // I/O操作，适合异步
});

// 不适合异步的场景：简单计算
int simple_calc = add_numbers(1, 2);  // 直接同步调用更高效

// 权衡异步的开销
if (operation_is_io_bound() && operation_will_take_long_time()) {
    use_async();
} else {
    use_sync();
}

9.3 性能优化原则

9.3.1 减少锁竞争

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
// 高竞争场景的优化
class HighContentionData {
private:
    std::vector<std::mutex> mutexes_;
    std::vector<int> data_;
    size_t num_shards_;
    
public:
    HighContentionData(size_t size, size_t num_shards)
        : num_shards_(num_shards), mutexes_(num_shards), data_(size) {}
    
    void update(size_t index, int value) {
        size_t shard = index % num_shards_;
        std::lock_guard<std::mutex> lock(mutexes_[shard]);
        data_[index] = value;
    }
    
    int get(size_t index) const {
        size_t shard = index % num_shards_;
        std::lock_guard<std::mutex> lock(mutexes_[shard]);
        return data_[index];
    }
};

9.3.2 使用无锁数据结构

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
// 当适用时，无锁结构可以提供更好的性能
#include <atomic>

class LockFreeCounter {
private:
    std::atomic<int> count_;
    
public:
    LockFreeCounter() : count_(0) {}
    
    void increment() {
        count_.fetch_add(1, std::memory_order_relaxed);
    }
    
    int get_count() const {
        return count_.load(std::memory_order_relaxed);
    }
};

9.3.3 合理使用线程局部存储

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
// 线程局部存储可以避免共享数据
thread_local int thread_specific_data = 0;

void thread_function(int id) {
    thread_specific_data = id;
    // 每个线程都有自己的副本
    std::cout << "Thread " << id << " has data: " << thread_specific_data << std::endl;
}

int main() {
    std::thread t1(thread_function, 1);
    std::thread t2(thread_function, 2);
    
    t1.join();
    t2.join();
    
    return 0;
}

9.4 错误处理原则

9.4.1 线程内异常处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
void thread_with_exception_handling() {
    try {
        // 可能抛出异常的操作
        risky_operation();
    } catch (const std::exception& e) {
        std::cerr << "Thread caught exception: " << e.what() << std::endl;
        // 适当的清理和错误处理
    } catch (...) {
        std::cerr << "Thread caught unknown exception" << std::endl;
    }
}

int main() {
    std::thread t(thread_with_exception_handling);
    t.join();
    return 0;
}

9.4.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
28
29
class GracefulShutdown {
private:
    std::atomic<bool> stop_flag_;
    std::thread worker_thread_;
    
public:
    GracefulShutdown() : stop_flag_(false) {
        worker_thread_ = std::thread([this]{
            while (!stop_flag_.load(std::memory_order_acquire)) {
                // 执行工作
                do_work();
                std::this_thread::sleep_for(std::chrono::milliseconds(100));
            }
            std::cout << "Thread shutting down gracefully" << std::endl;
        });
    }
    
    ~GracefulShutdown() {
        stop_flag_.store(true, std::memory_order_release);
        if (worker_thread_.joinable()) {
            worker_thread_.join();
        }
    }
    
private:
    void do_work() {
        // 实际的工作内容
    }
};

10. 常见问题与解决方案

10.1 死锁问题

10.1.1 死锁的产生条件

  1. 互斥条件：资源只能被一个线程持有

  2. 持有并等待：线程持有资源并等待其他资源

  3. 不可剥夺：资源不能被强制剥夺

  4. 循环等待：线程间形成等待循环

10.1.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
28
29
30
31
32
33
34
35
36
37
38
39
40
// 死锁示例
std::mutex mutex1, mutex2;

void deadlock_example1() {
    std::lock_guard<std::mutex> lock1(mutex1);
    std::this_thread::sleep_for(std::chrono::milliseconds(10));
    std::lock_guard<std::mutex> lock2(mutex2);  // 可能死锁
}

void deadlock_example2() {
    std::lock_guard<std::mutex> lock2(mutex2);
    std::this_thread::sleep_for(std::chrono::milliseconds(10));
    std::lock_guard<std::mutex> lock1(mutex1);  // 可能死锁
}

// 死锁解决方案：统一锁获取顺序
void deadlock_solution1() {
    std::lock(mutex1, mutex2);
    std::lock_guard<std::mutex> lock1(mutex1, std::adopt_lock);
    std::lock_guard<std::mutex> lock2(mutex2, std::adopt_lock);
}

// 死锁解决方案：使用try_lock
void deadlock_solution2() {
    while (true) {
        if (mutex1.try_lock()) {
            if (mutex2.try_lock()) {
                // 成功获取两个锁
                std::lock_guard<std::mutex> lock1(mutex1, std::adopt_lock);
                std::lock_guard<std::mutex> lock2(mutex2, std::adopt_lock);
                break;
            } else {
                mutex1.unlock();
                std::this_thread::sleep_for(std::chrono::milliseconds(10));
            }
        } else {
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
        }
    }
}

10.2 数据竞争问题

10.2.1 数据竞争的检测

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
// 数据竞争示例
int shared_data = 0;

void竞争_condition() {
    for (int i = 0; i < 100000; ++i) {
        shared_data++;  // 数据竞争！
    }
}

// 解决方案：使用原子操作
std::atomic<int> atomic_data(0);

void no_race_condition() {
    for (int i = 0; i < 100000; ++i) {
        atomic_data++;  // 原子操作，无数据竞争
    }
}

10.3 异步编程常见问题

10.3.1 异步任务的取消问题

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
// std::future不支持直接取消，需要自己实现
class CancellableFuture {
private:
    std::future<void> future_;
    std::atomic<bool>& cancelled_;
    
public:
    CancellableFuture(std::future<void>&& future, std::atomic<bool>& cancelled)
        : future_(std::move(future)), cancelled_(cancelled) {}
    
    bool wait_for(const std::chrono::milliseconds& timeout) {
        auto status = future_.wait_for(timeout);
        
        if (status == std::future_status::ready) {
            return true;
        } else if (cancelled_.load()) {
            return false;
        }
        
        return status == std::future_status::ready;
    }
};

10.3.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
28
29
30
31
32
// 确保异步结果的正确生命周期管理
class AsyncResultManager {
private:
    std::vector<std::future<void>> active_futures_;
    std::mutex mtx_;
    
public:
    template<typename Func>
    void start_async_task(Func&& func) {
        std::lock_guard<std::mutex> lock(mtx_);
        
        auto future = std::async(std::launch::async, [this, func = std::forward<Func>(func)]() {
            func();
            cleanup_completed_tasks();
        });
        
        active_futures_.emplace_back(std::move(future));
    }
    
private:
    void cleanup_completed_tasks() {
        std::lock_guard<std::mutex> lock(mtx_);
        
        active_futures_.erase(
            std::remove_if(active_futures_.begin(), active_futures_.end(),
                [](auto& future) {
                    return future.wait_for(std::chrono::seconds(0)) == std::future_status::ready;
                }),
            active_futures_.end()
        );
    }
};

10.4 调试技巧

10.4.1 线程安全的调试宏

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
#ifdef DEBUG
#define LOCK_GUARD(mutex) \
    std::lock_guard<std::mutex> lock(mutex); \
    std::cout << "Thread " << std::this_thread::get_id() << " locked " << #mutex << " at " << __FILE__ << ":" << __LINE__ << std::endl;
#else
#define LOCK_GUARD(mutex) \
    std::lock_guard<std::mutex> lock(mutex);
#endif

class DebuggableData {
private:
    std::mutex mtx_;
    int data_;
    
public:
    void update(int value) {
        LOCK_GUARD(mtx_);
        data_ = value;
    }
};

10.4.2 异步操作的调试

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
// 异步操作的调试辅助
class AsyncDebugger {
public:
    template<typename T>
    static std::future<T> debug_future(std::future<T>&& future, const std::string& name) {
        return std::async(std::launch::async, [future = std::move(future), name]() mutable {
            std::cout << "Waiting for future: " << name << std::endl;
            
            try {
                auto result = future.get();
                std::cout << "Future completed: " << name << std::endl;
                return result;
            } catch (const std::exception& e) {
                std::cerr << "Future failed: " << name << " - " << e.what() << std::endl;
                throw;
            }
        });
    }
};

总结

C++ 并发编程是一个复杂而强大的领域，需要开发者同时掌握：

技术层面

  • 线程管理：std::thread 的创建、销毁和同步

  • 同步机制：mutex、condition_variable 的正确使用

  • 异步编程：future、promise、async 的灵活应用

  • 内存模型：原子操作和内存顺序的理解

  • 设计模式：各种并发和异步设计模式的应用

哲学思想层面

  • 并发本质：时间分片与空间共享的权衡

  • 异步哲学：重新定义等待，最大化资源利用

  • 设计原则：最小同步、不可变性、单一职责

  • 性能权衡：在正确性和性能之间找到平衡点

最佳实践

  • 始终使用 RAII 管理资源

  • 优先选择高层抽象

  • 避免共享可变状态

  • 合理处理异常和错误

  • 重视测试和调试

掌握这些知识和思想，能够帮助开发者编写出安全、高效、可维护的并发 C++ 程序。并发编程不仅是技术问题，更是一种思维方式和设计哲学的体现。

Date: October 17, 2025

Code: https://github.com/cplusplus-concurrency