C++ 并发编程知识点详解

std::mutex mtx;
void f() {
    std::lock_guard<std::mutex> lock(mtx);  // 构造即加锁
    // 临界区
    // 作用域结束（包括异常）自动解锁
}

std::mutex mtx;

// 1. 延迟加锁：先创建，稍后手动加锁
std::unique_lock<std::mutex> lock(mtx, std::defer_lock);  // 不加锁
lock.lock();        // 稍后手动加锁
lock.unlock();      // 提前解锁

// 2. 配合条件变量（必须用 unique_lock）
std::condition_variable cv;
std::unique_lock<std::mutex> lk(mtx);
cv.wait(lk, [] { return ready; });  // 条件变量要求 unique_lock

// 3. 可移动所有权
std::unique_lock<std::mutex> lk2 = std::move(lock);

#include <thread>

void task() { /* 耗时工作 */ }

std::thread t(task);
// ... 主线程做其他事
t.join();   // 阻塞，直到 t 执行完
// join 后主线程才能继续，保证 t 完成

std::thread t(task);
t.detach();   // t 与线程脱离，主线程不等待
// 主线程继续执行，t 独立在后台运行
// 线程结束后系统自动回收资源

void bad() {
    int local = 42;
    std::thread t([&local] { std::cout << local; });  // 捕获局部变量引用
    t.detach();   // ❌ 危险！local 可能已销毁，子线程访问悬挂引用
}

std::thread t(task);
// 析构前必须 join 或 detach
if (t.joinable()) {
    t.join();   // 或 t.detach();
}

void bad() {
    std::thread t(task);
    // 忘记 join/detach
}   // ❌ std::thread 析构 → std::terminate，程序崩溃

void good() {
    std::jthread t(task);
    // 不需要手动 join
}   // ✅ jthread 析构自动 join，安全

#include <thread>
#include <iostream>

void worker(std::stop_token st) {
    while (!st.stop_requested()) {   // 检查是否请求停止
        // 工作...
    }
}

int main() {
    std::jthread t(worker);
    // ...
    t.request_stop();   // 请求停止（协作式，worker 检查到后退出）
    // t 析构时也会自动 request_stop + join
}

#include <mutex>
#include <atomic>

// 1. 互斥锁版本：适合复杂逻辑
class ThreadSafeCounter {
public:
    void increment() {
        std::lock_guard<std::mutex> lock(mtx_);  // RAII 加锁
        ++count_;
    }
    int get() const {
        std::lock_guard<std::mutex> lock(mtx_);  // const 方法里锁需 mutable
        return count_;
    }
private:
    mutable std::mutex mtx_;
    int count_ = 0;
};

// 2. 原子版本：适合简单计数，无锁、更快
class AtomicCounter {
public:
    void increment() { count_.fetch_add(1, std::memory_order_relaxed); }
    int get() const { return count_.load(std::memory_order_relaxed); }
private:
    std::atomic<int> count_{0};
};

class Bad {
public:
    int& get() { return data_; }   // ❌ 返回内部引用，调用方可绕过锁修改
    std::mutex mtx_;               // ❌ 公有锁，调用方可能误用/忘记解锁
private:
    int data_;
};

#include <unistd.h>
#include <sys/wait.h>
#include <cstring>
#include <cstdio>

int main() {
    int fd[2];
    pipe(fd);                    // fd[0] 读端，fd[1] 写端
    pid_t pid = fork();
    if (pid == 0) {              // 子进程：写
        close(fd[0]);
        const char* msg = "hello parent";
        write(fd[1], msg, strlen(msg) + 1);
        close(fd[1]);
    } else {                     // 父进程：读
        close(fd[1]);
        char buf[64];
        read(fd[0], buf, sizeof(buf));
        printf("parent got: %s\n", buf);
        close(fd[0]);
        wait(nullptr);
    }
}

#include <future>
#include <thread>
#include <iostream>

int add(int a, int b) { return a + b; }

int main() {
    // 1. async：自动异步执行，返回 future
    std::future<int> f1 = std::async(std::launch::async, add, 1, 2);
    std::cout << f1.get() << "\n";   // 阻塞取结果

    // 2. packaged_task：包装任务，手动决定执行线程
    std::packaged_task<int(int,int)> task(add);
    std::future<int> f2 = task.get_future();   // 先取 future
    std::thread t(std::move(task), 3, 4);      // 在指定线程执行
    std::cout << f2.get() << "\n";
    t.join();

    // 3. promise：完全手动控制结果写入
    std::promise<int> p;
    std::future<int> f3 = p.get_future();
    std::thread producer([&p] { p.set_value(42); });  // 手动设值
    std::cout << f3.get() << "\n";
    producer.join();
}

std::async(std::launch::async, doWork);   // ❌ 临时 future 立即析构 → 阻塞等任务完成
// 相当于同步调用，失去异步意义

auto f = std::async(std::launch::async, doWork);  // ✅ 保存 future

auto f = std::async(std::launch::deferred, add, 1, 2);
// 此时 add 尚未执行
std::cout << f.get() << "\n";   // 这里才在当前线程执行 add

std::future<int> f = std::async([]() -> int { throw std::runtime_error("err"); });
try {
    f.get();   // 这里才抛出异常
} catch (const std::exception& e) {
    std::cout << e.what() << "\n";
}

#include <atomic>

std::atomic<int> cnt{0};

cnt.store(10);                       // 原子写
int v = cnt.load();                  // 原子读
cnt.fetch_add(1);                    // 原子加，返回旧值
cnt.fetch_sub(1);                    // 原子减
cnt.exchange(20);                    // 原子交换，返回旧值
bool ok = cnt.compare_exchange_weak(v, 30);  // CAS：等于 v 则改为 30

// 指定内存序（默认 seq_cst，最强）
cnt.fetch_add(1, std::memory_order_relaxed);  // 只要原子性、不强求顺序

int expected = 20;
// 若 cnt==20，则置为 30，返回 true；否则 expected 更新为实际值，返回 false
if (cnt.compare_exchange_strong(expected, 30)) {
    // 交换成功
}

#include <atomic>
#include <mutex>
#include <vector>
#include <algorithm>

// 场景 1：简单计数器 → 原子变量
std::atomic<int> count{0};
void addCount() { count.fetch_add(1); }

// 场景 2：复合操作（先判断再插入）→ 必须用锁
std::mutex mtx;
std::vector<int> v;
void insertIfAbsent(int x) {
    std::lock_guard<std::mutex> lock(mtx);
    // 检查 + 插入 必须作为整体原子执行，原子变量无法覆盖
    if (std::find(v.begin(), v.end(), x) == v.end())
        v.push_back(x);
}

// 伪代码：自旋锁
#include <atomic>

std::atomic_flag lock = ATOMIC_FLAG_INIT;
void lock_it() {
    while (lock.test_and_set(std::memory_order_acquire)) {
        // 忙等（自旋）：一直尝试直到成功
    }
}
void unlock_it() { lock.clear(std::memory_order_release); }

// futex 思想（简化）
// 1. 用户态 CAS：锁空闲则直接抢到，不发系统调用
// 2. CAS 失败：futex_wait 挂起线程
// 3. 解锁时若有人等待：futex_wake 唤醒

#include <atomic>
#include <thread>
#include <string>
#include <cassert>

std::atomic<bool> ready{false};
std::string data;

void producer() {
    data = "hello";                                 // 普通写
    ready.store(true, std::memory_order_release);   // release 发布
}

void consumer() {
    while (!ready.load(std::memory_order_acquire)) {}  // acquire 获取
    // acquire 成功后，一定能看到 producer 在 release 之前的所有写
    assert(data == "hello");
}

std::atomic<int> a{0}, b{0};
// relaxed 下，另一线程可能看到 a=1, b=0（顺序不定）
// seq_cst 下，顺序对所有线程一致

// ❌ 错误示范：检查条件与等待之间存在窗口期
// 线程 A 检查 ready == false，准备 sleep
// —— 就在此时，线程 B 把 ready 置 true 并 notify ——
// 但 A 还没进入 sleep，错过了这次通知 → 永远睡下去
while (!ready) {
    // sleep 之前 ready 被改 + notify，通知丢失
}

#include <mutex>
#include <condition_variable>

std::mutex mtx;
std::condition_variable cv;
bool ready = false;

// 消费者
void consumer() {
    std::unique_lock<std::mutex> lock(mtx);   // 1. 先加锁保护 ready
    cv.wait(lock, [] { return ready; });      // 2. 原子地 解锁+挂起；唤醒后重新加锁
    // 到这里 ready 必为 true
}

// 生产者
void producer() {
    {
        std::lock_guard<std::mutex> lock(mtx);  // 修改条件也要加锁
        ready = true;
    }
    cv.notify_one();                            // 通知（可不在锁内）
}

#include <thread>
#include <iostream>

thread_local int tls_counter = 0;   // 每个线程一份

void worker() {
    ++tls_counter;   // 只修改本线程自己的副本
    std::cout << tls_counter << "\n";
}

int main() {
    std::thread t1(worker);   // 输出 1
    std::thread t2(worker);   // 输出 1（各自独立）
    t1.join(); t2.join();
}

#include <mutex>
#include <condition_variable>
#include <queue>

std::mutex mtx;
std::condition_variable cv;
std::queue<int> q;
const int MAX = 10;

// 生产者
void produce(int x) {
    std::unique_lock<std::mutex> lock(mtx);
    cv.wait(lock, [] { return q.size() < MAX; });  // 队列未满才生产
    q.push(x);
    cv.notify_one();
}

// 消费者
void consume() {
    std::unique_lock<std::mutex> lock(mtx);
    cv.wait(lock, [] { return !q.empty(); });       // 队列非空才消费
    int x = q.front();
    q.pop();
    cv.notify_one();
}

while (!pred()) {
    cv.wait(lock);
}

#include <mutex>

std::mutex a, b;

void f1() {
    std::lock_guard<std::mutex> l1(a);
    std::lock_guard<std::mutex> l2(b);   // 先 a 后 b
}

void f2() {
    std::lock_guard<std::mutex> l1(b);   // ❌ 先 b 后 a，与 f1 顺序相反
    std::lock_guard<std::mutex> l2(a);   // 可能死锁
}

#include <mutex>

std::mutex a, b;
void safe() {
    std::scoped_lock lock(a, b);   // 一次性锁住，内部处理顺序，异常安全
    // 同时持有 a、b
}

std::unique_lock<std::mutex> l1(a, std::defer_lock);
std::unique_lock<std::mutex> l2(b, std::defer_lock);
if (std::try_lock(l1, l2) == -1) { /* 成功获得两把 */ }
else { /* 至少一把没拿到，不强行等待 */ }

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

class ThreadPool {
public:
    explicit ThreadPool(size_t n) : stop_(false) {
        for (size_t i = 0; i < n; ++i)
            workers_.emplace_back([this] { workerLoop(); });
    }

    // 提交任务，返回 future 以获取结果
    template <class F, class... Args>
    auto submit(F&& f, Args&&... args)
        -> std::future<std::invoke_result_t<F, Args...>> {
        using R = std::invoke_result_t<F, Args...>;
        auto task = std::make_shared<std::packaged_task<R()>>(
            std::bind(std::forward<F>(f), std::forward<Args>(args)...));
        std::future<R> res = task->get_future();
        {
            std::lock_guard<std::mutex> lock(mtx_);
            tasks_.emplace([task] { (*task)(); });
        }
        cv_.notify_one();
        return res;
    }

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

private:
    void workerLoop() {
        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();
        }
    }

    std::vector<std::thread> workers_;
    std::queue<std::function<void()>> tasks_;
    std::mutex mtx_;
    std::condition_variable cv_;
    bool stop_;
};

#include <coroutine>

// 一个最简单的 generator 示意（无栈协程）
struct Generator {
    struct promise_type {
        int value;
        Generator get_return_object() {
            return {std::coroutine_handle<promise_type>::from_promise(*this)};
        }
        std::suspend_always initial_suspend() { return {}; }
        std::suspend_always final_suspend() noexcept { return {}; }
        std::suspend_always yield_value(int v) { value = v; return {}; }
        void return_void() {}
        void unhandled_exception() {}
    };
    std::coroutine_handle<promise_type> h;
};

Generator numbers() {
    for (int i = 0; i < 3; ++i)
        co_yield i;   // 只能在这种显式点挂起
}

#include <thread>

// 线程：每个任务一个线程，适合 CPU 计算
std::thread t(computeHeavy);
t.join();

// 协程：IO 等待时挂起，让出线程
Task fetchUrl() {
    auto resp = co_await httpGet("https://example.com");  // 等待期间线程可干别的
    co_return parse(resp);
}

#include <shared_mutex>
#include <mutex>

std::shared_mutex rw_mtx;
int data = 0;

int read() {
    std::shared_lock<std::shared_mutex> lock(rw_mtx);  // 多读并发
    return data;
}
void write(int v) {
    std::unique_lock<std::shared_mutex> lock(rw_mtx);  // 写独占
    data = v;
}

#include <atomic>

std::atomic<int> v{0};

void cas_demo() {
    int expected = 0;
    // 若 v == expected，则 v = 1 并返回 true；否则把 expected 更新为 v 的当前值
    bool ok = v.compare_exchange_strong(expected, 1);
}

// 无锁自增：CAS 循环（等价于 fetch_add 的思路）
void lockfree_inc(std::atomic<int>& x) {
    int old = x.load(std::memory_order_relaxed);
    while (!x.compare_exchange_weak(old, old + 1)) {
        // 失败说明 x 被别的线程改了，old 已被更新，重试
    }
}

线程1: 读 top == A
线程2: 弹出 A（top → B → 改回 A？）   // 中间经历 B
线程1: CAS(top, A, ...) 成功，但栈结构已破坏

#include <atomic>
#include <cstdint>

// tagged pointer：把版本号打包进指针的高位（或结构体）
struct TaggedPtr {
    void* ptr;
    uint64_t tag;   // 每次修改 +1
};

// 用 128 位 CAS（需 is_always_lock_free 支持）同时比较 ptr + tag
// 简化示意：把 tag 和 ptr 合并成一个原子值再 compare_exchange

#include <shared_mutex>
#include <mutex>
#include <vector>

class Cache {
public:
    int read(int key) {
        std::shared_lock<std::shared_mutex> lock(mtx_);   // 共享锁：多读并发
        return data_.at(key);
    }
    void write(int key, int val) {
        std::unique_lock<std::shared_mutex> lock(mtx_);   // 独占锁：写互斥
        data_[key] = val;
    }
private:
    std::shared_mutex mtx_;
    std::vector<int> data_;
};

#include <atomic>
#include <mutex>

class Singleton {
public:
    static Singleton* instance() {
        Singleton* p = ptr_.load(std::memory_order_acquire);   // 1. acquire 读
        if (p == nullptr) {                                    // 第一次检查
            std::lock_guard<std::mutex> lock(mtx_);
            p = ptr_.load(std::memory_order_relaxed);          // 2. 锁内再查
            if (p == nullptr) {                                // 第二次检查
                p = new Singleton();
                ptr_.store(p, std::memory_order_release);      // 3. release 写
            }
        }
        return p;
    }
private:
    Singleton() = default;
    static std::atomic<Singleton*> ptr_;
    static std::mutex mtx_;
};
std::atomic<Singleton*> Singleton::ptr_{nullptr};
std::mutex Singleton::mtx_;

class Singleton {
public:
    static Singleton& instance() {
        static Singleton inst;   // C++11 magic static：线程安全初始化
        return inst;
    }
private:
    Singleton() = default;
};

#include <atomic>
#include <thread>
#include <string>
#include <cassert>

std::atomic<bool> ready{false};
std::string payload;

void producer() {
    payload = "done";                                  // (1) 普通写
    ready.store(true, std::memory_order_release);      // (2) release 发布
}

void consumer() {
    while (!ready.load(std::memory_order_acquire)) {}  // (3) acquire 获取
    assert(payload == "done");                         // (4) 一定能看到 (1)
}

struct Bad {
    int a;   // 线程 1 频繁写 a
    int b;   // 线程 2 频繁写 b —— a、b 大概率在同一缓存行
};
// 两个线程写 a、b 会互相失效缓存行，性能暴跌

#include <atomic>
#include <cstddef>

// 方案 1：alignas 对齐到 64 字节缓存行
struct alignas(64) Aligned {
    std::atomic<int> value{0};
    // 一个缓存行只放这一个变量
};

// 方案 2：手动 padding 填满一个缓存行
struct Padded {
    std::atomic<int> value{0};
    char pad[64 - sizeof(int)];   // 补齐到 64 字节，隔开相邻变量
};

// 方案 3（C++17）：硬件破坏性大小，跨平台更稳
struct alignas(std::hardware_destructive_interference_size) Opt {
    std::atomic<int> value{0};
};

#include <atomic>

class SpinLock {
public:
    void lock() {
        while (flag_.test_and_set(std::memory_order_acquire)) {
            // 忙等：持有时间极短时，空转比睡眠切换更划算
        }
    }
    void unlock() {
        flag_.clear(std::memory_order_release);
    }
private:
    std::atomic_flag flag_ = ATOMIC_FLAG_INIT;
};

#include <queue>
#include <mutex>
#include <condition_variable>
#include <optional>

template <typename T>
class ThreadSafeQueue {
public:
    // 入队：加锁入队，再通知一个等待者
    void push(T value) {
        {
            std::lock_guard<std::mutex> lk(mtx_);
            q_.push(std::move(value));
        }
        cv_.notify_one();
    }

    // 出队：队列空则阻塞，谓词 + 循环防虚假唤醒
    T pop() {
        std::unique_lock<std::mutex> lk(mtx_);
        cv_.wait(lk, [this] { return !q_.empty(); });
        T value = std::move(q_.front());
        q_.pop();
        return value;
    }

    // 非阻塞出队：空则返回 nullopt
    std::optional<T> try_pop() {
        std::lock_guard<std::mutex> lk(mtx_);
        if (q_.empty()) return std::nullopt;
        T value = std::move(q_.front());
        q_.pop();
        return value;
    }

    bool empty() const {
        std::lock_guard<std::mutex> lk(mtx_);
        return q_.empty();
    }

private:
    mutable std::mutex mtx_;
    std::condition_variable cv_;
    std::queue<T> q_;
};

#include <atomic>
#include <memory>

template <typename T>
class MSQueue {
    struct Node {
        T data;
        std::atomic<Node*> next{nullptr};
    };
    std::atomic<Node*> head_;
    std::atomic<Node*> tail_;
    // head_ 始终指向哨兵节点，tail_ 指向最后一个节点（或倒数第二个，允许临时滞后）
public:
    void push(T val) {
        Node* n = new Node{val, nullptr};
        Node* t = tail_.load();
        // CAS 循环把 n 挂到 tail
        while (!t->next.compare_exchange_weak(t->next.load(), n)) { /* 重试 */ }
        tail_.compare_exchange_weak(t, n);   // 移动 tail（允许失败，别的线程会补）
    }
};

#include <latch>
#include <thread>
#include <vector>

void demo_latch() {
    const int N = 3;
    std::latch ready(N);               // 计数 N
    std::vector<std::thread> ts;
    for (int i = 0; i < N; ++i)
        ts.emplace_back([&] { ready.count_down(); });
    ready.wait();                      // 阻塞直到计数归零
    for (auto& t : ts) t.join();
}

#include <barrier>
#include <thread>
#include <vector>

void demo_barrier() {
    const int N = 3;
    int phase = 0;
    std::barrier sync(N, [&]() noexcept { ++phase; });   // 每轮到齐执行完成函数

    auto worker = [&](int id) {
        for (int r = 0; r < 2; ++r) {
            sync.arrive_and_wait();   // 每轮都在这等所有线程到齐
        }
    };
    std::vector<std::thread> ts;
    for (int i = 0; i < N; ++i) ts.emplace_back(worker, i);
    for (auto& t : ts) t.join();
    // phase == 2
}

#include <semaphore>
#include <thread>

void demo_semaphore() {
    std::counting_semaphore<3> sem(3);   // 最多 3 个并发

    auto worker = [&] {
        sem.acquire();     // 计数 -1；计数为 0 则阻塞
        // 临界区（最多 3 个线程同时进入）
        sem.release();     // 计数 +1
    };
}

#include <coroutine>
#include <iostream>

struct Task {
    struct promise_type {
        Task get_return_object() {
            return {std::coroutine_handle<promise_type>::from_promise(*this)};
        }
        std::suspend_always initial_suspend() { return {}; }   // 启动即挂起
        std::suspend_always final_suspend() noexcept { return {}; }
        void return_void() {}
        void unhandled_exception() {}
    };
    std::coroutine_handle<promise_type> h;
    void resume() { h.resume(); }
};

Task coro() {
    std::cout << "A\n";
    co_await std::suspend_always{};   // 显式挂起点
    std::cout << "B\n";
    co_return;
}