C++ 并发专题 - 实现一个线程安全的队列

一：概述

        本文使用 C++ 标准库中的多线程、条件变量、互斥锁等工具来实现一个线程安全的队列，而且使用多个线程来向队列中添加和获取数据。 
二：实现过程：

2. #include <iostream>
3. #include <queue>
4. #include <mutex>
5. #include <condition_variable>
6. #include <thread>
7. #include <vector>
9. template <typename T>
10. class ThreadSafeQueue {
11. public:
12.     // 向队列中添加元素
13.     void push(const T& value) {
14.         std::lock_guard<std::mutex> lock(mutex_);
15.         queue_.push(value);
16.         cond_var_.notify_one();  // 通知一个等待的线程
17.     }
19.     // 从队列中取出元素，如果队列为空，阻塞等待
20.     T pop() {
21.         std::unique_lock<std::mutex> lock(mutex_);
22.         cond_var_.wait(lock, [this] { return !queue_.empty(); });  // 等待直到队列非空
23.         T value = queue_.front();
24.         queue_.pop();
25.         return value;
26.     }
28.     // 判断队列是否为空
29.     bool empty() const {
30.         std::lock_guard<std::mutex> lock(mutex_);
31.         return queue_.empty();
32.     }
34. private:
35.     mutable std::mutex mutex_;            // 互斥锁，保护队列
36.     std::queue<T> queue_;                 // 基础队列
37.     std::condition_variable cond_var_;   // 条件变量，用于队列为空时的等待
38. };
40. // 示例：使用线程安全队列
41. void producer(ThreadSafeQueue<int>& queue, int numItems) {
42.     for (int i = 0; i < numItems; ++i) {
43.         queue.push(i);
44.         std::cout << "Produced: " << i << std::endl;
45.     }
46. }
48. void consumer(ThreadSafeQueue<int>& queue, int numItems) {
49.     for (int i = 0; i < numItems; ++i) {
50.         int item = queue.pop();
51.         std::cout << "Consumed: " << item << std::endl;
52.     }
53. }
55. int main() {
56.     ThreadSafeQueue<int> queue;
58.     const int numItems = 10;
59.     const int numProducers = 2;
60.     const int numConsumers = 2;
62.     std::vector<std::thread> threads;
64.     // 启动生产者线程
65.     for (int i = 0; i < numProducers; ++i) {
66.         threads.push_back(std::thread(producer, std::ref(queue), numItems / numProducers));
67.     }
69.     // 启动消费者线程
70.     for (int i = 0; i < numConsumers; ++i) {
71.         threads.push_back(std::thread(consumer, std::ref(queue), numItems / numConsumers));
72.     }
74.     // 等待所有线程完成
75.     for (auto& thr : threads) {
76.         thr.join();
77.     }
79.     return 0;
80. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。