多线程与并发

概述

C++11 引入了标准化的多线程支持，包括线程管理、互斥量、条件变量和异步操作等组件。C++17 增加了并行算法，C++20 引入了 jthread（自动 join 的线程）、信号量和闩/屏障。多线程编程的核心挑战是正确处理共享数据的同步，避免数据竞争和死锁。

C++ 的并发模型基于内存序和 happens-before 关系，理解这些概念是编写正确并发程序的基础。标准库提供的同步原语覆盖了从简单的互斥锁到复杂的无锁编程的各种场景。

基础概念

同步原语一览

数据竞争与同步

数据竞争是指两个或更多线程并发访问同一内存位置，且至少一个是写操作，且没有同步关系。数据竞争导致未定义行为，是并发编程中最常见的 bug 来源。

快速上手

创建和管理线程

#include <thread>
#include <iostream>

void task(int id) {
    std::cout << "线程 " << id << " 正在执行" << std::endl;
}

int main() {
    // 创建线程
    std::thread t1(task, 1);
    std::thread t2(task, 2);

    // 等待线程完成
    t1.join();
    t2.join();

    // C++20: jthread 自动 join
    std::jthread t3(task, 3);  // 析构时自动 join
    return 0;
}

使用互斥锁保护共享数据

#include <mutex>
#include <vector>

class ThreadSafeCounter {
    int count_ = 0;
    std::mutex mtx_;
public:
    void increment() {
        std::lock_guard<std::mutex> lock(mtx_);  // RAII 加锁
        ++count_;
    }

    int get() const {
        std::lock_guard<std::mutex> lock(mtx_);
        return count_;
    }
};

异步操作

#include <future>
#include <iostream>

int heavyComputation() {
    std::this_thread::sleep_for(std::chrono::seconds(2));
    return 42;
}

int main() {
    // 异步启动任务
    auto future = std::async(std::launch::async, heavyComputation);

    // 做其他工作...

    // 获取结果（如果未完成会阻塞）
    int result = future.get();
    std::cout << "结果: " << result << std::endl;  // 42
    return 0;
}

详细用法

unique_lock 灵活锁管理

#include <mutex>
#include <condition_variable>

class BoundedBuffer {
    std::vector<int> buffer_;
    size_t capacity_;
    std::mutex mtx_;
    std::condition_variable not_full_;
    std::condition_variable not_empty_;

public:
    BoundedBuffer(size_t cap) : capacity_(cap) {}

    void produce(int value) {
        std::unique_lock<std::mutex> lock(mtx_);
        not_full_.wait(lock, [this]() { return buffer_.size() < capacity_; });
        buffer_.push_back(value);
        not_empty_.notify_one();
    }

    int consume() {
        std::unique_lock<std::mutex> lock(mtx_);
        not_empty_.wait(lock, [this]() { return !buffer_.empty(); });
        int value = buffer_.back();
        buffer_.pop_back();
        not_full_.notify_one();
        return value;
    }
};

读写锁（shared_mutex）

#include <shared_mutex>
#include <map>
#include <string>

class ThreadSafeCache {
    mutable std::shared_mutex mtx_;
    std::map<std::string, std::string> data_;
public:
    // 读操作：允许多个线程同时读
    std::string get(const std::string& key) const {
        std::shared_lock<std::shared_mutex> lock(mtx_);  // 共享锁
        auto it = data_.find(key);
        return it != data_.end() ? it->second : "";
    }

    // 写操作：独占访问
    void set(const std::string& key, const std::string& value) {
        std::unique_lock<std::shared_mutex> lock(mtx_);  // 独占锁
        data_[key] = value;
    }
};

条件变量

#include <condition_variable>
#include <queue>
#include <thread>

template<typename T>
class ThreadSafeQueue {
    std::queue<T> queue_;
    mutable std::mutex mtx_;
    std::condition_variable cv_;
    bool stopped_ = false;

public:
    void push(T value) {
        {
            std::lock_guard<std::mutex> lock(mtx_);
            queue_.push(std::move(value));
        }
        cv_.notify_one();
    }

    bool pop(T& value) {
        std::unique_lock<std::mutex> lock(mtx_);
        cv_.wait(lock, [this]() { return !queue_.empty() || stopped_; });
        if (stopped_ && queue_.empty()) return false;
        value = std::move(queue_.front());
        queue_.pop();
        return true;
    }

    void stop() {
        {
            std::lock_guard<std::mutex> lock(mtx_);
            stopped_ = true;
        }
        cv_.notify_all();
    }
};

常见场景

线程池

#include <vector>
#include <queue>
#include <thread>
#include <functional>
#include <future>

class ThreadPool {
    std::vector<std::thread> workers_;
    std::queue<std::function<void()>> tasks_;
    std::mutex mtx_;
    std::condition_variable cv_;
    bool stop_ = false;

public:
    ThreadPool(size_t threads) {
        for (size_t i = 0; i < threads; ++i) {
            workers_.emplace_back([this]() {
                while (true) {
                    std::function<void()> task;
                    {
                        std::unique_lock<std::mutex> lock(mtx_);
                        cv_.wait(lock, [this]() {
                            return stop_ || !tasks_.empty();
                        });
                        if (stop_ && tasks_.empty()) return;
                        task = std::move(tasks_.front());
                        tasks_.pop();
                    }
                    task();
                }
            });
        }
    }

    template<typename F>
    auto submit(F&& f) -> std::future<decltype(f())> {
        auto task = std::make_shared<std::packaged_task<decltype(f())()>>(
            std::forward<F>(f));
        auto result = task->get_future();
        {
            std::lock_guard<std::mutex> lock(mtx_);
            tasks_.emplace([task]() { (*task)(); });
        }
        cv_.notify_one();
        return result;
    }

    ~ThreadPool() {
        {
            std::lock_guard<std::mutex> lock(mtx_);
            stop_ = true;
        }
        cv_.notify_all();
        for (auto& w : workers_) w.join();
    }
};

并行数据处理

#include <algorithm>
#include <execution>

void parallelProcess(std::vector<int>& data) {
    // C++17 并行算法
    std::for_each(std::execution::par, data.begin(), data.end(),
        [](int& n) { n = n * 2 + 1; });

    // 并行排序
    std::sort(std::execution::par, data.begin(), data.end());
}

注意事项

• 始终在创建线程后调用 join() 或 detach()，否则析构时会调用 std::terminate
• 优先使用 RAII 包装器（lock_guard、unique_lock）管理锁，避免手动加锁/解锁
• 避免在持有锁时调用未知代码（回调函数等），可能导致死锁
• 条件变量的 wait 必须使用谓词形式，避免虚假唤醒
• std::async 的默认启动策略可能是延迟执行，需要异步执行时应使用 std::launch::async
• C++20 的 jthread 在析构时自动请求停止并 join，比 thread 更安全

进阶用法

C++20 信号量

#include <semaphore>

// 计数信号量：限制并发访问数量
std::counting_semaphore<10> semaphore{3};  // 最多3个并发

void limitedAccess() {
    semaphore.acquire();  // 获取许可（计数 -1）
    // 最多3个线程同时执行此处
    doWork();
    semaphore.release();  // 释放许可（计数 +1）
}

// 二元信号量
std::binary_semaphore signal{0};

// 等待信号
signal.acquire();  // 阻塞直到 release

// 发送信号
signal.release();

C++20 闩与屏障

#include <barrier>
#include <latch>

// 闩（Latch）：一次性同步点
std::latch done{5};  // 计数器初始为5

void worker() {
    doWork();
    done.count_down();  // 计数 -1
}

// 主线程等待所有工作完成
done.wait();  // 阻塞直到计数为0

// 屏障（Barrier）：可重复使用的同步点
std::barrier sync_point{5};  // 5个线程同步

void phasedWorker() {
    for (int phase = 0; phase < 3; ++phase) {
        doPhaseWork(phase);
        sync_point.arrive_and_wait();  // 等待所有线程完成当前阶段
    }
}

C++20 jthread 与停止令牌

#include <thread>

void monitoredTask(std::stop_token token) {
    while (!token.stop_requested()) {
        doWork();
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
    std::cout << "任务已停止" << std::endl;
}

// 使用 jthread
std::jthread t(monitoredTask);

// 请求停止
t.request_stop();  // 设置停止标志
t.join();          // 等待线程退出