【Linux】线程间同步实践 —— 生产消费模型-linux运维-PHP中文网

【Linux】线程间同步实践 —— 生产消费模型

星夢妙者

发布： 2025-07-20 11:20:12

原创

794人浏览过

在线程互斥的场景中，我们使用锁来确保只有一个线程能够访问临界区，从而保证了线程安全。然而，仅依赖一把锁并不能保证线程访问的顺序性。例如，在学校的vip自习室（仅限一人使用）中，钥匙存放在门口，学生可以取走钥匙进入自习室，离开时将钥匙放回原处。假设小明离开自习室后，将钥匙放回原处，但随后立即又拿起钥匙再次进入自习室，如此反复，其他学生对此表示不满。

这种情况表明，仅凭一把锁无法确保线程执行的顺序性，因此需要进行同步，以确保所有人访问自习室时不仅安全，还具有一定的顺序性。同样地，线程之间也可以通过条件变量实现同步。

条件变量

int pthread_cond_init(pthread_cond_t *restrict cond, const pthread_condattr_t *restrictattr);: 初始化接口 int pthread_cond_destroy(pthread_cond_t *cond): 销毁接口 int pthread_cond_wait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex);: 在条件不满足时阻塞等待 int pthread_cond_broadcast(pthread_cond_t *cond);: 条件满足时，唤醒所有线程，开始竞争 int pthread_cond_signal(pthread_cond_t *cond);: 条件满足时，唤醒一个线程。条件变量需要一个线程队列和相应的通知机制，才能保证线程同步。

生产消费模型

2.1 什么是生产消费模型

生产消费模型可以通过一个生动的例子来解释：

生产者：作为方便面的生产者进行生产，生产有一定的限制，生产量满足需求后不再生产。平台：作为方便面的销售平台，从生产者那里获得方便面，并销售给消费者。作为中间人，调控方便面的数量需求。消费者：作为方便面的消费者，从平台获取方便面。通过平台的中转，可以实现生产与消费的解耦，通过中间平台可以快速将产品提供给消费者，又可以在库存不足时通知生产者进行生产。

当然，生产者和消费者可能有多个，这会产生竞争关系，通过线程同步（锁与条件变量）来协调，从而支持并发处理。

总结，生产消费模型遵循"321"原则：

一个交易场所（特定数据结构形式存在的一段内存空间）两种角色（生产角色，消费角色）：生产线程，消费线程三种关系：生产与生产（互斥关系），消费与消费（互斥关系），生产与消费。

生产者消费者模型的优点

解耦支持并发支持忙闲不均

2.2 为何要使用生产消费模型

生产者消费者模式通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而是通过阻塞队列进行通讯。因此，生产者生产完数据后无需等待消费者处理，直接将数据放入阻塞队列；消费者不直接向生产者索要数据，而是直接从阻塞队列中取出。阻塞队列相当于一个缓冲区，平衡了生产者和消费者的处理能力。这个阻塞队列就是用来给生产者和消费者解耦的。

实践生产消费模型 — 阻塞队列

在多线程编程中，阻塞队列（Blocking Queue）是一种常用于实现生产者和消费者模型的数据结构。其与普通队列的区别在于，当队列为空时，从队列中获取元素的操作将被阻塞，直到队列中有元素被放入；当队列满时，向队列中存放元素的操作也会被阻塞，直到有元素被从队列中取出（以上操作都是基于不同的线程来说的，线程在对阻塞队列进行操作时会被阻塞）。

3.1 框架搭建

阻塞队列的本质仍然是队列，因此底层需要一个队列来存储数据（使用模板来适配各种类型的数据）。根据实际需求，这个队列不能存储过多数据，需要有一个边界值来限制。此外，为了保证临界区操作的安全，需要使用锁来保护。

百川大模型

百川智能公司推出的一系列大型语言模型产品

查看详情

#pragma once
#include <mutex>
#include <queue>
#include <pthread.h>
#include <unistd.h>
<p>const int num = 5;</p><p>template <class T>
class BlockQueue {
private:
// 判断函数
bool Full() {
return _bq.size() >= _max_cp;
}
bool Empty() {
return _bq.size() == 0;
}</p><p>public:
// 构造函数
BlockQueue()
: _max_cp(num)
{
pthread_mutex_init(&_mtx, nullptr);
}</p><pre class="brush:php;toolbar:false;"><code>// 析构函数
~BlockQueue()
{
    pthread_mutex_destroy(&_mtx);
}

void Pop(T* data) {
}

void Equeue(const T data) {
}

登录后复制

private: // 队列来存储数据 std::queue<T> _bq; // 加锁 pthread_mutex_t _mtx; int _max_cp; };

我们主要要实现的就是两个接口：

Pop: 消费者使用，用来获取一个数据 Equeue: 生产者使用，用来存入数据

3.2 Pop与Equeue

在Pop与Equeue中进行的操作：

判断是否可以获取/插入，涉及判断就是非原子操作，需要加锁！在不可以的情况下就要进行阻塞（阻塞后会自动解锁），等待被唤醒！因此需要加入两个条件变量来进行判断！进行获取/插入唤醒生产者/消费者，唤醒对应的条件变量即可！

void Pop(T<em> data) {
pthread_mutex_lock(&_mtx);
// 队列没有数据，阻塞等待
// while保证代码的鲁棒性
// 函数返回也要再次进行判断，保证不为空！
while (Empty()) {
pthread_cond_wait(&_c_cond, &_mtx);
}
// 被唤醒了 || 队列中有数据
// 出队列
</em>data = _bq.front();
_bq.pop();
pthread_mutex_unlock(&_mtx);
// 读取走了可以唤醒生产者
pthread_cond_signal(&_p_cond);
}</p><p>void Equeue(const T data) {
// 插入和判断都是临界区操作，上锁保证线程安全
pthread_mutex_lock(&_mtx);
// 如果容量满了，就要阻塞，等待消费者消费
// 函数返回也要再次进行判断，保证没有满！
while (Full()) {
// 满了，生产者不能生产，必须等待
// 这里是临界区！！！阻塞后会自动解锁
// 被调用的时候，除了让自己继续排队等待，还会释放传入的锁！
// 函数返回时，回到临界区，会参与锁的竞争，获取到锁之后进行返回！
pthread_cond_wait(&_p_cond, &_mtx);
}
// 入队列
_bq.push(data);
pthread_mutex_unlock(&_mtx);
// 入队列，可以唤醒消费者
pthread_cond_signal(&_c_cond);
}

登录后复制

注意：

pthread_mutex_unlock(&_mtx);
pthread_cond_signal(&_c_cond);

登录后复制

这两步操作分别是解锁和唤醒条件变量。这两个函数的顺序没有要求。因为唤醒一个线程，都要进行竞争锁，只有对应的阻塞的线程获得到锁才会返回。因此，唤醒与解锁的顺序并不影响后续线程的运行。

3.3 测试运行

我们来测试一下使用两个线程来进行消费和生产的：

#include "BlockQueue.hpp"</p><h1>include <stdlib.h></h1><h1>include <iostream></h1><p>void<em> Consumer(void</em> args) {
srand(time(nullptr) ^ getpid());
BlockQueue<int><em> bq = static_cast<BlockQueue<int></em>>(args);
while (true) {
int data;
bq->Pop(&data);
std::cout << "消费者消费了：" << data << std::endl;
sleep(1);
}
return nullptr;
}</p><p>void<em> Productor(void</em> args) {
srand(time(nullptr) ^ getpid());
BlockQueue<int><em> bq = static_cast<BlockQueue<int></em>>(args);
while (true) {
int data = rand() % 10;
bq->Equeue(data);
std::cout << "生产者生产了：" << data << std::endl;
sleep(1);
}
return nullptr;
}</p><p>int main() {
BlockQueue<int> bq;
pthread_t c, p;
pthread_create(&c, nullptr, Consumer, &bq);
pthread_create(&p, nullptr, Productor, &bq);
pthread_join(c, nullptr);
pthread_join(p, nullptr);
return 0;
}

登录后复制

【Linux】线程间同步实践 —— 生产消费模型