生产者消费者

发表于2021-12-24更新于2025-08-27阅读次数

简单的模型 - 产1消1

基于互斥量、条件变量以及全局标志。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
#include<iostream>
#include<thread>
#include<mutex>
#include<condition_variable>
#include<semaphore.h>
using namespace std;
mutex mtx;
condition_variable cv;
int number = 0;
bool full = false;

void produce()
{
    unique_lock<mutex> locker(mtx);
    for(int i = 0; i < 10; ++i)
    {
        while(full)
        {
            cv.wait(locker);
        }
        number = i;
        full = true;
        cout << this_thread::get_id() << "produce: " << number << endl;
        cv.notify_all();
    }
}
void consume()
{
    unique_lock<mutex> locker(mtx);
    for(int i = 0; i < 10; ++i)
    {
        while(!full)
        {
            cv.wait(locker);
        }
        cout << this_thread::get_id() << "consume: " << number << endl;
        full = false;
        cv.notify_all();
    }
}
int main()
{
    thread producer1(produce);
    thread producer2(produce);
    thread producer3(produce);
    thread consumer1(consume);
    thread consumer2(consume);
    thread consumer3(consume);
    producer1.join();
    producer2.join();
    producer3.join();
    consumer1.join();
    consumer2.join();
    consumer3.join();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
139780600784640produce: 0
139780592391936consume: 0
139780609177344produce: 0
139780575606528consume: 0
139780617570048produce: 0
139780575606528consume: 0
139780600784640produce: 1
139780592391936consume: 1
139780617570048produce: 1
139780592391936consume: 1
139780617570048produce: 2
139780592391936consume: 2
139780617570048produce: 3
139780592391936consume: 3
139780617570048produce: 4
139780592391936consume: 4
139780617570048produce: 5
139780592391936consume: 5
139780617570048produce: 6
139780592391936consume: 6
139780617570048produce: 7
139780592391936consume: 7
139780617570048produce: 8
139780592391936consume: 8
139780600784640produce: 2
139780583999232consume: 2
139780609177344produce: 1
139780583999232consume: 1
139780609177344produce: 2
139780583999232consume: 2
139780600784640produce: 3
139780583999232consume: 3
139780600784640produce: 4
139780583999232consume: 4
139780609177344produce: 3
139780575606528consume: 3
139780600784640produce: 5
139780583999232consume: 5
139780600784640produce: 6
139780575606528consume: 6
139780600784640produce: 7
139780583999232consume: 7
139780600784640produce: 8
139780575606528consume: 8
139780600784640produce: 9
139780583999232consume: 9
139780609177344produce: 4
139780575606528consume: 4
139780609177344produce: 5
139780575606528consume: 5
139780609177344produce: 6
139780583999232consume: 6
139780609177344produce: 7
139780575606528consume: 7
139780609177344produce: 8
139780583999232consume: 8
139780609177344produce: 9
139780575606528consume: 9
139780617570048produce: 9
139780575606528consume: 9

循环队列+信号量

图示

image-20211224144301657

线程不安全代码

这是线程不安全的版本。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
#include<iostream>
#include<thread>
#include<mutex>
#include<semaphore>
template<class Type>
class Queue
{
private:
    Queue(const Queue&) = delete;
    Queue& operator=(const Queue&) = delete;
public:
    Queue(int sz = 8) :data(NULL), maxsize(sz), front(0), tail(0), cursize(0)
    {
        data = (Type*)malloc(sizeof(Type) * maxsize);
        if (NULL == data) exit(1);
    }
    ~Queue()
    {
        free(data);
        data = NULL;
        maxsize = cursize = front = tail = 0;
    }
    int Cursize() const { return cursize; }
    bool IsEmpty() const { return cursize == 0; }
    int MaxSize() const { return maxsize; }
    bool IsFull() const { return cursize == maxsize; }
    Type& Front() { return data[front]; }
    const Type& Front() const { return data[front]; }
    Type& Back() { return data[tail]; }
    const Type& Back() const { return data[tail]; }

    bool Push(const Type& x)
    {
        if (IsFull()) return false;
        data[tail] = x;
        cursize += 1;
        tail = (tail + 1) % maxsize;   // 确保了取值范围在 0 ~ 7
        return true;
    }
    bool Pop()
    {
        if (IsEmpty()) return false;
        front = (front + 1) % maxsize; // 确保了取值范围在 0 ~ 7
        cursize -= 1;
        return true;
    }

private:
    Type* data;
    int maxsize;
    // 队头
    int front;
    // 队尾
    int tail;
    int cursize;
    mutable std::mutex mtx;
};
constinit int product_num = 100;
constinit int n = 8;
Queue<int> qu(n);
std::counting_semaphore produce_sema(8); // 生产者信号量
std::counting_semaphore consume_sema(0); // 消费者信号量
// 下载器
void produce()
{
    for (int i = 0; i < product_num; ++i)
    {
        produce_sema.acquire();
        std::cout << "producer: " << i << std::endl;
        qu.Push(i);
        consume_sema.release();
    }
}
// 播放器
void consume()
{
    for (int i = 0; i < product_num; ++i)
    {
        consume_sema.acquire();
        std::cout << "consumer: " << qu.Front() << std::endl;
        qu.Pop();
        produce_sema.release();
    }
}
int main()
{
    std::thread produce_th(produce);
    std::thread consume_th(consume);

    produce_th.join();
    consume_th.join();
}

结果

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
producer: 0
producer: 1
producer: 2
producer: 3
consumer: 0
consumer: 1
consumer: 2
consumer: 3
producer: 4
producer: 5
producer: 6
producer: 7consumer: 4
consumer: 5
consumer: 6

producer: 8
producer: 9
producer: 10
producer: 11
producer: 12
producer: 13
producer: 14
consumer: 7
consumer: 8
consumer: 9
consumer: 10
consumer: 11
consumer: 12
consumer: 13
consumer: 14
producer: 15
producer: 16
producer: 17
producer: 18
producer: 19
producer: 20
consumer: 15
consumer: 16
consumer: 17
consumer: 18
consumer: 19
consumer: 20
producer: 21
producer: 22
producer: 23
producer: 24
consumer: 21
consumer: 22
consumer: 23
consumer: 24
producer: 25
producer: 26
producer: 27
consumer: 25
consumer: 26
consumer: 27
producer: 28
producer: 29
producer: 30
producer: 31
producer: 32
producer: 33
producer: 34
producer: 35
consumer: 28
consumer: 29
consumer: 30
consumer: 31
consumer: 32
consumer: 33
producer: 36
producer: 37
producer: 38
producer: 39
consumer: 34
consumer: 35
consumer: 36
consumer: 37
consumer: 38
consumer: 39
producer: 40
producer: 41
producer: 42
producer: 43
producer: 44
producer: 45
producer: 46
producer: 47
consumer: 40
consumer: 41
consumer: 42
consumer: 43
consumer: 44
consumer: 45
consumer: 46
producer: 48
producer: 49
producer: 50
producer: 51
producer: 52
producer: 53
consumer: 47
consumer: 48
consumer: 49
producer: 54
producer: 55
producer: 56
producer: 57
consumer: 50
consumer: producer: 58
51
consumer: 52
consumer: 53
producer: 59
producer: 60
producer: 61
consumer: 54
consumer: 55
consumer: 56
consumer: 57
consumer: 58
consumer: 59
consumer: 60
consumer: 61
producer: 62
producer: 63
producer: 64
consumer: 62
consumer: 63
consumer: 64
producer: 65
producer: 66
producer: 67
producer: 68
producer: 69
producer: 70
producer: 71
producer: 72
consumer: 65
consumer: producer: 73
66
consumer: 67
consumer: 68
consumer: 69
consumer: 70
consumer: 71
consumer: 72
consumer: 73
producer: 74
producer: 75
producer: 76
producer: 77
producer: 78
producer: 79
producer: 80
producer: 81
consumer: 74
consumer: 75
consumer: 76
consumer: 77
consumer: 78
consumer: 79
consumer: 80
consumer: 81
producer: 82
producer: 83
producer: 84
producer: 85
producer: 86
producer: 87
producer: 88
producer: 89
consumer: 82
consumer: 83
consumer: 84
consumer: 85
consumer: 86
consumer: 87
consumer: 88
consumer: 89
producer: 90
producer: 91
producer: 92
producer: 93
producer: 94
producer: 95
producer: 96
producer: 97
consumer: 90
consumer: 91
consumer: 92
consumer: 93
consumer: 94
consumer: 95
consumer: 96
consumer: 97
producer: 98
producer: 99
consumer: 98
consumer: 99

加mutex代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
#include<iostream>
#include<thread>
#include<mutex>
#include<semaphore>
#include<print>
template<class Type>
class Queue
{
private:
    Queue(const Queue&) = delete;
    Queue& operator=(const Queue&) = delete;
public:
    Queue(int sz = 8) :data(NULL), maxsize(sz), front(0), tail(0), cursize(0)
    {
        data = (Type*)malloc(sizeof(Type) * maxsize);
        if (NULL == data) exit(1);
    }
    ~Queue()
    {
        free(data);
        data = NULL;
        maxsize = cursize = front = tail = 0;
    }
    int Cursize() const 
    {
        std::lock_guard<std::mutex> lock(mtx);
        return cursize;
    }
    bool IsEmpty() const
    {
        return Cursize() == 0;
    }
    int MaxSize() const
    {
        return maxsize;
    }
    bool IsFull() const
    {
        return Cursize() == maxsize;
    }
    Type& Front()
    {
        std::lock_guard<std::mutex> lock(mtx);
        return data[front];
    }
    const Type& Front() const
    {
        std::lock_guard<std::mutex> lock(mtx);
        return data[front];
    }
    Type& Back()
    {
        std::lock_guard<std::mutex> lock(mtx);
        return data[tail];
    }
    const Type& Back() const
    {
        std::lock_guard<std::mutex> lock(mtx);
        return data[tail];
    }

    bool Push(const Type& x)
    {
        if (IsFull()) return false;
        std::lock_guard<std::mutex> lock(mtx);
        data[tail] = x;
        cursize += 1;
        tail = (tail + 1) % maxsize;   // 确保了取值范围在 0 ~ 7
        return true;
    }
    bool Pop()
    {
        if (IsEmpty()) return false;
        std::lock_guard<std::mutex> lock(mtx);
        front = (front + 1) % maxsize; // 确保了取值范围在 0 ~ 7
        cursize -= 1;
        return true;
    }

private:
    Type* data;
    int maxsize;
    // 队头
    int front;
    // 队尾
    int tail;
    int cursize;
    mutable std::mutex mtx;
};
constinit int product_num = 100;
constinit int n = 8;
Queue<int> qu(n);
std::counting_semaphore produce_sema(8); // 生产者信号量
std::counting_semaphore consume_sema(0); // 消费者信号量
// 下载器
void produce()
{
    for (int i = 0; i < product_num; ++i)
    {
        produce_sema.acquire();
        std::println("{} {}", "producer:", i);
        qu.Push(i);
        consume_sema.release();
    }
}
// 播放器
void consume()
{
    for (int i = 0; i < product_num; ++i)
    {
        consume_sema.acquire();
        std::println("{} {}", "consumer:", qu.Front());
        qu.Pop();
        produce_sema.release();
    }
}
int main()
{
    std::thread produce_th(produce);
    std::thread consume_th(consume);

    produce_th.join();
    consume_th.join();
}

结果：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
producer: 0
producer: 1
producer: 2
producer: 3
producer: 4
consumer: 0
consumer: 1
consumer: 2
producer: 5
producer: 6
producer: 7
producer: 8
consumer: 3
producer: 9
producer: 10
producer: 11
consumer: 4
consumer: 5
consumer: 6
producer: 12
consumer: 7
consumer: 8
consumer: 9
consumer: 10
consumer: 11
producer: 13
producer: 14
consumer: 12
consumer: 13
producer: 15
consumer: 14
consumer: 15
producer: 16
producer: 17
producer: 18
producer: 19
consumer: 16
producer: 20
producer: 21
producer: 22
producer: 23
producer: 24
consumer: 17
consumer: 18
consumer: 19
producer: 25
producer: 26
producer: 27
consumer: 20
consumer: 21
producer: 28
consumer: 22
consumer: 23
producer: 29
producer: 30
consumer: 24
consumer: 25
consumer: 26
consumer: 27
producer: 31
producer: 32
consumer: 28
producer: 33
producer: 34
producer: 35
producer: 36
consumer: 29
consumer: 30
consumer: 31
producer: 37
producer: 38
consumer: 32
consumer: 33
consumer: 34
consumer: 35
consumer: 36
producer: 39
producer: 40
consumer: 37
consumer: 38
consumer: 39
consumer: 40
producer: 41
producer: 42
producer: 43
consumer: 41
consumer: 42
consumer: 43
producer: 44
producer: 45
producer: 46
producer: 47
consumer: 44
producer: 48
producer: 49
consumer: 45
consumer: 46
producer: 50
consumer: 47
consumer: 48
consumer: 49
consumer: 50
producer: 51
producer: 52
producer: 53
producer: 54
consumer: 51
consumer: 52
consumer: 53
consumer: 54
producer: 55
producer: 56
producer: 57
producer: 58
producer: 59
producer: 60
consumer: 55
producer: 61
consumer: 56
producer: 62
consumer: 57
consumer: 58
producer: 63
producer: 64
consumer: 59
consumer: 60
consumer: 61
consumer: 62
consumer: 63
producer: 65
consumer: 64
producer: 66
producer: 67
producer: 68
consumer: 65
consumer: 66
consumer: 67
consumer: 68
producer: 69
producer: 70
producer: 71
producer: 72
producer: 73
producer: 74
producer: 75
consumer: 69
consumer: 70
consumer: 71
consumer: 72
consumer: 73
producer: 76
producer: 77
consumer: 74
consumer: 75
consumer: 76
producer: 78
producer: 79
producer: 80
consumer: 77
consumer: 78
consumer: 79
consumer: 80
producer: 81
producer: 82
consumer: 81
consumer: 82
producer: 83
producer: 84
producer: 85
producer: 86
consumer: 83
producer: 87
producer: 88
consumer: 84
producer: 89
producer: 90
producer: 91
producer: 92
consumer: 85
consumer: 86
consumer: 87
producer: 93
producer: 94
producer: 95
consumer: 88
consumer: 89
consumer: 90
consumer: 91
producer: 96
producer: 97
producer: 98
consumer: 92
consumer: 93
producer: 99
consumer: 94
consumer: 95
consumer: 96
consumer: 97
consumer: 98
consumer: 99

无锁队列

共有三个线程：观察者线程（调度线程）、生产者线程、消费者线程。

共有三个队列。

生产者给第一个队列生产数据，若生产完毕或这个队列已满，生产者通知观察者。观察者则调度队列，把存放数据的队列移交到消费者端，把第二个队列（空队列）给生产者，再继续生产。若第二个队列也满了，也移交给消费者端，把第三个队列（空队列）给生产者。如此，消费者消费的过程中，生产者仍可以不断生产，直到没有空的队列。

相应地，只要有空的队列或已使用过的队列，则移交给生产者端。（把队列入队，有点像二维队列的感觉）

这个模型中，控制线程起关键桥梁作用，通知线程间信息交流。

其次，这个模型的好处在于，相比简单的生产者消费者模型，不是单纯地对同一个缓冲区容器进行读写。可以进行分库、分表，还可把读写操作分开（根据模式进行轮换），粒度变细，提高并发性能。

下去查资料

  1. 总线锁的汇编实现
  2. 缓冲锁的汇编实现