数据共享问题与锁

需要预先学习：

• 多线程编程基础

---

1. 数据竞争：共享数据的问题

多线程共享内存是优势，也是问题的源头。当两个或多个线程同时访问同一内存位置，且至少有一个是写操作，就构成数据竞争（data race）。C++ 标准规定数据竞争是未定义行为（UB）：程序可能输出任何结果，可能崩溃，也可能"碰巧正确"。

看一个经典例子——两个线程各做 100 万次 counter++：

#include <iostream>
#include <thread>

int counter = 0;

void add()
{
    for (int i = 0; i < 1000000; ++i)
    {
        ++counter;      // 数据竞争: 两个线程同时读写 counter
    }
}

int main()
{
    std::thread t1(add);
    std::thread t2(add);
    t1.join();
    t2.join();
    std::cout << "counter = " << counter << std::endl;
    return 0;
}

直觉上结果应该是 2000000，实际运行可能得到：

counter = 1173468
counter = 1329507

每次运行结果都不同。为什么？因为 ++counter 不是一条指令，而是三步：读 counter 到寄存器 → 寄存器 +1 → 写回内存。两个线程的这三步会交错：

线程 1: 读 counter(=5) → 寄存器1 = 6
线程 2: 读 counter(=5) → 寄存器2 = 6    ← 线程1 还没写回, 读到旧值
线程 1: 写回 counter = 6
线程 2: 写回 counter = 6                ← 一次递增丢失了!

这就是丢失更新（lost update）：两次递增只生效了一次。交错方式由内核调度决定，所以每次运行结果不同。

注意：这只是最简单的竞争。真实世界里数据竞争还伴随编译器/CPU 的指令重排、缓存不一致（《原子操作》中讲），表现可能更诡异。任何共享可变数据，都必须用同步机制保护——最基础的就是互斥量。

---

2. 互斥量 std::mutex

互斥量（mutex，mutual exclusion，互斥排除）保证同一时刻只有一个线程能进入临界区。持有锁的线程独占访问权，其他线程在 lock() 处阻塞等待，直到锁被释放。

#include <iostream>
#include <mutex>
#include <thread>

int counter = 0;
std::mutex mtx;              // 全局互斥量

void add()
{
    for (int i = 0; i < 1000000; ++i)
    {
        mtx.lock();          // 加锁: 其他线程在此阻塞
        ++counter;           // 临界区: 同一时刻只有一个线程执行
        mtx.unlock();        // 解锁: 唤醒等待的线程
    }
}

int main()
{
    std::thread t1(add);
    std::thread t2(add);
    t1.join();
    t2.join();
    std::cout << "counter = " << counter << std::endl;   // 稳定输出 2000000
    return 0;
}

现在每次运行都稳定输出 2000000。

std::mutex 的核心成员：

成员函数	行为
lock()	加锁；锁已被占用则阻塞等待
unlock()	解锁
try_lock()	尝试加锁；成功返回 true，锁被占用立即返回 false（不阻塞）

手动 lock/unlock 的问题：如果临界区抛出异常，unlock() 永远不会执行，锁一直被占用，其他线程永久阻塞——死锁。所以裸 lock()/unlock() 只适合绝不会抛异常的临界区，工程上必须用 RAII 包装（lock_guard 等）。

注：std::mutex 不可拷贝、不可移动，所以通常声明为全局对象、类成员，或通过 std::ref 传引用（见《线程传参》）。

---

3. std::lock_guard：RAII 加锁

std::lock_guard 是互斥量的 RAII 包装：构造时加锁，析构时自动解锁。无论临界区是正常返回还是抛异常，析构都会执行，锁必然被释放：

#include <iostream>
#include <mutex>
#include <stdexcept>
#include <thread>

std::mutex mtx;

void risky()
{
    std::lock_guard<std::mutex> guard(mtx);   // 构造: 加锁
    std::cout << "进入临界区" << std::endl;

    if (rand() % 2 == 0)
    {
        throw std::runtime_error("出错了");   // 抛异常也不怕
    }
    // 析构: 自动解锁, 无论是否抛异常
}

int main()
{
    std::thread t1(risky);
    std::thread t2(risky);
    t1.join();
    t2.join();
    std::cout << "两个线程都结束了, 没有死锁" << std::endl;
    return 0;
}

lock_guard 的用法极其简单：构造即加锁，离开作用域即解锁。它是绝大多数场景的默认选择，代价是不能手动解锁、不能延迟加锁——需要这些能力时用 unique_lock。

注：RAII 思想在《智能指针与RAII》中讲过：把资源的获取与释放绑定到对象的构造与析构，锁正是最典型的资源之一。

---

4. std::unique_lock：更灵活的锁

std::unique_lock 在 lock_guard 的基础上提供了更多能力：

• 延迟加锁：构造时不加锁，之后手动 lock()
• 提前解锁：临界区结束前手动 unlock()（如只读阶段后）
• 可移动：锁的所有权可以在 unique_lock 之间转移
• 配合条件变量：condition_variable::wait 必须使用 unique_lock（《多线程编程进阶》）

#include <iostream>
#include <mutex>
#include <thread>

std::mutex mtx;

void work()
{
    std::unique_lock<std::mutex> guard(mtx);   // 构造即加锁（默认行为）

    // ... 需要锁保护的临界区 ...

    guard.unlock();                            // 提前解锁, 后面的操作不需要锁
    std::cout << "锁已释放, 执行不需要锁的操作" << std::endl;
    std::this_thread::sleep_for(std::chrono::milliseconds(10));

    guard.lock();                              // 需要时再次加锁
    std::cout << "重新加锁" << std::endl;
}

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

延迟加锁的典型场景：先做一些不依赖共享数据的准备工作，再进入临界区，缩短持锁时间：

std::unique_lock<std::mutex> guard(mtx, std::defer_lock);   // 延迟加锁
prepare();          // 不持锁, 做准备工作
guard.lock();       // 真正需要时才加锁
use_shared_data();

unique_lock 可移动的特性让"把锁交给另一个函数"成为可能：

std::unique_lock<std::mutex> guard(mtx);
some_function(std::move(guard));   // 锁的所有权转移, 由 some_function 负责解锁

特性	lock_guard	unique_lock
构造加锁	是	是（可延迟）
手动解锁	否	是
再次加锁	否	是
可移动	否	是
配合条件变量	否	是
额外开销	无	极小（多一个状态标志）

注：两者性能几乎无差别，unique_lock 的功能是 lock_guard 的超集。不需要额外能力时用 lock_guard 更直白，需要灵活性时用 unique_lock。

---

5. 特殊互斥量

---

5.1 std::recursive_mutex：可重入

std::recursive_mutex 允许同一线程重复加锁，锁内部维护持有线程和加锁计数，解锁次数与加锁次数匹配才真正释放：

std::recursive_mutex mtx;

void outer()
{
    std::lock_guard<std::recursive_mutex> g(mtx);
    inner();              // inner 又要加同一把锁
}

void inner()
{
    std::lock_guard<std::recursive_mutex> g(mtx);   // 同一线程再次加锁, 合法
}

普通 std::mutex 遇到这种情况会死锁（自己锁自己，永远等不到释放）。recursive_mutex 解决"函数互相调用、都要加同一把锁"的递归场景。

注意：能用 recursive_mutex 的地方往往说明锁的粒度设计有问题——递归加锁掩盖了"临界区边界不清"的事实。优先重构代码（把需要锁的部分抽到不递归的函数），recursive_mutex 是兜底方案。

---

5.2 std::timed_mutex：限时等待

std::timed_mutex 提供带超时的加锁：try_lock_for(duration) 等待指定时长，超时仍未拿到锁返回 false；try_lock_until(time_point) 等待到指定时刻：

#include <iostream>
#include <mutex>
#include <thread>

std::timed_mutex mtx;

void worker(int id)
{
    // 最多等 100ms, 拿不到锁就不等了
    if (mtx.try_lock_for(std::chrono::milliseconds(100)))
    {
        std::cout << "线程 " << id << " 拿到锁" << std::endl;
        std::this_thread::sleep_for(std::chrono::milliseconds(200));
        mtx.unlock();
    }
    else
    {
        std::cout << "线程 " << id << " 超时放弃" << std::endl;
    }
}

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

限时等待适合"拿不到锁就做别的事"的优雅降级场景，也常用于避免死锁（见下文）：等不到锁就放弃重试，而不是无限阻塞。

---

5.3 std::shared_mutex：读写锁

std::shared_mutex（C++17；C++14 中为 std::shared_timed_mutex）区分读与写两种访问：

• 写锁（独占）：std::unique_lock<std::shared_mutex>，同一时刻只有一个线程持有
• 读锁（共享）：std::shared_lock<std::shared_mutex>，多个线程可以同时持有读锁

读写锁的原理：读与读不互斥，读与写互斥，写与写互斥。适合"读多写少"的数据结构——多个读者并行读，只有写者才独占：

#include <iostream>
#include <map>
#include <mutex>
#include <shared_mutex>
#include <thread>

class ThreadSafeCache
{
public:
    int get(int key)
    {
        std::shared_lock<std::shared_mutex> lock(mtx_);   // 读锁: 可多个线程同时持有
        auto it = data_.find(key);
        return it == data_.end() ? -1 : it->second;
    }

    void put(int key, int value)
    {
        std::unique_lock<std::shared_mutex> lock(mtx_);   // 写锁: 独占
        data_[key] = value;
    }

private:
    std::map<int, int> data_;
    std::shared_mutex mtx_;
};

int main()
{
    ThreadSafeCache cache;
    cache.put(1, 100);

    std::thread readers[4];
    for (int i = 0; i < 4; ++i)
    {
        readers[i] = std::thread([&cache] {
            for (int k = 0; k < 1000; ++k)
            {
                cache.get(1);    // 4 个读者可同时读, 互不阻塞
            }
        });
    }
    for (auto& t : readers)
    {
        t.join();
    }
    std::cout << "读取完成, value = " << cache.get(1) << std::endl;
    return 0;
}

注意 std::shared_lock 是 unique_lock 的"共享版"——支持延迟加锁、手动解锁等。只在确认读操作安全（不会修改数据）时使用读锁，否则共享读会把竞争问题放回来。

---

5.4 std::call_once：只初始化一次

std::call_once(flag, func) 保证 func 在多线程环境下只被调用一次，即使多个线程同时调用 call_once。std::once_flag 是配套的标志对象：

#include <iostream>
#include <mutex>
#include <thread>

std::once_flag init_flag;

void init()
{
    std::cout << "初始化只执行一次" << std::endl;
    // 加载配置、连接数据库等
}

void use()
{
    std::call_once(init_flag, init);   // 无论多少线程进来, init 只跑一次
    std::cout << "使用资源" << std::endl;
}

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

初始化只执行一次
使用资源
使用资源
使用资源

call_once 是线程安全的懒初始化标准方案（比"双重检查锁"更可靠），常用于全局单例、资源池的首次创建。

---

6. 死锁

---

6.1 什么是死锁

死锁（deadlock）指多个线程互相等待对方持有的锁，谁也拿不到自己需要的锁，全部永久阻塞。最经典的场景——两个线程按相反顺序锁两把锁：

#include <iostream>
#include <mutex>
#include <thread>

std::mutex mtx_a;
std::mutex mtx_b;

void thread_a()
{
    std::lock_guard<std::mutex> g1(mtx_a);
    std::this_thread::sleep_for(std::chrono::milliseconds(10));   // 制造交错
    std::lock_guard<std::mutex> g2(mtx_b);   // 等待 mtx_b...
}

void thread_b()
{
    std::lock_guard<std::mutex> g1(mtx_b);
    std::this_thread::sleep_for(std::chrono::milliseconds(10));
    std::lock_guard<std::mutex> g2(mtx_a);   // 等待 mtx_a...
}

int main()
{
    std::thread t1(thread_a);
    std::thread t2(thread_b);
    t1.join();     // 两个线程互相等待, 程序永久挂起
    t2.join();
    return 0;
}

时间线：t1 拿到 mtx_a 等 mtx_b，t2 拿到 mtx_b 等 mtx_a——双方都持有一把对方需要的锁，谁也不让。程序卡死，join 永远不会返回。

---

6.2 死锁的四个必要条件

死锁必须同时满足四个条件：

条件	含义
互斥	锁同一时刻只能被一个线程持有
持有并等待	持有至少一把锁，又在等待另一把锁
不可剥夺	锁只能由持有者主动释放
循环等待	存在"A 等 B 的锁，B 等 A 的锁"的等待环

打破任意一个条件即可避免死锁。工程上最实用的是打破循环等待：所有线程按相同顺序加锁。

---

6.3 避免死锁：std::lock 与 std::scoped_lock

需要同时锁多把锁时，std::lock(l1, l2, ...) 可以一次锁住多个互斥量——它内部采用"尝试 + 回退"策略，保证不会死锁（要么全部拿到，要么全部不拿）。配套 std::scoped_lock（C++17）把 std::lock 包装成 RAII，构造时用 std::lock 语义锁所有互斥量，析构时按相反顺序解锁：

#include <iostream>
#include <mutex>
#include <thread>

std::mutex mtx_a;
std::mutex mtx_b;

void thread_a()
{
    std::scoped_lock lock(mtx_a, mtx_b);   // 一次锁两把, 内部保证不死锁
    std::this_thread::sleep_for(std::chrono::milliseconds(10));
    // 使用 a 和 b 保护的资源
}

void thread_b()
{
    std::scoped_lock lock(mtx_b, mtx_a);   // 顺序反了也没关系
    std::this_thread::sleep_for(std::chrono::milliseconds(10));
    // 使用 a 和 b 保护的资源
}

int main()
{
    std::thread t1(thread_a);
    std::thread t2(thread_b);
    t1.join();
    t2.join();
    std::cout << "两个线程都正常结束, 没有死锁" << std::endl;
    return 0;
}

std::scoped_lock 的类模板参数可以推导（C++17 CTAD），std::scoped_lock lock(mtx_a, mtx_b); 会自动推导为 std::scoped_lock<std::mutex, std::mutex>。

注意：std::lock / scoped_lock 解决的是"多把锁"场景。单把锁的死锁（自己锁自己、忘记解锁）它们管不了，要靠 RAII 和 recursive_mutex 解决。

其他避免死锁的辅助手段：

• 按固定顺序加锁（如按地址大小排序），从根上消除循环等待
• 用 try_lock 系列：拿不到就放弃重试，不无限等待
• 缩小持锁范围：尽量只持一把锁、短时间持锁，减少与其他锁的交集
• 避免锁内调用未知代码：锁内调用回调函数、第三方库，可能间接形成等待环

---

7. 锁的粒度

锁的粒度指临界区的大小：一次锁保护的代码量、数据量。粒度选择是性能与正确性的权衡：

• 粗粒度（大锁）：实现简单、不易出错，但并发度低——无关操作也被串行化，多线程优势被抵消
• 细粒度（小锁）：并发度高，但锁多、容易死锁，加解锁开销占比上升

// 粗粒度: 整个函数一把锁, 简单但并发度低
void process_all()
{
    std::lock_guard<std::mutex> g(mtx);
    step1();   // 这些步骤并不都需要锁
    step2();
    step3();
}

// 细粒度: 只有真正访问共享数据的地方加锁
void process_all()
{
    step1();                          // 不需要锁
    {
        std::lock_guard<std::mutex> g(mtx);
        shared_data_ += step2();      // 只有这里需要锁
    }
    step3();                          // 不需要锁
}

实践建议：

• 默认从粗粒度开始，性能实测不足时再细化
• 加锁代码尽量短：只包住真正访问共享数据的语句
• 避免在临界区内做 IO、网络、sleep 等慢操作——它们让其他线程空等
• 临界区越短，持锁时间越短，死锁窗口也越小

注：unique_lock 的手动 unlock() 就是为了支持"提前解锁"的细粒度写法——需要锁的部分用锁，不需要的部分及时释放。

---

8. 用锁保护数据的常见模式

1. 封装数据：锁与数据绑定，不对外暴露裸数据

class ThreadSafeCounter
{
public:
    void increment()
    {
        std::lock_guard<std::mutex> g(mtx_);
        ++value_;
    }

    int get()
    {
        std::lock_guard<std::mutex> g(mtx_);
        return value_;
    }

private:
    int value_ = 0;
    mutable std::mutex mtx_;   // mutable: const 成员函数中也能加锁
};

原则：数据私有、锁私有、所有访问都经过加锁的成员函数。绝不把指向内部数据的裸指针/引用返回给外部——那等于把锁保护的数据重新暴露出去：

class Bad
{
public:
    int* get_data()      // 错误! 外部拿到裸指针后可以绕过锁修改数据
    {
        return &data_;
    }

private:
    int data_;
    std::mutex mtx_;
};

2. 锁保护的对象生命周期：锁能保护数据不被并发修改，但不能保护对象本身不被销毁。如果线程 A 持锁访问对象，线程 B 在锁外 delete 了对象，A 解引用就是悬空访问。这类问题需要 shared_ptr 管理生命周期（《线程传参》），锁只负责同步访问。

3. 何时可以不用锁：单线程独享的数据、只读的共享数据（初始化后不再修改）、由原子操作保护的简单变量（《原子操作》）都不需要锁。锁是同步手段，不是目的——能用更轻量手段解决的竞争，不必上锁。

---

9. 锁的对比总结

锁类型	头文件	特点	典型场景
std::mutex	<mutex>	基础互斥	默认选择
std::recursive_mutex	<mutex>	同一线程可重复加锁	递归调用、函数互相调用
std::timed_mutex	<mutex>	限时等待	超时降级、避免死锁
std::shared_mutex	<shared_mutex>	读写锁，读读并行	读多写少的缓存、配置表
std::once_flag + call_once	<mutex>	只初始化一次	线程安全懒初始化

配套的 RAII 包装：

包装器	能力
std::lock_guard	构造加锁、析构解锁，最简
std::unique_lock	延迟加锁、手动解锁、可移动、配合条件变量
std::shared_lock	共享（读）锁，配合 shared_mutex
std::scoped_lock	一次锁多把互斥量，死锁安全

到这一步，你已经能用锁保护共享数据、避免死锁了。但锁不是同步的唯一手段——原子操作可以在无锁的情况下保护简单数据，条件变量能让线程高效等待事件，这些在接下来的文章中展开。

---

需要进一步学习：

• 多线程编程进阶
• 原子操作

---