多线程编程中的共享数据与线程安全
多线程可以让程序同时处理多个任务，但线程之间一旦访问同一份数据，就必须回答一个问题：当多个线程同时读写它时，结果是否仍然正确？这正是共享数据与线程安全的核心。
线程安全不是给变量加一把锁就结束了。锁的范围、数据的所有权、操作之间的关系、对象的生命周期，以及线程如何退出，都会影响程序是否正确。本文以 C++17 为例，逐步说明常见做法和容易忽略的边界。示例可使用支持 C++17 的编译器，例如 GCC 或 Clang，并通过 -pthread 链接线程库：
g++ -std=c++17 -O2 -Wall -Wextra -pthread example.cpp -o example一、什么是共享数据与数据竞争
如果多个线程都能访问同一对象，这个对象就是共享数据。访问不只有修改：读写同一对象、遍历容器时另一个线程修改容器、读取对象字段时另一个线程销毁对象，都可能构成并发访问问题。
在 C++ 内存模型中，如果两个线程对同一个内存位置进行冲突访问（例如至少有一次写入），而这些访问之间没有满足要求的同步关系，就形成数据竞争。程序一旦发生数据竞争，行为未定义；它不只是“偶尔读到旧值”，也不能依赖某个平台上观察到的结果。
需要区分两个概念：
- 数据竞争是语言层面的未定义行为，必须避免。
- 竞态条件是更广义的逻辑错误：程序结果取决于线程执行时序。即使每次内存访问都受锁保护，多个操作组合起来仍可能有竞态条件。
例如，“检查余额”和“扣除余额”分别加锁，并不一定能保证整个转账操作正确。线程安全通常要从完整的不变量和操作边界出发，而不是只看某一行代码。
二、先确定共享数据的所有权
在引入锁之前，先问：这份数据是否真的需要共享？最容易推理的并发设计，往往是让数据只归一个线程所有，再通过消息或任务把工作交给它。数据不共享，通常就不需要为它设计锁。
常见选择可以按复杂度从低到高考虑：
- 线程私有数据：每个线程持有自己的局部对象，最后再汇总结果。
- 不可变数据：初始化后不再修改，多个线程只读。要确保对象发布之后确实不会再通过其他别名修改。
- 互斥锁保护：适合多个字段共同维持不变量、需要执行较复杂操作的共享状态。
- 原子变量：适合单个简单数值或标志位的并发读写；复杂状态通常需要额外设计。
- 消息队列或任务队列：让线程通过发送消息交换数据，减少直接共享可变状态。
不要因为“读操作比较简单”就默认并发读取总是安全。如果另一个线程可能同时写入或销毁对象，仍需同步。标准库对特定对象和操作规定了并发使用规则，但不能据此推断所有容器、迭代器和对象组合都安全。
三、用互斥锁保护一组相关状态
std::mutex 用于互斥访问：同一时刻只有一个线程能够持有锁。实际代码通常使用 std::lock_guard 或 std::unique_lock 管理锁，而不是手动调用 lock() 和 unlock()。RAII 管理方式能在正常返回、异常退出等路径上自动释放锁。
下面的示例用互斥锁保护余额。余额的读取和修改都通过成员函数完成，不暴露可绕过锁的引用：
#include <iostream>
#include <mutex>
class Account {
public:
explicit Account(long long initial_balance)
: balance_(initial_balance) {}
bool withdraw(long long amount) {
if (amount <= 0) {
return false;
}
std::lock_guard<std::mutex> lock(mutex_);
if (balance_ < amount) {
return false;
}
balance_ -= amount;
return true;
}
long long balance() const {
std::lock_guard<std::mutex> lock(mutex_);
return balance_;
}
private:
mutable std::mutex mutex_;
long long balance_;
};
int main() {
Account account(1000);
account.withdraw(250);
std::cout << account.balance() << '\n';
}balance() 是 const 成员函数，但仍需要锁，因此互斥量声明为 mutable。这里的锁保护的是余额本身及“余额不能被扣成负数”这一操作约束，而不只是某一次赋值。
这个示例只演示单个账户。若要实现两个账户之间的转账，必须在同一事务边界内保护扣款和入账。若每个账户各有一把锁，两个线程以相反顺序锁定账户就可能死锁。C++17 的 std::scoped_lock 可以一次性锁定多个互斥量，并处理加锁顺序问题：
#include <mutex>
class Account {
public:
explicit Account(long long initial_balance)
: balance_(initial_balance) {}
static bool transfer(Account& from, Account& to, long long amount) {
if (amount <= 0 || &from == &to) {
return false;
}
std::scoped_lock lock(from.mutex_, to.mutex_);
if (from.balance_ < amount) {
return false;
}
from.balance_ -= amount;
to.balance_ += amount;
return true;
}
private:
std::mutex mutex_;
long long balance_;
friend class Account;
};
int main() {
Account a(1000);
Account b(200);
Account::transfer(a, b, 150);
}示例假设余额和转账金额不会超出 long long 的表示范围。真实金融或计费系统还要定义货币精度、溢出检查、失败回滚和持久化事务等规则；使用互斥锁本身并不能提供这些业务保证。
四、锁的范围、锁顺序与死锁
持锁时间应尽量短，但也不能短到破坏操作原子性。常见做法是：在锁内检查和修改共享状态，将耗时计算、网络请求、文件操作等放到锁外。尤其要避免持锁调用未知回调，因为回调可能再次尝试获取同一把锁，或者等待另一个需要该锁的线程。
多个锁交错使用时要建立一致的顺序。例如系统规定总是先锁编号较小的对象，再锁编号较大的对象。另一种方式是使用 std::scoped_lock 同时获取多个锁。无论采用哪种方式，都要检查所有调用路径是否遵守规则。
递归互斥量 std::recursive_mutex 允许同一线程重复获取同一互斥量，但它并不会自动修复锁的设计问题。通常应先厘清调用层级和锁的所有权，再决定是否确实需要递归锁。
还要注意对象生命周期：持有指向某对象的指针，并不意味着该对象在另一个线程使用期间一定存在。若对象可能并发销毁，需要通过明确的所有权方案保证生命周期，例如由管理线程控制退出与销毁顺序，或使用合适的智能指针共享所有权。共享所有权解决的是对象存活问题，不等于对象内部状态自动线程安全。
五、原子变量：适合简单的独立状态
原子类型让针对该原子对象的特定操作以原子方式执行。它适用于计数器、简单状态标志等场景。下面使用原子计数器，让多个线程分别累加：
#include <atomic>
#include <iostream>
#include <thread>
#include <vector>
int main() {
std::atomic<int> counter{0};
std::vector<std::thread> threads;
for (int i = 0; i < 4; ++i) {
threads.emplace_back([&counter] {
for (int n = 0; n < 100000; ++n) {
counter.fetch_add(1, std::memory_order_relaxed);
}
});
}
for (auto& thread : threads) {
thread.join();
}
std::cout << counter.load() << '\n';
}正常运行时，示例输出 400000。每次 fetch_add 都是原子操作，所以不会出现普通自增的读改写冲突。这里使用 memory_order_relaxed，因为这个计数器只用于统计，不承担发布其他数据或建立线程间先后关系的职责。原子性不等于所有操作都有同样的同步语义。
不熟悉内存序时，可以先使用默认的顺序一致性操作，例如 counter.fetch_add(1) 或 counter.load()。默认顺序一致性通常更容易理解，但并不自动修复算法上的逻辑问题。只有在有明确证明和性能依据时，才考虑更弱的内存序。
特别要记住：原子变量不会让一串操作自动成为一个原子事务。下面这种“先检查再更新”不能仅因为变量是原子的，就视为不可分割的整体：
if (count.load() < limit) {
count.fetch_add(1);
}两个线程可能同时读到小于 limit 的值，之后都执行加一，导致超过上限。若必须保证“检查并增加”整体不可被打断，可用互斥锁保护整段逻辑，或仔细设计基于比较并交换的循环，并证明边界条件正确。
原子操作也不保证一定由硬件无锁实现。标准库提供的原子类型是否无锁，取决于类型和平台；可以用 is_lock_free() 查询，但“无锁”并不等于更快、更公平，也不意味着程序不存在饥饿或逻辑错误。
六、条件变量与生产者—消费者队列
线程需要等待某个条件成立时，不应通常采用反复检查变量的忙等循环。std::condition_variable 允许线程在条件不满足时休眠，并在状态变化后被唤醒。条件变量必须和互斥量配合使用，等待时要用谓词检查条件，因为唤醒可能是虚假的，也可能有多个线程竞争同一资源。
下面是一个可运行的有界队列。队列满时，生产者等待；队列空时，消费者等待。调用 close() 后不再接受新数据，消费者会在队列清空后退出：
#include <condition_variable>
#include <cstddef>
#include <deque>
#include <mutex>
#include <optional>
#include <thread>
#include <iostream>
template <typename T>
class BoundedQueue {
public:
explicit BoundedQueue(std::size_t capacity)
: capacity_(capacity) {
if (capacity_ == 0) {
throw std::invalid_argument("capacity must be positive");
}
}
bool push(T value) {
std::unique_lock<std::mutex> lock(mutex_);
not_full_.wait(lock, [this] {
return closed_ || queue_.size() < capacity_;
});
if (closed_) {
return false;
}
queue_.push_back(std::move(value));
not_empty_.notify_one();
return true;
}
std::optional<T> pop() {
std::unique_lock<std::mutex> lock(mutex_);
not_empty_.wait(lock, [this] {
return closed_ || !queue_.empty();
});
if (queue_.empty()) {
return std::nullopt;
}
T value = std::move(queue_.front());
queue_.pop_front();
not_full_.notify_one();
return value;
}
void close() {
{
std::lock_guard<std::mutex> lock(mutex_);
closed_ = true;
}
not_empty_.notify_all();
not_full_.notify_all();
}
private:
const std::size_t capacity_;
std::mutex mutex_;
std::condition_variable not_empty_;
std::condition_variable not_full_;
std::deque<T> queue_;
bool closed_ = false;
};
int main() {
BoundedQueue<int> queue(8);
std::thread producer([&queue] {
for (int i = 1; i <= 10; ++i) {
if (!queue.push(i)) {
return;
}
}
queue.close();
});
std::thread consumer([&queue] {
while (auto value = queue.pop()) {
std::cout << *value << '\n';
}
});
producer.join();
consumer.join();
}这段代码还需要包含 <stdexcept> 和 <utility>，以提供 std::invalid_argument 与 std::move。完整编译时请把它们加入头文件区：
#include <stdexcept>
#include <utility>队列容量必须大于零；类型 T 需要满足容器存储和移动操作所要求的条件。示例由生产者在发送结束后关闭队列，因此消费者能退出。若多个生产者同时工作，应由统一的协调者在所有生产者完成后关闭队列；否则某个生产者提前关闭，其他生产者的后续写入会失败。线程退出协议是队列设计的一部分，不能只依赖通知消费者“有数据”。
七、常见错误及其原因
1. 认为简单自增天然安全
普通的 counter++ 通常包含读取、计算和写回，不是不可分割操作。多个线程同时执行可能丢失更新。计数器应使用原子类型，或由互斥锁保护。
2. 只保护写入，不保护读取
若一个线程写、另一个线程在没有同步的情况下读，仍可能构成数据竞争。共享状态的所有访问路径都要遵循同一同步约定，不能只给“明显的修改语句”加锁。
3. 锁住容器，却把内部引用带到锁外
例如在锁内取得元素引用，解锁后再使用；此时另一个线程可能修改或重分配容器，引用就可能失效。应在锁内完成使用，或者返回值副本，或者设计明确的所有权和生命周期机制。
4. 以为两个安全函数的组合也安全
单独线程安全的“查询”和“更新”函数，组合调用时可能被其他线程插入操作。需要检查的是完整业务动作，例如检查并扣款、查找并插入、读取版本并提交更新。
5. 持锁等待另一个线程
如果当前线程持有互斥量并等待另一个线程完成，而另一个线程需要同一互斥量才能继续，就会形成死锁。不要在锁内无审慎设计地调用 join()、等待条件变量之外的阻塞操作，或执行可能回调当前对象的代码。
6. 线程对象析构前没有妥善处理
对仍然可连接的 std::thread 对象直接析构会调用 std::terminate。要明确执行 join()，或在确认适用时调用 detach()。分离线程会继续运行，必须保证它访问的对象不会提前销毁，也要考虑如何报告错误和控制结束。许多场景更适合用 std::jthread（C++20）自动管理连接和停止请求，但停止仍需在线程函数中配合处理。
7. 用 volatile 代替同步
C++ 的 volatile 主要用于特定的外部可观察访问需求，不提供线程间互斥，也不建立原子操作或线程同步关系。并发共享状态应使用互斥量、原子类型或其他符合内存模型的机制。
八、性能与边界：锁不是越多越好
互斥量可以使程序正确，但会带来竞争和等待。把所有状态塞进同一把全局锁，可能让本可并行的任务串行执行；为每个小字段各建一把锁，则会提高锁管理与死锁推理复杂度。通常先按不变量和操作边界划分锁，再通过测量寻找真正的瓶颈。
线程数量没有适用于所有程序的固定上限。可创建线程数受操作系统、资源限制、栈大小、运行时和程序负载影响。线程数超过硬件并发能力后，频繁切换可能降低吞吐量；但 I/O 密集型任务和计算密集型任务的合适并发度也不同。应根据任务类型、部署环境和实测结果确定，而不是把某个经验数字当成标准保证。
标准库互斥量也不承诺公平调度；等待线程可能长时间得不到锁。原子操作的具体性能同样依赖平台和争用情况。性能优化不能以未经证明的内存序、无锁数据结构或缩小锁范围为名破坏正确性。
九、线程安全与安全用途的边界
线程安全解决的是并发访问下的数据一致性和程序行为问题，不等于访问控制或信息安全。加锁不会自动验证用户权限、保护网络协议、加密敏感数据，也不能阻止拒绝服务。互斥量通常只协调同一进程内的线程，不能替代跨进程锁、数据库事务或分布式协调机制。
用于安全敏感场景时，还应检查锁的范围是否可能被攻击者控制的耗时操作拖长，是否存在锁顺序导致的拒绝服务风险，是否会在日志或错误消息中暴露敏感信息。不要在持锁状态下等待不可信输入或远程响应。对密钥等敏感数据，线程安全也不会自动保证内存清零、访问隔离或安全销毁。
十、实用检查清单
- 能否让数据归单一线程所有，或改为传递不可变数据？
- 共享对象的所有读写是否遵循同一同步规则？
- 锁保护的是完整的不变量和业务操作，还是只保护了零散语句？
- 是否有引用、指针、迭代器或回调在锁外继续访问共享对象？
- 多把锁是否有稳定顺序，等待和回调是否可能造成死锁？
- 所有线程是否有明确的启动、停止、通知、异常处理和连接流程？
- 原子变量是否只承担它实际能够保证的操作，内存序是否有清楚理由？
- 边界值、对象生命周期和资源上限是否经过检查？
总结来说，线程安全首先是状态所有权与操作边界的设计问题。简单独立的计数或标志可以考虑原子变量；多个字段共同维护业务约束时，互斥锁通常更容易验证；需要等待数据或容量变化时，使用条件变量和明确的退出协议；能避免共享可变状态时，则优先避免。把这些边界设计清楚，再谈优化，通常比事后追查偶发的数据竞争和死锁更省力。
© 版权声明
文章版权归作者所有，未经允许请勿转载。
