C++ 无锁环形缓冲区 (`Disruptor` 模式) 的 C++ 实现与性能分析

哈喽，各位好！

今天我们来聊聊一个高性能的消息传递利器：C++ 无锁环形缓冲区，也就是常说的 Disruptor 模式。这玩意儿在并发编程领域可是个明星，能让你在多线程环境下安全又高效地传递数据，避免各种锁带来的性能损耗。

一、 什么是环形缓冲区？为啥要用无锁的？

想象一下，你有一个固定大小的数组，数据就像流水一样，从一端流入，从另一端流出。当数据到达数组末尾时，它会绕回到数组的开头，就像一个环一样。这就是环形缓冲区。

• 优点： 读写操作简单高效，内存分配固定，避免了频繁的 new 和 delete，适用于高吞吐量的场景。
• 缺点： 容量固定，可能会出现缓冲区满或空的情况，需要合理的控制策略。

那为啥要用无锁呢？因为锁虽然能保证线程安全，但也会带来性能开销，特别是在高并发的情况下，锁的竞争会变得非常激烈，导致线程阻塞，降低整体吞吐量。无锁数据结构则利用原子操作等技术，避免了锁的使用，从而提高并发性能。

二、 Disruptor 模式的核心思想

Disruptor 模式的核心思想是：

1. 预分配环形缓冲区： 预先分配好一块连续的内存空间作为环形缓冲区，避免了动态内存分配带来的开销。
2. 单一写入者： 只有一个线程负责写入数据到环形缓冲区，避免了写入竞争。
3. 多个读取者： 可以有多个线程同时从环形缓冲区读取数据。
4. 序号追踪： 使用序号（Sequence）来追踪读写的位置，通过原子操作来更新序号，保证线程安全。
5. 缓存行填充： 为了避免伪共享（false sharing），在关键变量周围填充缓存行大小的字节，保证每个变量都独占一个缓存行。

三、 C++ 无锁环形缓冲区的实现

下面是一个简单的 C++ 无锁环形缓冲区的实现，为了便于理解，我们先从最基本的功能开始，然后再逐步完善。

#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 write(const T& data) {
        // CAS 操作，尝试将 write_sequence_ + 1 赋值给 write_sequence_
        size_t current_write_sequence = write_sequence_.load(std::memory_order_relaxed);
        size_t next_write_sequence = current_write_sequence + 1;

        if (next_write_sequence - read_sequence_.load(std::memory_order_acquire) > capacity_) {
            // 缓冲区已满
            return false;
        }

        size_t index = next_write_sequence % capacity_;
        buffer_[index] = data;

        // 使用 compare_exchange_weak 实现 CAS 操作
        while (!write_sequence_.compare_exchange_weak(current_write_sequence, next_write_sequence,
                                                      std::memory_order_release, std::memory_order_relaxed)) {
            // CAS 失败，重新读取 write_sequence_ 的值，重试
            // 注意这里不能使用 memory_order_acquire，否则可能导致死锁
            ;
        }
        return true;
    }

    bool read(T& data) {
        size_t current_read_sequence = read_sequence_.load(std::memory_order_relaxed);
        size_t next_read_sequence = current_read_sequence + 1;

        if (next_read_sequence > write_sequence_.load(std::memory_order_acquire)) {
            // 缓冲区为空
            return false;
        }

        size_t index = next_read_sequence % capacity_;
        data = buffer_[index];

        while (!read_sequence_.compare_exchange_weak(current_read_sequence, next_read_sequence,
                                                     std::memory_order_release, std::memory_order_relaxed)) {
            ;
        }

        return true;
    }

private:
    size_t capacity_;
    std::vector<T> buffer_;
    std::atomic<size_t> write_sequence_{0};
    std::atomic<size_t> read_sequence_{0};
};

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

    // 写入线程
    std::thread writer([&]() {
        for (int i = 0; i < 100; ++i) {
            while (!ring_buffer.write(i)) {
                // 缓冲区满，等待
                std::this_thread::sleep_for(std::chrono::milliseconds(1));
            }
            std::cout << "Write: " << i << std::endl;
        }
    });

    // 读取线程
    std::thread reader([&]() {
        int data;
        for (int i = 0; i < 100; ++i) {
            while (!ring_buffer.read(data)) {
                // 缓冲区空，等待
                std::this_thread::sleep_for(std::chrono::milliseconds(1));
            }
            std::cout << "Read: " << data << std::endl;
        }
    });

    writer.join();
    reader.join();

    return 0;
}

代码解释：

• RingBuffer 类： 封装了环形缓冲区的核心逻辑。
• capacity_： 缓冲区容量。
• buffer_： 存储数据的 std::vector。
• write_sequence_ 和 read_sequence_： 原子变量，分别记录写入和读取的位置。
• write() 方法： 写入数据到缓冲区。
  • 使用 compare_exchange_weak 实现 CAS 操作，原子性地更新 write_sequence_。
  • 如果缓冲区已满，则返回 false，表示写入失败。
• read() 方法： 从缓冲区读取数据。
  • 使用 compare_exchange_weak 实现 CAS 操作，原子性地更新 read_sequence_。
  • 如果缓冲区为空，则返回 false，表示读取失败。
• main() 函数： 创建一个 RingBuffer 对象，并启动一个写入线程和一个读取线程进行测试。

四、 进一步优化：缓存行填充和事件监听

上面的代码只是一个简单的示例，还有很多可以优化的地方。

1. 缓存行填充（Padding）： 为了避免伪共享，可以在 write_sequence_ 和 read_sequence_ 周围填充缓存行大小的字节。

template <typename T>
class RingBuffer {
private:
    // 填充缓存行，避免伪共享
    struct alignas(64) Sequence {
        std::atomic<size_t> value{0};
    };

    Sequence write_sequence_;
    char padding1[64]; // 填充缓存行
    Sequence read_sequence_;
    char padding2[64]; // 填充缓存行
    ...
};

2. 事件监听（Event Listener）： 可以添加事件监听机制，当有新的数据写入缓冲区时，通知读取线程。这样可以避免读取线程一直轮询检查缓冲区是否为空。

#include <condition_variable>
#include <mutex>

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

    bool write(const T& data) {
        // CAS 操作，尝试将 write_sequence_ + 1 赋值给 write_sequence_
        size_t current_write_sequence = write_sequence_.value.load(std::memory_order_relaxed);
        size_t next_write_sequence = current_write_sequence + 1;

        if (next_write_sequence - read_sequence_.value.load(std::memory_order_acquire) > capacity_) {
            // 缓冲区已满
            return false;
        }

        size_t index = next_write_sequence % capacity_;
        buffer_[index] = data;

        // 使用 compare_exchange_weak 实现 CAS 操作
        while (!write_sequence_.value.compare_exchange_weak(current_write_sequence, next_write_sequence,
                                                      std::memory_order_release, std::memory_order_relaxed)) {
            // CAS 失败，重新读取 write_sequence_ 的值，重试
            // 注意这里不能使用 memory_order_acquire，否则可能导致死锁
            ;
        }

        // 通知读取线程
        {
            std::lock_guard<std::mutex> lock(mutex_);
            data_available_ = true;
        }
        condition_.notify_one();

        return true;
    }

    bool read(T& data) {
        std::unique_lock<std::mutex> lock(mutex_);
        condition_.wait(lock, [&] { return write_sequence_.value.load(std::memory_order_acquire) > read_sequence_.value.load(std::memory_order_relaxed); });

        size_t current_read_sequence = read_sequence_.value.load(std::memory_order_relaxed);
        size_t next_read_sequence = current_read_sequence + 1;

        if (next_read_sequence > write_sequence_.value.load(std::memory_order_acquire)) {
            // 缓冲区为空
            data_available_ = false;
            return false;
        }

        size_t index = next_read_sequence % capacity_;
        data = buffer_[index];

        while (!read_sequence_.value.compare_exchange_weak(current_read_sequence, next_read_sequence,
                                                     std::memory_order_release, std::memory_order_relaxed)) {
            ;
        }

        return true;
    }

private:
    // 填充缓存行，避免伪共享
    struct alignas(64) Sequence {
        std::atomic<size_t> value{0};
    };

    size_t capacity_;
    std::vector<T> buffer_;

    Sequence write_sequence_;
    char padding1[64]; // 填充缓存行
    Sequence read_sequence_;
    char padding2[64]; // 填充缓存行

    std::mutex mutex_;
    std::condition_variable condition_;
    bool data_available_ = false;
};

代码解释：

• 添加了 std::mutex 和 std::condition_variable 用于线程同步。
• write() 方法在写入数据后，通知读取线程。
• read() 方法在读取数据前，等待数据可用。

五、 性能分析和测试

光说不练假把式，我们来测试一下环形缓冲区的性能。

测试环境：

• CPU：Intel Core i7-8700K
• 内存：32GB
• 操作系统：Ubuntu 20.04
• 编译器：g++ 9.4.0

测试用例：

• 创建一个环形缓冲区。
• 启动一个写入线程，向缓冲区写入大量数据。
• 启动多个读取线程，从缓冲区读取数据。
• 统计写入和读取的吞吐量。

测试代码：

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

// RingBuffer 的代码 (这里省略，使用上面优化后的代码)

int main() {
    size_t capacity = 1024 * 1024; // 1MB
    int num_threads = 4;
    long long num_messages = 10000000;

    RingBuffer<long long> ring_buffer(capacity);

    // 写入线程
    std::thread writer([&]() {
        auto start = std::chrono::high_resolution_clock::now();
        for (long long i = 0; i < num_messages; ++i) {
            while (!ring_buffer.write(i)) {
                // 缓冲区满，等待
                std::this_thread::sleep_for(std::chrono::microseconds(1));
            }
        }
        auto end = std::chrono::high_resolution_clock::now();
        auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count();
        std::cout << "Writer: " << num_messages << " messages in " << duration << " ms" << std::endl;
        std::cout << "Writer Throughput: " << (double)num_messages / duration * 1000 << " msg/s" << std::endl;
    });

    // 读取线程
    std::vector<std::thread> readers;
    for (int i = 0; i < num_threads; ++i) {
        readers.emplace_back([&]() {
            long long data;
            long long count = 0;
            while (count < num_messages) {
                if (ring_buffer.read(data)) {
                    count++;
                } else {
                    std::this_thread::sleep_for(std::chrono::microseconds(1));
                }
            }
        });
    }

    writer.join();
    for (auto& reader : readers) {
        reader.join();
    }

    std::cout << "All threads finished." << std::endl;

    return 0;
}

测试结果（示例）：

线程数	写入吞吐量 (msg/s)
1	1500000
2	1400000
4	1300000

性能分析：

• 随着线程数的增加，写入吞吐量略有下降，这是因为多个读取线程会竞争 read_sequence_，导致 CAS 操作的失败率增加。
• 整体来说，环形缓冲区的性能非常高，可以满足高吞吐量的需求。

六、 总结和注意事项

无锁环形缓冲区是一种高性能的消息传递机制，适用于多线程并发环境。

优点：

• 避免了锁的开销，提高了并发性能。
• 内存分配固定，避免了频繁的 new 和 delete。
• 读写操作简单高效。

缺点：

• 容量固定，需要合理的控制策略。
• 实现较为复杂，需要仔细考虑线程安全问题。

注意事项：

• 正确使用原子操作，保证线程安全。
• 避免伪共享，使用缓存行填充。
• 合理选择缓冲区容量，避免缓冲区满或空的情况。
• 根据实际应用场景，选择合适的同步机制（例如，事件监听）。
• 仔细测试，确保代码的正确性和性能。

希望今天的分享能帮助你更好地理解和使用 C++ 无锁环形缓冲区。 记住，不要过度追求完美，选择最适合你应用场景的方案才是最重要的！ 感谢大家的聆听！