C++ Lock-Free 数据结构：环形缓冲区、无锁队列的实现

好的，各位观众老爷们，今天咱来聊聊C++ Lock-Free 数据结构，尤其是环形缓冲区和无锁队列。这俩玩意儿，听起来高大上，实际上就是提升并发性能的利器，用得好了，能让你的程序跑得飞起。

啥是Lock-Free？

首先，咱们得明白什么是Lock-Free。简单来说，传统的锁（mutex, semaphore啥的）在多线程环境下，一个线程拿着锁，其他线程就得等着，这就是阻塞。Lock-Free的意思是，即使一个线程挂掉了，其他线程也能继续执行，不会被阻塞。当然，实现起来也没那么简单，得用到原子操作，也就是CPU保证的最小的操作单元，要么全部完成，要么啥也不做。

为啥要用Lock-Free？

好处多多啊！

避免死锁： 锁用不好容易死锁，Lock-Free就没这烦恼。
提高性能： 减少了线程之间的竞争和上下文切换，尤其是高并发场景。
容错性好： 一个线程挂掉不影响其他线程。

当然，Lock-Free也不是万能的，它也有缺点：

实现复杂： 需要对内存模型、原子操作非常熟悉，容易出错。
调试困难： 并发问题本来就难调试，Lock-Free更是难上加难。
可能活锁： 多个线程都在尝试执行，但谁也无法成功，类似交通堵塞。

环形缓冲区（Ring Buffer）

环形缓冲区，又叫循环缓冲区，Circular Buffer。可以把它想象成一个水管，数据从一头进去，从另一头出来，满了就绕回来。这玩意儿在音频、视频处理、网络通信中很常见。

环形缓冲区实现思路

底层存储： 用数组或者std::vector来存储数据。
读写指针： 两个指针，一个指向下一个要读取的位置（read_index），一个指向下一个要写入的位置（write_index）。
容量控制： 记录缓冲区的容量（capacity）和当前数据量（size）。
环绕： 当读写指针到达缓冲区末尾时，重新回到开头。

Lock-Free环形缓冲区实现

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

template <typename T>
class RingBuffer {
public:
    RingBuffer(size_t capacity) : capacity_(capacity), buffer_(capacity) {}

    bool enqueue(const T& item) {
        size_t current_write_index = write_index_.load(std::memory_order_relaxed);
        size_t next_write_index = (current_write_index + 1) % capacity_;

        // 检查是否满了
        size_t current_read_index = read_index_.load(std::memory_order_acquire);
        if (next_write_index == current_read_index) {
            return false; // 满了
        }

        // 尝试写入
        buffer_[current_write_index] = item;
        write_index_.store(next_write_index, std::memory_order_release);
        return true;
    }

    bool dequeue(T& item) {
        size_t current_read_index = read_index_.load(std::memory_order_relaxed);
        // 检查是否为空
        size_t current_write_index = write_index_.load(std::memory_order_acquire);
        if (current_read_index == current_write_index) {
            return false; // 空了
        }

        // 尝试读取
        item = buffer_[current_read_index];
        size_t next_read_index = (current_read_index + 1) % capacity_;
        read_index_.store(next_read_index, std::memory_order_release);
        return true;
    }

    size_t size() const {
        size_t read = read_index_.load(std::memory_order_acquire);
        size_t write = write_index_.load(std::memory_order_acquire);
        if (write >= read) {
            return write - read;
        } else {
            return capacity_ - (read - write);
        }
    }

    size_t capacity() const {
        return capacity_;
    }

private:
    std::vector<T> buffer_;
    std::atomic<size_t> read_index_{0};
    std::atomic<size_t> write_index_{0};
    size_t capacity_;
};

int main() {
    RingBuffer<int> ring_buffer(10);

    // 生产者线程
    std::thread producer([&]() {
        for (int i = 0; i < 20; ++i) {
            while (!ring_buffer.enqueue(i)) {
                std::this_thread::sleep_for(std::chrono::microseconds(10)); // 缓冲区满了，稍等一下
            }
            std::cout << "Produced: " << i << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(50));
        }
    });

    // 消费者线程
    std::thread consumer([&]() {
        for (int i = 0; i < 20; ++i) {
            int item;
            while (!ring_buffer.dequeue(item)) {
                std::this_thread::sleep_for(std::chrono::microseconds(10)); // 缓冲区空了，稍等一下
            }
            std::cout << "Consumed: " << item << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(70));
        }
    });

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

    return 0;
}

代码解释

std::atomic<size_t> read_index_ 和 std::atomic<size_t> write_index_: 这两个是原子变量，保证了读写指针的原子性操作。
enqueue(): 入队操作，首先检查缓冲区是否已满，满了就返回false。没满就将数据写入，并更新write_index_。
dequeue(): 出队操作，首先检查缓冲区是否为空，空了就返回false。不空就读取数据，并更新read_index_。
std::memory_order_relaxed, std::memory_order_acquire, std::memory_order_release: 这些是内存顺序，用来保证多线程环境下的数据一致性。memory_order_relaxed是最宽松的，只保证原子性，不保证顺序性。memory_order_acquire和memory_order_release配合使用，可以建立happens-before关系，保证数据的可见性。

内存顺序（Memory Order）

这玩意儿有点复杂，但很重要。简单理解就是，编译器和CPU为了优化性能，可能会对指令进行重排序。内存顺序就是用来告诉编译器和CPU，哪些操作不能重排序，哪些操作需要保证可见性。

常用的内存顺序有：

std::memory_order_relaxed: 最宽松的顺序，只保证原子性。
std::memory_order_acquire: 获取锁的语义，保证读取操作发生在其他线程释放锁之后。
std::memory_order_release: 释放锁的语义，保证写入操作发生在其他线程获取锁之前。
std::memory_order_acq_rel: 同时具有acquire和release语义。
std::memory_order_seq_cst: 最强的顺序，保证所有操作都按照代码顺序执行，开销最大。

无锁队列（Lock-Free Queue）

无锁队列也是一种常用的并发数据结构，它允许多个线程同时进行入队和出队操作，而不需要使用锁。

无锁队列实现思路

底层存储： 通常使用链表来实现，因为链表的插入和删除操作比较方便。
头尾指针： 两个原子指针，一个指向队列的头部（head），一个指向队列的尾部（tail）。
CAS操作： 使用Compare-and-Swap (CAS) 原子操作来更新头尾指针，保证并发安全。

Lock-Free队列实现 (使用CAS)

#include <iostream>
#include <atomic>
#include <memory>
#include <thread>
#include <chrono>

template <typename T>
struct Node {
    T data;
    std::atomic<Node<T>*> next;

    Node(const T& data) : data(data), next(nullptr) {}
};

template <typename T>
class LockFreeQueue {
public:
    LockFreeQueue() : head_(new Node<T>({})), tail_(head_.load()) {} // 初始化，head和tail指向一个空节点
    ~LockFreeQueue() {
        Node<T>* current = head_.load();
        while (current != nullptr) {
            Node<T>* next = current->next.load();
            delete current;
            current = next;
        }
    }

    void enqueue(const T& value) {
        Node<T>* new_node = new Node<T>(value);
        Node<T>* tail = tail_.load(std::memory_order_acquire);
        Node<T>* next = nullptr;

        while (true) {
            next = tail->next.load(std::memory_order_acquire);
            if (tail == tail_.load(std::memory_order_acquire)) { // 检查tail是否被其他线程修改过
                if (next == nullptr) {
                    if (tail->next.compare_exchange_weak(next, new_node, std::memory_order_release, std::memory_order_relaxed)) {
                        tail_.compare_exchange_weak(tail, new_node, std::memory_order_release, std::memory_order_relaxed); // 尝试更新tail，允许失败
                        return;
                    }
                } else {
                    tail_.compare_exchange_weak(tail, next, std::memory_order_release, std::memory_order_relaxed); // 帮助其他线程完成tail的更新
                }
            } else {
                tail = tail_.load(std::memory_order_acquire); // 从新读取tail
            }
        }
    }

    bool dequeue(T& value) {
        Node<T>* head = head_.load(std::memory_order_acquire);
        Node<T>* next = nullptr;

        while (true) {
            next = head->next.load(std::memory_order_acquire);
            if (head == head_.load(std::memory_order_acquire)) {
                if (next == nullptr) {
                    return false; // 队列为空
                }

                if (head_.compare_exchange_weak(head, next, std::memory_order_release, std::memory_order_relaxed)) {
                    value = next->data;
                    delete head;
                    return true;
                }
            } else {
                head = head_.load(std::memory_order_acquire); // 从新读取head
            }
        }
    }

private:
    std::atomic<Node<T>*> head_;
    std::atomic<Node<T>*> tail_;
};

int main() {
    LockFreeQueue<int> queue;

    // 生产者线程
    std::thread producer([&]() {
        for (int i = 0; i < 20; ++i) {
            queue.enqueue(i);
            std::cout << "Enqueued: " << i << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(50));
        }
    });

    // 消费者线程
    std::thread consumer([&]() {
        for (int i = 0; i < 20; ++i) {
            int value;
            if (queue.dequeue(value)) {
                std::cout << "Dequeued: " << value << std::endl;
            } else {
                std::cout << "Queue is empty." << std::endl;
                i--; // 重试
            }
            std::this_thread::sleep_for(std::chrono::milliseconds(70));
        }
    });

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

    return 0;
}

代码解释

std::atomic<Node<T>*> head_ 和 std::atomic<Node<T>*> tail_: 原子指针，指向队列的头和尾。
enqueue(): 入队操作，使用CAS操作来更新tail_指针，保证只有一个线程能够成功将新节点添加到队列尾部。
dequeue(): 出队操作，使用CAS操作来更新head_指针，保证只有一个线程能够成功从队列头部删除节点。
ABA问题： CAS操作的一个经典问题。假设一个值A被读取，然后一些操作发生，之后又变回了A。CAS操作会认为这个值没有被改变，然后继续操作。但在并发环境下，这可能会导致问题。解决方案之一是使用版本号，每次修改都增加版本号，这样即使值变回了A，版本号也不一样了。

CAS (Compare-and-Swap)

CAS是一种原子操作，它比较内存中的值与期望值，如果相等，就用新值替换内存中的值。整个过程是原子的，不会被其他线程打断。

C++中可以使用std::atomic::compare_exchange_weak或std::atomic::compare_exchange_strong来实现CAS操作。

compare_exchange_weak: 允许虚假失败（spurious failure），即使内存中的值与期望值相等，也可能返回false。通常在一个循环中使用，直到成功为止。
compare_exchange_strong: 保证只有在内存中的值与期望值相等时才返回true。

选择哪种Lock-Free结构？

特性	环形缓冲区	无锁队列
底层存储	数组或`std::vector`	链表
容量	固定	动态
适用场景	固定大小的数据流，例如音频、视频处理	大小不确定的数据流，例如任务队列
性能	通常更快，因为内存是连续的	插入删除更灵活，但可能涉及更多内存分配和释放
实现复杂度	相对简单	相对复杂，需要处理ABA问题等

总结

Lock-Free数据结构是提升并发性能的有效手段，但实现起来比较复杂，需要对内存模型、原子操作有深入的理解。环形缓冲区适合固定大小的数据流，而无锁队列适合大小不确定的数据流。选择哪种结构取决于具体的应用场景。

好了，今天的分享就到这里。希望大家有所收获，下次再见！记住，并发编程，且行且珍惜！搞不好就秃头了！

发表回复 取消回复

发表回复取消回复