假设我们不使用互斥锁,并且我们在两个线程中分别调用 enqueue
和 dequeue
方法。
#include <iostream>
#include <thread>
template <typename T> class Queue {
private:
static constexpr int MAX_SIZE = 1000;
T items[MAX_SIZE];
int front, rear;
public:
Queue() : front(-1), rear(-1) {}
bool isEmpty() { return front == -1; }
void enqueue(const T &value) {
if ((rear + 1) % MAX_SIZE == front) {
std::cout << "Queue is full" << std::endl;
return;
}
if (front == -1) {
front = rear = 0;
} else {
rear = (rear + 1) % MAX_SIZE;
}
items[rear] = value;
}
T dequeue() {
if (front == -1) {
std::cout << "Queue is empty" << std::endl;
return T{};
}
T removedItem = items[front];
if (front == rear) {
front = rear = -1;
} else {
front = (front + 1) % MAX_SIZE;
}
return removedItem;
}
};
void producer(Queue<int> &q) {
for (int i = 0; i < 100; ++i) {
q.enqueue(i);
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
}
void consumer(Queue<int> &q) {
while (true) {
if (!q.isEmpty()) {
int value = q.dequeue();
std::cout << "Dequeued: " << value << std::endl;
}
std::this_thread::sleep_for(std::chrono::milliseconds(15));
}
}
int main() {
Queue<int> q;
std::thread t1(producer, std::ref(q));
std::thread t2(consumer, std::ref(q));
t1.join();
t2.join();
return 0;
}
在这个例子中,没有使用任何锁机制来保护 enqueue
和 dequeue
操作。下面是可能出现的问题的几种情况:
数据竞争: 两个线程可能同时访问和修改
front
和rear
变量,导致它们的值不一致。例如:- 线程 A (producer) 在插入一个新元素时,刚刚修改了
rear
,还没来得及更新items[rear]
; - 线程 B (consumer) 可能同时读取
front
并进行删除操作,这时front
和rear
之间的关系可能不一致。
- 线程 A (producer) 在插入一个新元素时,刚刚修改了
丢失数据: 由于没有同步,
enqueue
操作可能会覆盖队列中的数据,而dequeue
操作可能会读取错误的数据或读取到无效的数据。例如:- 线程 A 在
rear
更新前被中断,线程 B 读取到的是尚未正确插入的数据; - 线程 B 在
front
更新前被中断,线程 A 插入的数据可能会被错误地覆盖。
- 线程 A 在
未定义行为: 多线程并发修改共享数据时,可能会导致程序行为不可预测甚至崩溃。
用互斥锁来解决这个问题
#include <condition_variable>
#include <iostream>
#include <mutex>
#include <thread>
template <typename T> class ThreadSafeQueue {
private:
static constexpr int MAX_SIZE = 1000;
T queue[MAX_SIZE];
int front, rear;
std::mutex mtx; // 互斥量,用于保护共享数据
std::condition_variable cv; // 条件变量
public:
ThreadSafeQueue() : front(-1), rear(-1) {}
void push(const T &value) {
std::lock_guard<std::mutex> lock(mtx);
if ((rear + 1) % MAX_SIZE == front) {
std::cout << "Queue is full" << std::endl;
return;
}
if (front == -1) {
front = rear = 0;
} else {
rear = (rear + 1) % MAX_SIZE;
}
queue[rear] = value;
// 通知等待的消费者线程
cv.notify_one();
}
T pop() {
// 等待数据准备就绪
std::unique_lock<std::mutex> lock(mtx);
cv.wait(lock, [this] { return front != -1; }); // 等待条件变量满足
// 处理数据
T removedItem = queue[front];
if (front == rear) {
front = rear = -1;
} else {
front = (front + 1) % MAX_SIZE;
}
return removedItem;
}
};
void producer(ThreadSafeQueue<int> &tsq) {
for (int i = 0;; ++i) {
std::cout << "Producing: " << i << std::endl;
tsq.push(i);
std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
}
void consumer(ThreadSafeQueue<int> &tsq) {
while (1) {
int value = tsq.pop();
std::cout << "Consuming: " << value << std::endl;
}
}
int main() {
ThreadSafeQueue<int> tsq;
std::thread t1(producer, std::ref(tsq));
std::thread t2(consumer, std::ref(tsq));
t1.join();
t2.join();
}