C++ `std::atomic` 原子操作：无锁编程的基石与内存顺序

好的，没问题。

各位观众，各位朋友，大家好！今天我们来聊聊C++中一个非常重要的概念，也是让很多程序员“闻风丧胆”但又不得不面对的家伙：std::atomic原子操作。为什么说它重要呢？因为它是实现无锁编程的基石！为什么说它让人害怕呢？因为它涉及到了内存模型和各种顺序，稍不留神就会掉进坑里，程序行为变得诡异莫测。

别担心，今天我们就用最通俗易懂的方式，把std::atomic扒个精光，让大家彻底掌握它。

一、什么是原子操作？

首先，我们要搞清楚什么是原子操作。你可以把原子操作想象成一个“要么全做，要么全不做”的操作。就像原子弹爆炸一样，要么炸得稀巴烂，要么啥事没有，不存在炸一半的情况。

在并发编程中，原子操作保证了对某个变量的访问是排他的，不会被其他线程打断。这意味着，即使多个线程同时修改同一个原子变量，最终的结果也是确定的，不会出现数据竞争（Data Race）。

举个例子，假设我们有一个全局变量count，多个线程同时对它进行count++操作。如果没有原子操作的保护，很可能出现下面的情况：

1. 线程A读取count的值（假设是10）。
2. 线程B读取count的值（也是10）。
3. 线程A将count的值加1，写回内存（count变成11）。
4. 线程B将count的值加1，写回内存（count还是11）。

结果是，两个线程都进行了count++操作，但count只增加了1。这就是数据竞争！

如果count是一个原子变量，那么count++就是一个原子操作，上面的情况就不会发生。

二、std::atomic：C++中的原子卫士

C++11引入了std::atomic模板类，它允许我们创建原子变量，并对它们进行原子操作。

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

std::atomic<int> count(0); // 定义一个原子整型变量

void increment() {
    for (int i = 0; i < 10000; ++i) {
        count++; // 原子递增操作
    }
}

int main() {
    std::vector<std::thread> threads;
    for (int i = 0; i < 10; ++i) {
        threads.emplace_back(increment);
    }

    for (auto& thread : threads) {
        thread.join();
    }

    std::cout << "Count: " << count << std::endl; // 输出结果应该是100000
    return 0;
}

在这个例子中，count是一个std::atomic<int>类型的变量。count++操作是原子递增操作，即使多个线程同时执行，最终count的值也会是100000，不会出现数据竞争。

三、std::atomic提供的原子操作

std::atomic提供了丰富的原子操作，包括：

• load(): 原子读取变量的值。
• store(): 原子写入变量的值。
• exchange(): 原子交换变量的值。
• compare_exchange_weak(): 原子比较并交换变量的值（弱形式）。
• compare_exchange_strong(): 原子比较并交换变量的值（强形式）。
• fetch_add(): 原子加法。
• fetch_sub(): 原子减法。
• fetch_and(): 原子与运算。
• fetch_or(): 原子或运算。
• fetch_xor(): 原子异或运算。

这些操作都可以在不同的内存顺序下执行，我们稍后会详细讨论内存顺序。

四、无锁编程：告别锁的束缚

使用std::atomic可以实现无锁编程，避免了锁的开销和死锁的风险。无锁编程的核心思想是使用原子操作来同步线程，而不是使用锁。

当然，无锁编程并不是万能的。它通常比基于锁的编程更复杂，更容易出错，并且可能需要更多的CPU资源。因此，在选择无锁编程时，一定要仔细权衡其优缺点。

一个简单的无锁队列的例子：

#include <atomic>
#include <memory>
#include <iostream>

template <typename T>
class LockFreeQueue {
private:
    struct Node {
        std::shared_ptr<T> data;
        std::atomic<Node*> next;

        Node(std::shared_ptr<T> data) : data(data), next(nullptr) {}
    };

    std::atomic<Node*> head;
    std::atomic<Node*> tail;

public:
    LockFreeQueue() : head(new Node(nullptr)), tail(head.load()) {}

    void enqueue(std::shared_ptr<T> data) {
        Node* newNode = new Node(data);
        Node* tailNode = tail.load(std::memory_order_relaxed); // relaxed load

        // Loop until we successfully append the new node
        while (true) {
            Node* nextNode = tailNode->next.load(std::memory_order_relaxed); // relaxed load

            if (tailNode != tail.load(std::memory_order_relaxed)) { // another thread changed tail
                tailNode = tail.load(std::memory_order_relaxed);
                continue;
            }

            if (nextNode == nullptr) {
                if (tailNode->next.compare_exchange_weak(nextNode, newNode, std::memory_order_release, std::memory_order_relaxed)) { // release store
                    tail.compare_exchange_strong(tailNode, newNode, std::memory_order_release, std::memory_order_relaxed); // release store
                    return;
                }
            } else {
                tail.compare_exchange_strong(tailNode, nextNode, std::memory_order_release, std::memory_order_relaxed); // release store
            }
        }
    }

    std::shared_ptr<T> dequeue() {
        Node* headNode = head.load(std::memory_order_relaxed); // relaxed load
        Node* nextNode;

        while (true) {
            nextNode = headNode->next.load(std::memory_order_acquire); // acquire load

            if (headNode != head.load(std::memory_order_relaxed)) { // another thread changed head
                headNode = head.load(std::memory_order_relaxed);
                continue;
            }

            if (nextNode == nullptr) {
                return nullptr; // Queue is empty
            }

            if (head.compare_exchange_strong(headNode, nextNode, std::memory_order_release, std::memory_order_relaxed)) { // release store
                std::shared_ptr<T> data = nextNode->data;
                delete headNode;
                return data;
            }
        }
    }
};

int main() {
    LockFreeQueue<int> queue;

    std::thread producer([&]() {
        for (int i = 0; i < 10; ++i) {
            queue.enqueue(std::make_shared<int>(i));
            std::cout << "Enqueued: " << i << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
        }
    });

    std::thread consumer([&]() {
        for (int i = 0; i < 10; ++i) {
            std::shared_ptr<int> value = queue.dequeue();
            if (value) {
                std::cout << "Dequeued: " << *value << std::endl;
            } else {
                std::cout << "Queue is empty" << std::endl;
            }
            std::this_thread::sleep_for(std::chrono::milliseconds(20));
        }
    });

    producer.join();
    consumer.join();

    return 0;
}

五、内存顺序：原子操作的灵魂

现在，我们来聊聊std::atomic中最让人头疼的部分：内存顺序（Memory Order）。内存顺序决定了原子操作对其他线程的可见性。简单来说，就是原子操作的执行顺序和对其他线程的影响。

C++提供了六种内存顺序：

• std::memory_order_relaxed: 最宽松的内存顺序。只保证原子性，不保证顺序性。
• std::memory_order_consume: 消费顺序。用于建立数据依赖关系。
• std::memory_order_acquire: 获取顺序。用于同步线程，保证在读取原子变量之前，所有之前的写入操作都对当前线程可见。
• std::memory_order_release: 释放顺序。用于同步线程，保证在写入原子变量之后，所有之前的写入操作都对其他线程可见。
• std::memory_order_acq_rel: 获取-释放顺序。同时具有获取和释放的特性。
• std::memory_order_seq_cst: 顺序一致性。最强的内存顺序，保证所有线程看到的原子操作顺序都是一致的。

这些内存顺序听起来很抽象，我们来逐一解释一下：

1. std::memory_order_relaxed

   std::memory_order_relaxed只保证原子性，不保证顺序性。这意味着，编译器可以随意地重新排序使用std::memory_order_relaxed的原子操作，只要保证每个操作都是原子执行的即可。

   什么情况下可以使用std::memory_order_relaxed呢？当你只需要保证原子性，而不需要保证顺序性时，就可以使用它。例如，一个简单的计数器：

   #include <atomic>
   #include <iostream>
   #include <thread>
   
   std::atomic<int> counter(0);
   
   void increment() {
       for (int i = 0; i < 100000; ++i) {
           counter.fetch_add(1, std::memory_order_relaxed);
       }
   }
   
   int main() {
       std::thread t1(increment);
       std::thread t2(increment);
   
       t1.join();
       t2.join();
   
       std::cout << "Counter: " << counter << std::endl; // 结果应该是200000
       return 0;
   }

   在这个例子中，我们只需要保证counter的递增操作是原子的，不需要保证递增操作的顺序。因此，我们可以使用std::memory_order_relaxed。

2. std::memory_order_consume、std::memory_order_acquire、std::memory_order_release

   这三种内存顺序通常一起使用，用于建立线程间的同步关系。它们实现了所谓的“happens-before”关系。

   • std::memory_order_release: 当一个线程写入一个原子变量时，使用std::memory_order_release。这保证了在该写入操作之前的所有写入操作，对其他线程可见。

   • std::memory_order_acquire: 当一个线程读取一个原子变量时，使用std::memory_order_acquire。这保证了在该读取操作之后的所有读取操作，都能够看到std::memory_order_release线程写入的值。

   • std::memory_order_consume: 与std::memory_order_acquire类似，但是它只建立数据依赖关系。如果一个线程读取一个原子变量，并且使用该变量的值来访问其他数据，那么可以使用std::memory_order_consume来保证数据依赖关系。

   举个例子，假设我们有一个生产者-消费者模型：

   #include <atomic>
   #include <thread>
   #include <iostream>
   #include <memory>
   
   std::atomic<int*> data;
   std::atomic<bool> ready(false);
   
   void producer() {
       int* ptr = new int(42);
       data.store(ptr, std::memory_order_release);
       ready.store(true, std::memory_order_release);
       std::cout << "Producer: Data ready!" << std::endl;
   }
   
   void consumer() {
       while (!ready.load(std::memory_order_acquire)); // 等待数据准备好
       int* ptr = data.load(std::memory_order_consume); // 消费数据
       if (ptr != nullptr) {
           std::cout << "Consumer: Data received: " << *ptr << std::endl;
           delete ptr;
       }
   }
   
   int main() {
       std::thread t1(producer);
       std::thread t2(consumer);
   
       t1.join();
       t2.join();
   
       return 0;
   }

   在这个例子中，producer线程首先分配内存，然后将指针存储到data原子变量中，并使用std::memory_order_release。然后，它将ready原子变量设置为true，也使用std::memory_order_release。

   consumer线程首先等待ready原子变量变为true，使用std::memory_order_acquire。然后，它读取data原子变量，使用std::memory_order_consume。

   这样，我们就建立了一个happens-before关系：producer线程的所有写入操作，都happens-before consumer线程读取data的值。这意味着，consumer线程一定能够看到producer线程写入的data的值。

   如果我们将consumer中的data.load(std::memory_order_consume)改为data.load(std::memory_order_relaxed)，那么consumer线程可能无法看到producer线程写入的data的值，因为我们没有建立happens-before关系。

3. std::memory_order_acq_rel

   std::memory_order_acq_rel同时具有std::memory_order_acquire和std::memory_order_release的特性。它通常用于修改原子变量的操作，例如compare_exchange_weak和compare_exchange_strong。

   举个例子，假设我们有一个自旋锁：

   #include <atomic>
   #include <thread>
   #include <iostream>
   
   class SpinLock {
   private:
       std::atomic<bool> locked = {false};
   
   public:
       void lock() {
           while (locked.exchange(true, std::memory_order_acq_rel));
       }
   
       void unlock() {
           locked.store(false, std::memory_order_release);
       }
   };
   
   SpinLock lock;
   int counter = 0;
   
   void increment() {
       for (int i = 0; i < 100000; ++i) {
           lock.lock();
           counter++;
           lock.unlock();
       }
   }
   
   int main() {
       std::thread t1(increment);
       std::thread t2(increment);
   
       t1.join();
       t2.join();
   
       std::cout << "Counter: " << counter << std::endl; // 结果应该是200000
       return 0;
   }

   在这个例子中，locked.exchange(true, std::memory_order_acq_rel)原子操作同时具有获取和释放的特性。它保证了在获取锁之前的所有写入操作，对其他线程可见；并且保证了在释放锁之后的所有写入操作，对其他线程可见。

4. std::memory_order_seq_cst

   std::memory_order_seq_cst是最强的内存顺序。它保证所有线程看到的原子操作顺序都是一致的。这意味着，如果一个线程A先执行了原子操作X，然后执行了原子操作Y，那么所有其他线程都必须先看到操作X，然后才能看到操作Y。

   std::memory_order_seq_cst是最容易理解的内存顺序，但也是性能最差的。因为它需要保证所有线程看到的操作顺序都是一致的，这需要在底层进行大量的同步。

   默认情况下，std::atomic的所有操作都使用std::memory_order_seq_cst。

   什么时候可以使用std::memory_order_seq_cst呢？当你需要保证所有线程看到的原子操作顺序都是一致的，并且对性能要求不高时，可以使用它。

六、选择合适的内存顺序

选择合适的内存顺序非常重要。如果选择过于宽松的内存顺序，可能会导致数据竞争和程序错误。如果选择过于严格的内存顺序，可能会导致性能下降。

以下是一些选择内存顺序的建议：

• 如果只需要保证原子性，而不需要保证顺序性，可以使用std::memory_order_relaxed。
• 如果需要建立线程间的同步关系，可以使用std::memory_order_acquire和std::memory_order_release。
• 如果需要同时具有获取和释放的特性，可以使用std::memory_order_acq_rel。
• 如果需要保证所有线程看到的原子操作顺序都是一致的，可以使用std::memory_order_seq_cst。

七、总结

std::atomic是C++中实现无锁编程的重要工具。它提供了丰富的原子操作，并允许我们选择不同的内存顺序。

理解std::atomic和内存顺序需要一定的努力，但这是值得的。掌握了这些知识，你就可以编写出高效、可靠的并发程序。

希望今天的讲解对大家有所帮助！谢谢大家！