C++如何实现并发队列 C++线程安全队列的实现

#include <iostream>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <thread>

template <typename T>
class ConcurrentQueue {
private:
    std::queue<T> q;
    std::mutex m;
    std::condition_variable cv;

public:
    void enqueue(T value) {
        {
            std::lock_guard<std::mutex> lock(m);
            q.push(value);
        }
        cv.notify_one();
    }

    T dequeue() {
        std::unique_lock<std::mutex> lock(m);
        cv.wait(lock, [this]{ return !q.empty(); });
        T value = q.front();
        q.pop();
        return value;
    }

    bool try_dequeue(T& value) {
        std::lock_guard<std::mutex> lock(m);
        if (q.empty()) {
            return false;
        }
        value = q.front();
        q.pop();
        return true;
    }

    bool empty() const {
        std::lock_guard<std::mutex> lock(m);
        return q.empty();
    }
};

int main() {
    ConcurrentQueue<int> cq;

    std::thread producer([&]() {
        for (int i = 0; i < 10; ++i) {
            cq.enqueue(i);
            std::cout << "Enqueued: " << i << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
        }
    });

    std::thread consumer([&]() {
        for (int i = 0; i < 10; ++i) {
            int value = cq.dequeue();
            std::cout << "Dequeued: " << value << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(200));
        }
    });

    producer.join();
    consumer.join();

    return 0;
}

#include <iostream>
#include <thread>
#include <vector>
#include <random>
#include <chrono>

// 假设已经有了ConcurrentQueue的定义

int main() {
    ConcurrentQueue<int> cq;
    const int num_threads = 8;
    const int num_operations = 10000;

    std::vector<std::thread> threads;

    // 生产者线程
    for (int i = 0; i < num_threads / 2; ++i) {
        threads.emplace_back([&]() {
            std::random_device rd;
            std::mt19937 gen(rd());
            std::uniform_int_distribution<> distrib(1, 100);

            for (int j = 0; j < num_operations; ++j) {
                int value = distrib(gen);
                cq.enqueue(value);
                //std::cout << "Thread " << std::this_thread::get_id() << " enqueued: " << value << std::endl; // 调试用，正式环境移除
                std::this_thread::sleep_for(std::chrono::microseconds(distrib(gen))); // 模拟不同步的生产速度
            }
        });
    }

    // 消费者线程
    for (int i = num_threads / 2; i < num_threads; ++i) {
        threads.emplace_back([&]() {
            std::random_device rd;
            std::mt19937 gen(rd());
            std::uniform_int_distribution<> distrib(1, 100);
            int value;
            for (int j = 0; j < num_operations; ++j) {
                if (cq.try_dequeue(value)) {
                    //std::cout << "Thread " << std::this_thread::get_id() << " dequeued: " << value << std::endl; // 调试用，正式环境移除
                    std::this_thread::sleep_for(std::chrono::microseconds(distrib(gen))); // 模拟不同步的消费速度
                } else {
                   //std::cout << "Thread " << std::this_thread::get_id() << " failed to dequeue (empty)" << std::endl; // 调试用，正式环境移除
                   std::this_thread::yield(); // 让出时间片，避免空转
                }
            }
        });
    }

    for (auto& thread : threads) {
        thread.join();
    }

    std::cout << "All threads finished." << std::endl;

    // 最终检查队列是否为空，以及进行其他必要的验证
    while (!cq.empty()) {
      int value;
      if (cq.try_dequeue(value)) {
        std::cout << "Remaining value in queue: " << value << std::endl; // 如果队列不为空，打印剩余元素
      } else {
        std::cout << "Unexpected error: Queue is not empty but dequeue failed." << std::endl;
        break;
      }
    }


    return 0;
}