[已解决]多线程无锁设计的实现：生产者-消费者问题

cjgank

1、问题描述：由UNIX网络编程中生产者-消费者问题延申，生产者线程和消费者线程不利用锁机制（互斥锁、读写锁）能否实现呢？

个人思路：通过数据结构来避免锁的使用，但是有一个前提条件：消费者线程读取（和处理）数据的速度要比生产者线程写入数据要快，否则将造成数据丢失。

2、实现单生产者-单消费者情况（c\c++）



3、实现单生产者-多消费者情况（c\c++）


4、有大佬在项目中遇到过上述问题并付诸实践了吗？期待大家的帮助，十分感谢。

最佳答案

月排行榜 / 总排行榜

Mike_python小

2023-10-19 10:14:45

生产者-消费者问题的多线程无锁设计实现：

1. 思路：
   通过使用数据结构来避免锁的使用，可以使用队列作为数据结构，并设置一个缓冲区来存放生产者生成的数据。在生产者向缓冲区中写入数据时，需要判断缓冲区是否已满，如果已满则等待一段时间再尝试写入；在消费者从缓冲区中读取数据时，需要判断缓冲区是否为空，如果为空则等待一段时间再尝试读取。

2. 单生产者-单消费者情况（C/C++）：

1. ++
2. #include <iostream>
3. #include <vector>
4. #include <thread>
5. #include <chrono>
6. #include <mutex>
7. #include <condition_variable>
9. using namespace std;
11. class CircularQueue {
12. public:
13.     CircularQueue(int capacity) : _capacity(capacity), _head(0), _tail(0), _size(0), _queue(capacity) {}
15.     void push(int data) {
16.         unique_lock<mutex> lock(_mutex);
17.         // 缓冲区已满，等待
18.         while (_size == _capacity) {
19.             _cond_full.wait(lock);
20.         }
21.         _queue[_tail] = data;
22.         _tail = (_tail + 1) % _capacity;
23.         ++_size;
24.         _cond_empty.notify_one();  // 唤醒等待读取数据的消费者
25.     }
27.     int pop() {
28.         unique_lock<mutex> lock(_mutex);
29.         // 缓冲区为空，等待
30.         while (_size == 0) {
31.             _cond_empty.wait(lock);
32.         }
33.         int data = _queue[_head];
34.         _head = (_head + 1) % _capacity;
35.         --_size;
36.         _cond_full.notify_one(); // 唤醒等待写入数据的生产者
37.         return data;
38.     }
40. private:
41.     const int _capacity;
42.     int _head;
43.     int _tail;
44.     int _size;
45.     vector<int> _queue;
46.     mutex _mutex;
47.     condition_variable _cond_empty;
48.     condition_variable _cond_full;
49. };
51. void producer(CircularQueue& q) {
52.     for (int i = 0; i < 10; ++i) {
53.         this_thread::sleep_for(chrono::seconds(1));  // 模拟生产者生成数据的时间
54.         q.push(i);
55.         cout << "producer produce data: " << i << endl;
56.     }
57. }
59. void consumer(CircularQueue& q) {
60.     for (int i = 0; i < 10; ++i) {
61.         int data = q.pop();
62.         cout << "consumer consume data: " << data << endl;
63.     }
64. }
66. int main() {
67.     CircularQueue q(5);  // 缓冲区容量为5
68.     thread t1(producer, ref(q));
69.     thread t2(consumer, ref(q));
70.     t1.join();
71.     t2.join();
72.     return 0;
73. }

复制代码

3. 单生产者-多消费者情况（C/C++）：

1. ++
2. #include <iostream>
3. #include <vector>
4. #include <thread>
5. #include <chrono>
6. #include <mutex>
7. #include <condition_variable>
9. using namespace std;
11. class CircularQueue {
12. public:
13.     CircularQueue(int capacity) : _capacity(capacity), _head(0), _tail(0), _size(0), _queue(capacity) {}
15.     void push(int data) {
16.         unique_lock<mutex> lock(_mutex);
17.         // 缓冲区已满，等待
18.         while (_size == _capacity) {
19.             _cond_full.wait(lock);
20.         }
21.         _queue[_tail] = data;
22.         _tail = (_tail + 1) % _capacity;
23.         ++_size;
24.         _cond_empty.notify_one(); // 唤醒等待读取数据的消费者
25.     }
27.     int pop() {
28.         unique_lock<mutex> lock(_mutex);
29.         // 缓冲区为空，等待
30.         while (_size == 0) {
31.             _cond_empty.wait(lock);
32.         }
33.         int data = _queue[_head];
34.         _head = (_head + 1) % _capacity;
35.         --_size;
36.         _cond_full.notify_one(); // 唤醒等待写入数据的生产者
37.         return data;
38.     }
40. private:
41.     const int _capacity;
42.     int _head;
43.     int _tail;
44.     int _size;
45.     vector<int> _queue;
46.     mutex _mutex;
47.     condition_variable _cond_empty;
48.     condition_variable _cond_full;
49. };
51. void producer(CircularQueue& q) {
52.     for (int i = 0; i < 10; ++i) {
53.         this_thread::sleep_for(chrono::seconds(1));  // 模拟生产者生成数据的时间
54.         q.push(i);
55.         cout << "producer produce data: " << i << endl;
56.     }
57. }
59. void consumer(CircularQueue& q, int id) {
60.     for (int i = 0; i < 5; ++i) {
61.         int data = q.pop();
62.         cout << "consumer " << id << " consume data: " << data << endl;
63.     }
64. }
66. int main() {
67.     CircularQueue q(3);  // 缓冲区容量为3
68.     thread t1(producer, ref(q));
69.     thread t2(consumer, ref(q), 1);
70.     thread t3(consumer, ref(q), 2);
71.     t1.join();
72.     t2.join();
73.     t3.join();
74.     return 0;
75. }

复制代码

4. 实践经验：
   实际项目中，使用无锁的生产者-消费者模型可以提高多线程程序的效率和性能。但是需要注意的是，如果消费者线程读取（和处理）数据的速度比生产者线程写入数据的速度还要慢，那么将会出现数据丢失的问题。此时需要考虑其他的解决方案，例如：增加缓冲区的容量、调整生产者与消费者的速度等。

球一个最佳答案谢谢啦！这对我非常重要！

跳转到最佳答案楼层