// // Created by xtkuang on 2025/5/30. // #ifndef CMVR_ES_RING_BUFFER_H #define CMVR_ES_RING_BUFFER_H #pragma once #include #include #include #include #include template class RingBuffer { public: explicit RingBuffer(size_t capacity) : capacity_(capacity) {} void push(const T& item) { std::lock_guard lock(mutex_); if (buffer_.size() >= capacity_) { buffer_.pop_front(); } buffer_.push_back(item); } std::vector getAll() const { std::lock_guard lock(mutex_); return std::vector(buffer_.begin(), buffer_.end()); } void clear() { std::lock_guard lock(mutex_); buffer_.clear(); } size_t size() const { std::lock_guard lock(mutex_); return buffer_.size(); } private: size_t capacity_; std::deque buffer_; mutable std::mutex mutex_; }; template class SPMCRingBuffer { public: explicit SPMCRingBuffer(size_t capacity) : buffer_(capacity), capacity_(capacity), head_(0), tail_(0) {} // 写入操作(仅支持单个生产者) void push(const T& item) { size_t head = head_.load(std::memory_order_relaxed); size_t tail = tail_.load(std::memory_order_acquire); buffer_[head % capacity_] = item; head = head + 1; head_.store(head, std::memory_order_release); if (head - tail >= capacity_) { // 队列满,覆盖最旧的数据 tail_.store(tail + 1, std::memory_order_release); } } // 单消费者使用(内部 tail_) std::optional pop() { size_t tail = tail_.load(std::memory_order_relaxed); size_t head = head_.load(std::memory_order_acquire); if (tail >= head) return std::nullopt; T value = buffer_[tail % capacity_]; tail_.store(tail + 1, std::memory_order_release); return value; } std::optional getLast() { size_t tail = tail_.load(std::memory_order_relaxed); size_t head = head_.load(std::memory_order_acquire); if (tail >= head) return std::nullopt; T value = buffer_[head_ % capacity_]; return value; } // 多消费者使用(每个读者独立维护 reader_tail) std::optional pop(size_t& reader_tail) const { size_t head = head_.load(std::memory_order_acquire); if (reader_tail >= head) return std::nullopt; if (head > reader_tail + capacity_) { // 数据已被覆盖,跳过无效读取区间 reader_tail = head - capacity_; return std::nullopt; } T value = buffer_[reader_tail % capacity_]; reader_tail++; return value; } size_t size() const { return head_.load(std::memory_order_acquire) - tail_.load(std::memory_order_acquire); } size_t getHead() const { return head_.load(std::memory_order_acquire); } size_t getTail() const { return tail_.load(std::memory_order_acquire); } bool empty() const { return size() == 0; } bool full() const { return size() >= capacity_; } void clear() { head_.store(0, std::memory_order_release); tail_.store(0, std::memory_order_release); } private: std::vector buffer_; const size_t capacity_; std::atomic head_; // 共享写指针 std::atomic tail_; // 共享读指针(仅用于 SPSC 模式) }; #endif //CMVR_ES_RING_BUFFER_H