C++環(huán)形緩沖區(qū)實踐與注意事項
更新時間：2025年11月04日 08:28:48 作者：oioihoii
環(huán)形緩沖區(qū)是一種高效的數(shù)據(jù)結(jié)構(gòu),特別適用于生產(chǎn)者-消費者場景、數(shù)據(jù)流處理和緩存管理,本文就來詳細(xì)介紹環(huán)形緩沖區(qū)的實現(xiàn)原理和實踐,感興趣的可以了解一下
環(huán)形緩沖區(qū)（Circular Buffer）是一種高效的數(shù)據(jù)結(jié)構(gòu)，特別適用于生產(chǎn)者-消費者場景、數(shù)據(jù)流處理和緩存管理。下面我將詳細(xì)介紹環(huán)形緩沖區(qū)的實現(xiàn)原理、代碼實踐和注意事項。
環(huán)形緩沖區(qū)核心概念
環(huán)形緩沖區(qū)通過固定大小的數(shù)組和兩個指針（讀指針和寫指針）實現(xiàn)循環(huán)使用存儲空間。當(dāng)指針到達(dá)數(shù)組末尾時，會回到數(shù)組開頭繼續(xù)操作。
完整實現(xiàn)代碼
#include <iostream>
#include <vector>
#include <mutex>
#include <condition_variable>
#include <thread>
#include <chrono>
#include <atomic>
#include <stdexcept>
template<typename T>
class CircularBuffer {
public:
// 構(gòu)造函數(shù)
explicit CircularBuffer(size_t size)
: buffer_(size), capacity_(size), read_pos_(0), write_pos_(0),
count_(0), is_full_(false) {
if (size == 0) {
throw std::invalid_argument("Buffer size must be greater than 0");
}
}
// 默認(rèn)析構(gòu)函數(shù)
~CircularBuffer() = default;
// 禁止拷貝和賦值
CircularBuffer(const CircularBuffer&) = delete;
CircularBuffer& operator=(const CircularBuffer&) = delete;
// 寫入數(shù)據(jù)（非阻塞）
bool push(const T& item) {
std::lock_guard<std::mutex> lock(mutex_);
if (is_full_) {
return false; // 緩沖區(qū)已滿
}
buffer_[write_pos_] = item;
write_pos_ = (write_pos_ + 1) % capacity_;
++count_;
is_full_ = (write_pos_ == read_pos_);
not_empty_.notify_one();
return true;
}
// 寫入數(shù)據(jù)（阻塞）
bool push_blocking(const T& item, std::chrono::milliseconds timeout = std::chrono::milliseconds(0)) {
std::unique_lock<std::mutex> lock(mutex_);
if (timeout.count() > 0) {
// 帶超時的等待
if (!not_full_.wait_for(lock, timeout, [this]() { return !is_full_; })) {
return false; // 超時
}
} else {
// 無限等待
not_full_.wait(lock, [this]() { return !is_full_; });
}
buffer_[write_pos_] = item;
write_pos_ = (write_pos_ + 1) % capacity_;
++count_;
is_full_ = (write_pos_ == read_pos_);
not_empty_.notify_one();
return true;
}
// 讀取數(shù)據(jù)（非阻塞）
bool pop(T& item) {
std::lock_guard<std::mutex> lock(mutex_);
if (empty()) {
return false; // 緩沖區(qū)為空
}
item = buffer_[read_pos_];
read_pos_ = (read_pos_ + 1) % capacity_;
--count_;
is_full_ = false;
not_full_.notify_one();
return true;
}
// 讀取數(shù)據(jù)（阻塞）
bool pop_blocking(T& item, std::chrono::milliseconds timeout = std::chrono::milliseconds(0)) {
std::unique_lock<std::mutex> lock(mutex_);
if (timeout.count() > 0) {
// 帶超時的等待
if (!not_empty_.wait_for(lock, timeout, [this]() { return !empty(); })) {
return false; // 超時
}
} else {
// 無限等待
not_empty_.wait(lock, [this]() { return !empty(); });
}
item = buffer_[read_pos_];
read_pos_ = (read_pos_ + 1) % capacity_;
--count_;
is_full_ = false;
not_full_.notify_one();
return true;
}
// 查看但不移除數(shù)據(jù)
bool peek(T& item) const {
std::lock_guard<std::mutex> lock(mutex_);
if (empty()) {
return false;
}
item = buffer_[read_pos_];
return true;
}
// 清空緩沖區(qū)
void clear() {
std::lock_guard<std::mutex> lock(mutex_);
read_pos_ = write_pos_;
count_ = 0;
is_full_ = false;
// 注意：這里不需要通知條件變量，因為清空操作不會讓緩沖區(qū)變滿
}
// 判斷是否為空
bool empty() const {
// 注意：當(dāng) read_pos_ == write_pos_ 且 !is_full_ 時為空
return (!is_full_ && (read_pos_ == write_pos_));
}
// 判斷是否已滿
bool full() const {
return is_full_;
}
// 獲取當(dāng)前元素數(shù)量
size_t size() const {
return count_;
}
// 獲取容量
size_t capacity() const {
return capacity_;
}
// 獲取剩余空間
size_t available() const {
return capacity_ - count_;
}
private:
std::vector<T> buffer_;
size_t capacity_;
mutable std::mutex mutex_;
std::condition_variable not_empty_;
std::condition_variable not_full_;
size_t read_pos_; // 讀位置
size_t write_pos_; // 寫位置
size_t count_; // 當(dāng)前元素數(shù)量
bool is_full_; // 是否已滿標(biāo)志
};
// 無鎖環(huán)形緩沖區(qū)（適用于單生產(chǎn)者單消費者場景）
template<typename T, size_t Size>
class LockFreeCircularBuffer {
public:
LockFreeCircularBuffer() : read_index_(0), write_index_(0) {}
bool push(const T& item) {
size_t current_write = write_index_.load(std::memory_order_relaxed);
size_t next_write = next_index(current_write);
// 檢查是否已滿
if (next_write == read_index_.load(std::memory_order_acquire)) {
return false;
}
buffer_[current_write] = item;
write_index_.store(next_write, std::memory_order_release);
return true;
}
bool pop(T& item) {
size_t current_read = read_index_.load(std::memory_order_relaxed);
// 檢查是否為空
if (current_read == write_index_.load(std::memory_order_acquire)) {
return false;
}
item = buffer_[current_read];
read_index_.store(next_index(current_read), std::memory_order_release);
return true;
}
bool empty() const {
return read_index_.load(std::memory_order_acquire) ==
write_index_.load(std::memory_order_acquire);
}
bool full() const {
size_t next_write = next_index(write_index_.load(std::memory_order_relaxed));
return next_write == read_index_.load(std::memory_order_acquire);
}
size_t size() const {
size_t write = write_index_.load(std::memory_order_acquire);
size_t read = read_index_.load(std::memory_order_acquire);
if (write >= read) {
return write - read;
} else {
return Size - read + write;
}
}
private:
size_t next_index(size_t current) const {
return (current + 1) % Size;
}
T buffer_[Size];
std::atomic<size_t> read_index_;
std::atomic<size_t> write_index_;
};
// 演示使用示例
void demo_usage() {
std::cout << "=== 環(huán)形緩沖區(qū)演示 ===" << std::endl;
// 創(chuàng)建容量為5的環(huán)形緩沖區(qū)
CircularBuffer<int> buffer(5);
// 基本操作演示
std::cout << "緩沖區(qū)容量: " << buffer.capacity() << std::endl;
std::cout << "初始大小: " << buffer.size() << std::endl;
std::cout << "是否為空: " << buffer.empty() << std::endl;
// 寫入數(shù)據(jù)
for (int i = 1; i <= 5; ++i) {
if (buffer.push(i)) {
std::cout << "寫入: " << i << std::endl;
}
}
std::cout << "寫入5個數(shù)據(jù)后大小: " << buffer.size() << std::endl;
std::cout << "是否已滿: " << buffer.full() << std::endl;
// 嘗試寫入第6個數(shù)據(jù)（應(yīng)該失?。?
if (!buffer.push(6)) {
std::cout << "寫入失敗 - 緩沖區(qū)已滿" << std::endl;
}
// 讀取數(shù)據(jù)
int value;
while (buffer.pop(value)) {
std::cout << "讀取: " << value << std::endl;
}
std::cout << "讀取所有數(shù)據(jù)后大小: " << buffer.size() << std::endl;
std::cout << "是否為空: " << buffer.empty() << std::endl;
}
// 生產(chǎn)者-消費者演示
void producer_consumer_demo() {
std::cout << "\n=== 生產(chǎn)者-消費者演示 ===" << std::endl;
CircularBuffer<int> buffer(10);
std::atomic<bool> stop_producer(false);
std::atomic<bool> stop_consumer(false);
// 生產(chǎn)者線程
std::thread producer([&]() {
for (int i = 1; i <= 15; ++i) {
if (buffer.push_blocking(i, std::chrono::milliseconds(100))) {
std::cout << "生產(chǎn): " << i << std::endl;
} else {
std::cout << "生產(chǎn)超時: " << i << std::endl;
}
std::this_thread::sleep_for(std::chrono::milliseconds(50));
}
stop_producer.store(true);
});
// 消費者線程
std::thread consumer([&]() {
int value;
while (!stop_producer.load() || !buffer.empty()) {
if (buffer.pop_blocking(value, std::chrono::milliseconds(200))) {
std::cout << "消費: " << value << std::endl;
}
std::this_thread::sleep_for(std::chrono::milliseconds(100));
}
stop_consumer.store(true);
});
producer.join();
consumer.join();
std::cout << "生產(chǎn)者-消費者演示結(jié)束" << std::endl;
}
// 性能測試
void performance_test() {
std::cout << "\n=== 性能測試 ===" << std::endl;
const int ITERATIONS = 1000000;
CircularBuffer<int> buffer(1000);
auto start = std::chrono::high_resolution_clock::now();
std::thread producer([&]() {
for (int i = 0; i < ITERATIONS; ++i) {
while (!buffer.push(i)) {
std::this_thread::yield();
}
}
});
std::thread consumer([&]() {
int value;
for (int i = 0; i < ITERATIONS; ++i) {
while (!buffer.pop(value)) {
std::this_thread::yield();
}
// 驗證數(shù)據(jù)完整性
if (value != i) {
std::cerr << "數(shù)據(jù)損壞: 期望 " << i << ", 得到 " << value << std::endl;
}
}
});
producer.join();
consumer.join();
auto end = std::chrono::high_resolution_clock::now();
auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(end - start);
std::cout << "處理 " << ITERATIONS << " 個元素耗時: "
<< duration.count() << " ms" << std::endl;
std::cout << "吞吐量: "
<< (ITERATIONS * 1000.0 / duration.count()) << " 操作/秒" << std::endl;
}
int main() {
try {
demo_usage();
producer_consumer_demo();
performance_test();
// 無鎖緩沖區(qū)演示
std::cout << "\n=== 無鎖環(huán)形緩沖區(qū)演示 ===" << std::endl;
LockFreeCircularBuffer<int, 5> lock_free_buffer;
for (int i = 1; i <= 3; ++i) {
if (lock_free_buffer.push(i)) {
std::cout << "無鎖緩沖區(qū)寫入: " << i << std::endl;
}
}
int value;
while (lock_free_buffer.pop(value)) {
std::cout << "無鎖緩沖區(qū)讀取: " << value << std::endl;
}
} catch (const std::exception& e) {
std::cerr << "錯誤: " << e.what() << std::endl;
return 1;
}
return 0;
}
關(guān)鍵注意事項
1. 線程安全設(shè)計
- 互斥鎖保護(hù)：使用
std::mutex保護(hù)共享數(shù)據(jù) - 條件變量：使用
std::condition_variable實現(xiàn)高效的等待通知機(jī)制 - 內(nèi)存序：無鎖版本中正確使用內(nèi)存序保證數(shù)據(jù)一致性
2. 空滿判斷策略
// 方法1：使用計數(shù)變量（推薦）
bool empty() const { return count_ == 0; }
bool full() const { return count_ == capacity_; }
// 方法2：使用標(biāo)志位
bool empty() const { return !is_full_ && (read_pos_ == write_pos_); }
bool full() const { return is_full_; }
3. 指針管理
// 指針前進(jìn) read_pos_ = (read_pos_ + 1) % capacity_; write_pos_ = (write_pos_ + 1) % capacity_;
4. 異常安全
- 構(gòu)造函數(shù)驗證參數(shù)有效性
- 使用 RAII 管理資源
- 提供強(qiáng)異常安全保證
編譯和運行
使用以下命令編譯：
g++ -std=c++11 -pthread -O2 circular_buffer.cpp -o circular_buffer
運行：
./circular_buffer
性能優(yōu)化建議
- 緩存友好：確保數(shù)據(jù)連續(xù)存儲，提高緩存命中率
- 減少鎖競爭：使用細(xì)粒度鎖或無鎖設(shè)計
- 批量操作：支持批量讀寫減少鎖開銷
- 內(nèi)存預(yù)分配：避免動態(tài)內(nèi)存分配
適用場景
- 數(shù)據(jù)流處理（音頻、視頻流）
- 生產(chǎn)者-消費者模式
- 網(wǎng)絡(luò)數(shù)據(jù)包緩沖
- 實時系統(tǒng)數(shù)據(jù)交換
- 日志記錄系統(tǒng)
這個實現(xiàn)提供了完整的環(huán)形緩沖區(qū)功能，包括線程安全、阻塞/非阻塞操作、異常安全等特性，可以直接在生產(chǎn)環(huán)境中使用。
到此這篇關(guān)于C++環(huán)形緩沖區(qū)實踐與注意事項的文章就介紹到這了,更多相關(guān)C++環(huán)形緩沖區(qū)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
本文主要解析了C/C++中數(shù)組和指針的區(qū)別，文章簡單易懂，對各位的工作學(xué)習(xí)有所幫助，有需求的朋友可以了解下2020-05-05
這篇文章介紹了C語言make和Makefile以及使用方法，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2022-01-01
這篇文章主要介紹了C++中繼承基類與派生類的區(qū)別,面向?qū)ο蟪绦蛟O(shè)計中最重要的一個概念是繼承。繼承允許我們依據(jù)另一個類來定義一個類，這使得創(chuàng)建和維護(hù)一個應(yīng)用程序變得更容易,需要的朋友可以參考下2023-05-05
本文詳細(xì)講解了C語言實現(xiàn)掃雷并附完整代碼，文中通過示例代碼介紹的非常詳細(xì)。對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值，需要的朋友可以參考下2021-11-11
本文介紹了如何在VS2022中配置和編譯使用Boost庫的步驟,包括下載Boost、解壓、配置環(huán)境變量和編譯等過程,具有一定的參考價值,感興趣的可以了解一下2024-12-12
這篇文章主要為大家詳細(xì)介紹了C++基于特征向量的KNN分類算法，具有一定的參考價值，感興趣的小伙伴們可以參考一下2018-12-12
虛函數(shù)主要通過V-Table虛函數(shù)表來實現(xiàn),該表主要包含一個類的虛函數(shù)的地址表,可解決繼承、覆蓋的問題,下面這篇文章主要給大家介紹了如何通過一篇文章帶你掌握C++虛函數(shù)的來龍去脈,需要的朋友可以參考下2022-12-12
今天講解的內(nèi)容，則是自己對于這三種語句一些細(xì)節(jié)的簡單介紹，分支語句：if，switch、循環(huán)語句：while，for，do while、goto語句，感興趣的小伙伴可以參考下面具體的文章內(nèi)容2021-09-09
Opencv基于CamShift算法實現(xiàn)目標(biāo)跟蹤
這篇文章主要為大家詳細(xì)介紹了Opencv基于CamShift算法實現(xiàn)目標(biāo)跟蹤，具有一定的參考價值，感興趣的小伙伴們可以參考一下2018-01-01
C語言實現(xiàn)小學(xué)生計算機(jī)輔助教學(xué)系統(tǒng)
這篇文章主要為大家詳細(xì)介紹了C語言實現(xiàn)小學(xué)生計算機(jī)輔助教學(xué)系統(tǒng)，具有一定的參考價值，感興趣的小伙伴們可以參考一下2019-03-03