先看带锁的实现。
带锁版本
circular_queue.h
// 头文件防卫
#ifndef CIRCULAR_QUEUE_H
#define CIRCULAR_QUEUE_H
#include <mutex> // 互斥量
#include <condition_variable> // 条件变量
template <typename T>
class CircularQueue {
public:
// 构造函数,初始化成员变量
explicit CircularQueue(size_t capacity) :
capacity_(capacity),
size_(0),
head_(0),
tail_(0),
buffer_(new T[capacity]) {}
// 析构函数,释放 buffer_ 内存
~CircularQueue() {
delete[] buffer_;
}
// 判断队列是否为空
bool empty() {
std::unique_lock<std::mutex> lock(mutex_);
return size_ == 0;
}
// 判断队列是否已满
bool full() {
std::unique_lock<std::mutex> lock(mutex_);
return size_ == capacity_;
}
// 获取队列中元素的数量
size_t size() {
std::unique_lock<std::mutex> lock(mutex_);
return size_;
}
// 获取队列的容量
size_t capacity() {
return capacity_;
}
// 将元素加入队列,可能会阻塞
bool push(const T& value, bool block = true) {
std::unique_lock<std::mutex> lock(mutex_);
if (block) {
// 如果队列已满,则等待队列不满
while (size_ == capacity_) {
not_full_.wait(lock);
}
} else {
// 如果队列已满,则返回 false
if (size_ == capacity_) {
return false;
}
}
// 将元素加入队列尾部,并更新 tail_ 和 size_
buffer_[tail_] = value;
tail_ = (tail_ + 1) % capacity_;
++size_;
// 通知一个等待在 not_empty_ 条件变量上的线程
not_empty_.notify_one();
return true;
}
// 将元素加入队列,可能会阻塞,使用右值引用
bool push(T&& value, bool block = true) {
std::unique_lock<std::mutex> lock(mutex_);
if (block) {
// 如果队列已满,则等待队列不满
while (size_ == capacity_) {
not_full_.wait(lock);
}
} else {
// 如果队列已满,则返回 false
if (size_ == capacity_) {
return false;
}
}
// 将元素加入队列尾部,并更新 tail_ 和 size_
buffer_[tail_] = std::move(value);
tail_ = (tail_ + 1) % capacity_;
++size_;
// 通知一个等待在 not_empty_ 条件变量上的线程
not_empty_.notify_one();
return true;
}
// 从队列中取出元素,可能会阻塞
bool pop(T& value, bool block = true) {
std::unique_lock<std::mutex> lock(mutex_);
if (block) {
// 如果队列为空,则等待队列不空
while (size_ == 0) {
not_empty_.wait(lock);
}
} else {
// 如果队列为空,则返回 false
if (size_ == 0) {
return false;
}
}
// 取出队列头部元素,并更新 head_ 和 size_
value = std::move(buffer_[head_]);
head_ = (head_ + 1) % capacity_;
--size_;
// 通知一个等待在 not_full_ 条件变量上的线程
not_full_.notify_one();
return true;
}
private:
const size_t capacity_; // 队列容量
size_t size_; // 队列中元素的数量
size_t head_; // 队列头部指针
size_t tail_; // 队列尾部指针
T* buffer_; // 队列缓冲区
std::mutex mutex_; // 互斥量,保护队列缓冲区和队列大小
std::condition_variable not_full_; // 条件变量,当队列满时等待
std::condition_variable not_empty_; // 条件变量,当队列空时等待
};
#endif // CIRCULAR_QUEUE_H
push和pop接口不指定第二个参数的话,默认是阻塞的,这一点使用时需要注意。
以下是CircularQueue类的单元测试示例代码:
#include <gtest/gtest.h>
#include <thread>
#include "circular_queue.h"
TEST(CircularQueueTest, EmptyQueue) {
CircularQueue<int> queue(10);
ASSERT_TRUE(queue.empty());
ASSERT_FALSE(queue.full());
ASSERT_EQ(queue.size(), 0);
ASSERT_EQ(queue.capacity(), 10);
}
TEST(CircularQueueTest, PushAndPop) {
CircularQueue<int> queue(3);
ASSERT_TRUE(queue.push(1));
ASSERT_EQ(queue.size(), 1);
ASSERT_FALSE(queue.empty());
ASSERT_FALSE(queue.full());
ASSERT_TRUE(queue.push(2));
ASSERT_EQ(queue.size(), 2);
ASSERT_FALSE(queue.empty());
ASSERT_FALSE(queue.full());
ASSERT_TRUE(queue.push(3));
ASSERT_EQ(queue.size(), 3);
ASSERT_FALSE(queue.empty());
ASSERT_TRUE(queue.full());
int value;
ASSERT_TRUE(queue.pop(value));
ASSERT_EQ(value, 1);
ASSERT_EQ(queue.size(), 2);
ASSERT_FALSE(queue.empty());
ASSERT_FALSE(queue.full());
ASSERT_TRUE(queue.pop(value));
ASSERT_EQ(value, 2);
ASSERT_EQ(queue.size(), 1);
ASSERT_FALSE(queue.empty());
ASSERT_FALSE(queue.full());
ASSERT_TRUE(queue.pop(value));
ASSERT_EQ(value, 3);
ASSERT_EQ(queue.size(), 0);
ASSERT_TRUE(queue.empty());
ASSERT_FALSE(queue.full());
ASSERT_FALSE(queue.pop(value,false));
}
TEST(CircularQueueTest, PushAndPopWithBlocking) {
CircularQueue<int> queue(2);
std::thread t([&queue]() {
int value = 0;
queue.pop(value);
ASSERT_EQ(value, 1);
queue.pop(value);
ASSERT_EQ(value, 2);
});
ASSERT_TRUE(queue.push(1));
ASSERT_TRUE(queue.push(2));
ASSERT_TRUE(queue.push(3));
t.join();
}
TEST(CircularQueueTest, PushAndPopWithNonBlocking) {
CircularQueue<int> queue(2);
int value;
ASSERT_TRUE(queue.push(1));
ASSERT_TRUE(queue.push(2));
ASSERT_FALSE(queue.push(3, false));
ASSERT_TRUE(queue.pop(value));
ASSERT_EQ(value, 1);
ASSERT_TRUE(queue.pop(value));
ASSERT_EQ(value, 2);
ASSERT_FALSE(queue.pop(value, false));
}
TEST(CircularQueueTest, MovePushAndPop) {
CircularQueue<std::string> queue(3);
ASSERT_TRUE(queue.push("hello"));
ASSERT_TRUE(queue.push("world"));
ASSERT_EQ(queue.size(), 2);
std::string value;
ASSERT_TRUE(queue.pop(value));
ASSERT_EQ(value, "hello");
ASSERT_EQ(queue.size(), 1);
ASSERT_TRUE(queue.push("foo"));
ASSERT_EQ(queue.size(), 2);
ASSERT_TRUE(queue.pop(value));
ASSERT_EQ(value, "world");
ASSERT_EQ(queue.size(), 1);
ASSERT_TRUE(queue.pop(value));
ASSERT_EQ(value, "foo");
ASSERT_EQ(queue.size(), 0);
}
TEST(CircularQueueTest, CopyPushAndPop) {
CircularQueue<std::string> queue(3);
ASSERT_TRUE(queue.push(std::string("hello")));
ASSERT_TRUE(queue.push(std::string("world")));
ASSERT_EQ(queue.size(), 2);
std::string value;
ASSERT_TRUE(queue.pop(value));
ASSERT_EQ(value, "hello");
ASSERT_EQ(queue.size(), 1);
ASSERT_TRUE(queue.push(std::string("foo")));
ASSERT_EQ(queue.size(), 2);
ASSERT_TRUE(queue.pop(value));
ASSERT_EQ(value, "world");
ASSERT_EQ(queue.size(), 1);
ASSERT_TRUE(queue.pop(value));
ASSERT_EQ(value, "foo");
ASSERT_EQ(queue.size(), 0);
}
TEST(CircularQueueTest, MultiThreadPushPop) {
const int num_threads = 4;
const int num_iterations = 10000;
const int queue_size = 100;
CircularQueue<int> queue(queue_size);
std::vector<std::thread> threads;
for (int i = 0; i < num_threads; ++i) {
threads.emplace_back([&queue, num_iterations]() {
for (int j = 0; j < num_iterations; ++j) {
queue.push(j);
}
});
}
for (int i = 0; i < num_threads; ++i) {
threads.emplace_back([&queue, num_iterations]() {
for (int j = 0; j < num_iterations; ++j) {
int value;
queue.pop(value);
}
});
}
for (auto& thread : threads) {
thread.join();
}
ASSERT_EQ(queue.size(), 0);
}
int main(int argc, char** argv) {
testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}
单元测试运行结果:
无锁版本
上面的循环队列使用锁保证了线程安全。
以下是一个基于C++11的线程安全且无锁的环形队列实现,支持阻塞读和非阻塞读,写也一样。
#include <atomic>
#include <condition_variable>
#include <functional>
#include <iostream>
#include <thread>
#include <vector>
template <typename T, size_t N>
class RingQueue {
public:
RingQueue() : read_idx_(0), write_idx_(0), data_{} {}
bool Push(const T& item, bool block = false) { return PushImpl(item, block); }
bool Push(T&& item, bool block = false) { return PushImpl(std::move(item), block); }
bool Pop(T& item, bool block = false) {
// 只有一个读者,read_idx_的读取可以不加锁
size_t current_read_idx = read_idx_.load(std::memory_order_relaxed);
// 保证读到write_idx_的变化,此处memory_order_acquire发挥的是可见性问题
while (current_read_idx == write_idx_.load(std::memory_order_acquire)) {
if (!block) {
return false;
}
std::this_thread::yield();
}
item = std::move(data_[current_read_idx]); // 必须先把数据读走read_idx_才能+1,memory_order_release保证了对item的写不会被重排到read_idx_ + 1之后
read_idx_.store(Next(current_read_idx), std::memory_order_release);
return true;
}
template <typename Func>
bool Pop(Func&& func, bool block = false) {
size_t current_read_idx = read_idx_.load(std::memory_order_relaxed);
while (current_read_idx == write_idx_.load(std::memory_order_acquire)) {
if (!block) {
return false;
}
std::this_thread::yield();
}
T item = std::move(data_[current_read_idx]);
read_idx_.store(Next(current_read_idx), std::memory_order_release);
func(std::move(item));
return true;
}
void PopAsync(const T& value, std::function<void(bool)> callback) {
auto task = [this, value, callback]() {
bool result = Pop(value, true);
callback(result);
};
std::thread(std::move(task)).detach();
}
bool IsEmpty() const {
return read_idx_.load(std::memory_order_acquire) ==
write_idx_.load(std::memory_order_acquire);
}
bool IsFull() const {
return Next(write_idx_.load(std::memory_order_acquire)) ==
read_idx_.load(std::memory_order_acquire);
}
private:
template <typename Item>
bool PushImpl(Item&& item, bool block = false) {
// 只有1个写线程, 所以write_idx_可以不加锁
size_t current_write_idx = write_idx_.load(std::memory_order_relaxed);
size_t next_write_idx = Next(current_write_idx);
// 读线程会修改read_idx_, 所以此处需要保证看到read_idx_的变化,此处memory_order_acquire保证的是可见性问题
while (next_write_idx == read_idx_.load(std::memory_order_acquire)) {
if(!block) {
return false;
}
std::this_thread::yield();
}
// 数据的写入必须在write_idx_+1之前
data_[current_write_idx] = std::forward<Item>(item);
// 保证之前的写操作对读线程可见,即读线程能立刻看到data_刚写入的数据,当然也包括write_idx_的+1变化,memory_order_release会保证对data_的写入在write_idx_+1的操作之前完成。
// 因为就算data_的赋值语句放在write_idx_+1之前,由于编译器或者运行期指令重排,并不一定能保证data_赋值语句就一定在write_idx_+1前执行。
write_idx_.store(next_write_idx, std::memory_order_release);
return true;
}
size_t Next(size_t current_idx) const { return (current_idx + 1) % (N+1); } // 此处笔者做了修改,N改成N+1
std::atomic<size_t> read_idx_;
std::atomic<size_t> write_idx_;
std::array<T, N+1> data_; // 此处笔者做了修改,N改成N+1
};
代码中使用了模板参数T和N,支持了不同数据类型和不同队列大小的选择,借助读写指针两个原子变量实现无锁环形队列。
但需要注意的是,这个只能实现一读一写的线程安全,存在多个读者或者多个写者时就线程不安全了。
无锁编程的难点在于对几个内存时序的理解。
补充下关于内存时序操作的解释。
C++定义了几种内存时序,这些时序规定了原子变量前后的所有内存操作(包括普通变量、原子变量)如何排序
std::memory_order_relaxed
只保正操作的原子性,对于同一个原子变量的多个操作之间不存在任何内存序的限制,也就是说,它们可以随意重排序,也可以在任意时刻被其他线程的操作所干扰。因此,使用std::memory_order_relaxed
时需要特别小心,必须确保操作的正确性不受此种松散的内存访问顺序的影响。
std::memory_order_relaxed
主要用于那些不需要任何同步机制的场合,比如计数器的自增、自减等操作,这些操作只需要保证结果的正确性,而不需要保证其执行的顺序。因此,std::memory_order_relaxed是最快的内存序,但也是最危险的一种内存序。
std::memory_order_acquire
确保所有之前的读操作都已经完成,然后再执行当前读取操作。这意味着,如果当前读取操作需要用到之前的读取操作的结果,那么它将能够正确地获取到这些结果。具体来说,当使用memory_order_acquire语义时,编译器和处理器都会保证当前线程所在的CPU核心(或处理器)在执行当前原子操作之前,会先将所有之前的读操作所获得的数据从CPU缓存中刷新到主内存中,以保证当前线程能够读取到其他线程对共享变量的最新修改。
使用memory_order_acquire语义可以保证程序的正确性,避免出现数据竞争的问题。但是,使用memory_order_acquire语义可能会降低程序的性能,因为它要求在执行原子操作之前,必须将所有之前的读操作都刷新到主内存中,这可能会导致缓存一致性协议的开销增加。因此,在实际编程中,应该根据具体情况选择合适的内存序语义。
std::memory_order_release
确保当前线程的所有写操作在该原子操作之前都已经完成,并且将这些写操作对其他线程可见。这样,其他线程就可以看到当前线程对共享数据所做的更改。这种释放操作通常用于同步操作,例如将一个共享变量的值更新后通知其他线程。在这种情况下,std::memory_order_release可以确保其他线程能够看到更新后的值。
Push
提供了插入左值和右值两个版本,提高C++所谓的一点点性能,PushImpl
提取了公共代码,实现代码复用,优雅!。
Pop
提供了两个版本,有个变体,不返回pop出来的对象,而是调用外部传入的回调,对pop出来的对象进行操作。
同样附上单元测试。
#include <gtest/gtest.h>
#include "RingQueue.h"
class RingQueueSingleThreadTest : public ::testing::Test {
protected:
RingQueue<int, 10> queue_;
};
TEST_F(RingQueueSingleThreadTest, PushAndPop) {
int value = 0;
EXPECT_FALSE(queue_.Pop(value));
EXPECT_TRUE(queue_.Push(1));
EXPECT_FALSE(queue_.IsEmpty());
EXPECT_TRUE(queue_.Pop(value));
EXPECT_EQ(value, 1);
EXPECT_TRUE(queue_.IsEmpty());
}
TEST_F(RingQueueSingleThreadTest, PushAndPopWithBlock) {
int value = 0;
std::thread t([&](){
std::this_thread::sleep_for(std::chrono::milliseconds(100));
queue_.Push(1, true);
});
EXPECT_TRUE(queue_.Pop(value, true));
EXPECT_EQ(value, 1);
t.join();
}
TEST_F(RingQueueSingleThreadTest, PushAndPopWithFunc) {
int value = 0;
queue_.Push(1);
queue_.Pop([&](int v){ value = v + 1; });
EXPECT_EQ(value, 2);
}
TEST_F(RingQueueSingleThreadTest, IsEmptyAndIsFull) {
EXPECT_TRUE(queue_.IsEmpty());
EXPECT_FALSE(queue_.IsFull());
for (int i = 0; i < 10; ++i) {
EXPECT_TRUE(queue_.Push(i));
}
EXPECT_TRUE(queue_.IsFull());
EXPECT_FALSE(queue_.IsEmpty());
int value = 0;
EXPECT_FALSE(queue_.Push(10));
EXPECT_TRUE(queue_.Pop(value));
EXPECT_FALSE(queue_.IsFull());
}
class RingQueueMultiThreadTest : public testing::Test {
protected:
virtual void SetUp() {
// 初始化数据
for (int i = 0; i < 1000; ++i) {
data_.push_back("data_" + std::to_string(i));
}
}
std::vector<std::string> data_;
};
TEST_F(RingQueueMultiThreadTest, MultiThreadTest) {
RingQueue<std::string, 10> queue;
// 写线程
std::thread writer([&queue, this]() {
for (const auto& item : data_) {
queue.Push(item, true);
}
});
// 读线程
std::thread reader([&queue, this]() {
int count = 0;
std::string item;
while (count < 1000) {
if (queue.Pop(item, true)) {
EXPECT_EQ(item, "data_" + std::to_string(count));
++count;
} else {
std::this_thread::yield();
}
}
});
writer.join();
reader.join();
}
int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}