C++多线程和循环队列

假设我们不使用互斥锁，并且我们在两个线程中分别调用 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 操作。下面是可能出现的问题的几种情况：

1. 数据竞争: 两个线程可能同时访问和修改 front 和 rear 变量，导致它们的值不一致。例如：

   • 线程 A (producer) 在插入一个新元素时，刚刚修改了 rear，还没来得及更新 items[rear]；
   • 线程 B (consumer) 可能同时读取 front 并进行删除操作，这时 front 和 rear 之间的关系可能不一致。

2. 丢失数据: 由于没有同步，enqueue 操作可能会覆盖队列中的数据，而 dequeue 操作可能会读取错误的数据或读取到无效的数据。例如：

   • 线程 A 在 rear 更新前被中断，线程 B 读取到的是尚未正确插入的数据；
   • 线程 B 在 front 更新前被中断，线程 A 插入的数据可能会被错误地覆盖。

3. 未定义行为: 多线程并发修改共享数据时，可能会导致程序行为不可预测甚至崩溃。

用互斥锁来解决这个问题

#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();
}