2.1、线程间共享数据问题

线程间共享数据的问题

当多个线程共享同一块数据时，问题的本质在于对共享数据修改的不确定性。

在单线程程序中，数据修改是线性的、可预测的。但在多线程环境下，不同线程对共享数据的读写操作可能以任意顺序交错执行，导致以下关键问题：

• • 数据不一致性：当一个线程正在修改数据时，另一个线程可能读取到处于"中间状态"的数据。
• • 数据竞争：当多个线程同时访问同一内存位置，且至少有一个线程在进行写操作时，就会发生数据竞争，这会导致未定义行为。

让我们看一个简单例子：

#include <iostream>
#include <thread>
#include <vector>

int counter = 0;

void increment() {
    for (int i = 0; i < 100000; ++i) {
        ++counter; // 问题点：多线程同时修改
    }
}

int main() {
    std::vector<std::thread> threads;
    for (int i = 0; i < 5; ++i) {
        threads.push_back(std::thread(increment));
    }
    
    for (auto& t : threads) {
        t.join();
    }
    
    std::cout << "计数器值: " << counter << std::endl; // 理论值应为500000
    return 0;
}

你可能认为最终结果应该是 500000（5个线程，每个线程递增100000次），但实际运行时，你很可能会得到一个小于500000的值。这就是线程间共享数据导致的问题。

条件竞争是什么？

条件竞争（Race Condition）是指程序的执行结果依赖于多线程执行的精确时序，而这种时序是不可预测的。

条件竞争需要满足三个关键条件：

• • 并发：存在多个执行流（线程或进程）同时运行
• • 共享对象：这些执行流访问同一个共享资源
• • 改变对象：至少有一个执行流会修改共享资源的状态

所谓"恶性条件竞争"，是指那些会导致程序行为不正确的条件竞争。这种问题特别难调试，因为：

• • 问题出现的概率可能很低
• • 在调试模式下可能完全消失（因为执行时间被改变）
• • 在负载较高时更容易出现

让我们分析上面计数器例子中的条件竞争：表面上看 ++counter 是一个简单操作，但实际上它涉及三个步骤：读取值、增加值、写回值。当多个线程同时执行这些步骤时，问题就出现了：

线程1读取counter值为42
线程2读取counter值为42
线程1将counter增加到43
线程1写回counter值43
线程2将counter增加到43（而不是44！）
线程2写回counter值43

结果是两个线程各执行一次递增操作，但计数器只增加了1而不是2。

防止恶性条件竞争

防止恶性条件竞争的核心思想是：确保在同一时刻只有一个线程能访问共享数据。C++提供了多种机制来实现这一点：

1. 1. 使用互斥量（Mutex）
   最常用的同步原语是互斥量，它能锁定共享数据，确保一次只有一个线程能访问：

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

int counter = 0;
std::mutex counter_mutex; // 互斥量保护counter

void increment() {
    for (int i = 0; i < 100000; ++i) {
        counter_mutex.lock();   // 加锁
        ++counter;              // 安全修改
        counter_mutex.unlock(); // 解锁
    }
}

int main() {
    std::vector<std::thread> threads;
    for (int i = 0; i < 5; ++i) {
        threads.push_back(std::thread(increment));
    }
    
    for (auto& t : threads) {
        t.join();
    }
    
    std::cout << "计数器值: " << counter << std::endl; // 现在会是500000
    return 0;
}

使用互斥量时，需要注意以下几点：

• • lock() 与 unlock() 必须成对匹配
• • 必须使用同一个互斥量进行加锁和解锁
• • 需要选择正确的临界区（被保护的代码区域）

1. 2. 使用RAII风格的锁管理
   直接使用 lock() 和 unlock() 容易出错，更好的做法是使用 std::lock_guard 或 std::unique_lock 进行自动锁管理：

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

int counter = 0;
std::mutex counter_mutex;

void increment() {
    for (int i = 0; i < 100000; ++i) {
        std::lock_guard<std::mutex> lock(counter_mutex); // 自动管理锁的生命周期
        ++counter;
    } // 作用域结束时自动解锁
}

int main() {
    // 与前例相同
}

使用 std::lock_guard 可以确保锁在作用域结束时自动释放，避免因忘记解锁导致的死锁问题。

1. 3. 使用原子操作
   对于简单的数据类型，可以使用原子操作来避免条件竞争：

#include <iostream>
#include <thread>
#include <vector>
#include <atomic>

std::atomic<int> counter(0); // 原子类型

void increment() {
    for (int i = 0; i < 100000; ++i) {
        ++counter; // 原子递增，无需互斥量
    }
}

int main() {
    // 与前例相同
}

原子操作在底层硬件上保证了操作的原子性，是一种"无锁"的同步方式，适合简单数据类型的并发访问。

1. 4. 设计并发安全的数据结构
   让我们设计一个更复杂的例子：线程安全的消息队列：

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

class MessageQueue {
public:
    // 向队列添加消息
    void addMessage(int message) {
        std::lock_guard<std::mutex> lock(m_mutex);
        m_messages.push_back(message);
    }
    
    // 从队列取出消息，如果队列为空返回false
    bool getMessage(int& message) {
        std::lock_guard<std::mutex> lock(m_mutex);
        if (m_messages.empty()) {
            return false;
        }
        
        message = m_messages.front();
        m_messages.pop_front();
        return true;
    }
    
    // 检查队列是否为空
    bool isEmpty() {
        std::lock_guard<std::mutex> lock(m_mutex);
        return m_messages.empty();
    }
    
private:
    std::list<int> m_messages;
    std::mutex m_mutex;
};

// 生产者线程
void producer(MessageQueue& queue) {
    for (int i = 0; i < 100; ++i) {
        queue.addMessage(i);
        std::cout << "生产消息: " << i << std::endl;
        std::this_thread::sleep_for(std::chrono::milliseconds(10)); // 模拟处理时间
    }
}

// 消费者线程
void consumer(MessageQueue& queue, int id) {
    for (int i = 0; i < 50; ++i) {
        int message;
        if (queue.getMessage(message)) {
            std::cout << "消费者 " << id << " 处理消息: " << message << std::endl;
        } else {
            std::cout << "消费者 " << id << " 找不到消息，队列为空" << std::endl;
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(20)); // 模拟处理时间
    }
}

int main() {
    MessageQueue messageQueue;
    
    std::thread producerThread(producer, std::ref(messageQueue));
    std::thread consumerThread1(consumer, std::ref(messageQueue), 1);
    std::thread consumerThread2(consumer, std::ref(messageQueue), 2);
    
    producerThread.join();
    consumerThread1.join();
    consumerThread2.join();
    
    return 0;
}

这个例子展示了如何设计一个线程安全的数据结构，通过互斥量保护共享数据，防止恶性条件竞争。

总结

正确使用可以提高程序性能，但处理不当就会导致难以调试的条件竞争问题。我们应当遵循以下原则：

• • 最小化共享：尽可能减少线程间共享的数据
• • 明确保护：对必须共享的数据使用适当的同步机制
• • 适当粒度：锁的粒度要合适，过大影响性能，过小无法保护完整的原子操作
• • 避免死锁：设计锁的获取顺序，避免循环等待

随着C++11及之后标准的发展，线程库提供了越来越丰富的同步工具，但技术本身只是手段，真正的挑战在于设计思想。良好的并发程序不仅仅是避免了错误，更在于它的可维护性、可扩展性和性能。
本文首发于【讳疾忌医-note】公众号，未经授权，不得转载。
（加入我的知识星球，免费获取账号，解锁所有文章。）