Linux多线程编程（三）：条件变量与阻塞队列生产者消费者模型

Linux 多线程编程（三）：条件变量与生产者消费者模型详解

深入理解线程同步机制，掌握条件变量的正确用法，并用阻塞队列实现经典的生产者消费者模型

前言

在多线程编程中，线程之间不仅需要互斥（Mutual Exclusion）来保护共享资源，还需要同步（Synchronization）来协调执行的先后顺序。如果同步设计不当，就会出现竞态条件（Race Condition），导致程序行为异常甚至崩溃。

条件变量（Condition Variable）是 POSIX 线程库提供的一种同步机制，它允许线程在某个条件不满足时主动挂起，直到其他线程改变条件并发出通知。本文将从原理到实践，全面讲解条件变量的使用，并基于阻塞队列实现经典的生产者消费者模型。

一、同步与竞态条件

1.1 什么是同步

同步是指：在保证数据安全的前提下，让线程能够按照某种特定的顺序访问临界资源，从而有效避免**饥饿（Starvation）**问题。简单来说，同步就是为了让线程“有序协作”。

例如，一个线程负责往队列里放数据，另一个线程负责取数据。如果队列为空，取数据的线程就应该等待，而不是反复尝试或报错；当生产者放入数据后，消费者被唤醒并开始工作。这种“先生产再消费”的次序就是同步的体现。

1.2 竞态条件

竞态条件是指：由于多个线程访问共享资源时的时序问题导致程序出现不可预期的结果。比如两个线程同时修改同一个全局变量，最终结果取决于谁先执行、谁后执行，这就是典型的竞态。

竞态条件是并发程序的第一大敌，它不仅难以重现，而且难以调试。因此，我们必须通过互斥锁 + 条件变量来消除竞态。

二、条件变量（Condition Variable）

2.1 为什么需要条件变量

考虑一个场景：线程 A 想要从队列中取数据，但队列为空。如果线程 A 采用**忙等待（Busy Waiting）**循环检测，会浪费大量 CPU；如果只是加锁、检查、解锁、再睡一会儿，又可能导致延迟或信号丢失。

条件变量正是为解决此类问题而生：它允许线程在条件不满足时阻塞（等待），直到其他线程改变了条件并通过条件变量唤醒等待的线程。

2.2 条件变量的核心函数

静态初始化可用 PTHREAD_COND_INITIALIZER 宏，无需调用 pthread_cond_init。

2.3 为什么 pthread_cond_wait 必须与互斥锁配合

关键点在于 “解锁 + 等待”必须是原子操作。如果不加互斥锁，可能出现：

pthread_cond_wait 将“释放锁”和“挂起等待”合并为一个原子操作，绝不会有空隙让信号丢失。同样，它返回时会重新获取锁，确保线程在检查条件时持有锁。

2.4 条件变量的使用规范（标准模式）

等待条件（以消费者为例）

pthread_mutex_lock(&mutex);
while (队列为空) { // 必须用 while，不能用 if
pthread_cond_wait(&cond, &mutex);
}
// 条件满足，执行操作（例如取出数据）
pthread_mutex_unlock(&mutex);

发送条件信号（以生产者为例）

pthread_mutex_lock(&mutex);
// 修改共享变量（例如入队）
pthread_cond_signal(&cond); // 或 broadcast
pthread_mutex_unlock(&mutex);

为什么用 while 而不是 if？
原因有二：一是虚假唤醒（Spurious Wakeup），POSIX 允许 pthread_cond_wait 在没有信号的情况下返回；二是即使是被信号唤醒，也可能在等待期间有其他线程抢先修改了条件，导致条件再次不满足。while 保证了每次被唤醒后都重新验证条件。

三、生产者消费者模型（Producer-Consumer Model）

3.1 模型背景与优点

生产者消费者模式通过一个 缓冲区（阻塞队列） 来解耦生产者和消费者：

• 生产者将数据放入缓冲区，无需关心消费者是谁、何时消费；
• 消费者从缓冲区取出数据，无需关心生产者的节奏。

这种设计的优点包括：

• 解耦：生产者和消费者不直接依赖，可独立演化。
• 支持并发：可以同时有多个生产者和消费者，提高吞吐量。
• 忙闲不均：缓冲区可以平抑波峰波谷，生产者快时数据暂存，消费者快时等待。

记忆口诀：321 原则

• 3 种关系：生产者与生产者（互斥）、消费者与消费者（互斥）、生产者与消费者（同步+互斥）
• 2 个角色：生产者和消费者
• 1 个缓冲区（阻塞队列）

3.2 使用 C++ 封装阻塞队列（BlockingQueue）

我们将基于 std::queue<int>，加上互斥锁和两个条件变量（full 和 empty）来实现一个线程安全的阻塞队列。

• full 条件变量：当队列满时，生产者在此等待。
• empty 条件变量：当队列空时，消费者在此等待。
  

下面是完整代码（单生产者单消费者示例，可轻松扩展为多对多）：

#include <iostream>
#include <queue>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>
#include <time.h>

#define NUM 8 // 队列容量

class BlockQueue {
private:
std::queue<int> q;
int cap;
pthread_mutex_t lock;
pthread_cond_t full; // 队列满时，生产者等待
pthread_cond_t empty; // 队列空时，消费者等待

private:
void LockQueue() { pthread_mutex_lock(&lock); }
void UnLockQueue() { pthread_mutex_unlock(&lock); }

void ProductWait() { pthread_cond_wait(&full, &lock); }
void ConsumeWait() { pthread_cond_wait(&empty, &lock); }

void NotifyProduct() { pthread_cond_signal(&full); }
void NotifyConsume() { pthread_cond_signal(&empty); }

bool IsEmpty() { return q.empty(); }
bool IsFull() { return (q.size() == cap); }

public:
BlockQueue(int _cap = NUM) : cap(_cap) {
pthread_mutex_init(&lock, NULL);
pthread_cond_init(&full, NULL);
pthread_cond_init(&empty, NULL);
}

~BlockQueue() {
pthread_mutex_destroy(&lock);
pthread_cond_destroy(&full);
pthread_cond_destroy(&empty);
}

// 生产者放入数据
void PushData(const int &data) {
LockQueue();
while (IsFull()) { // 满则等待
std::cout << "队列已满，生产者等待…" << std::endl;
ProductWait(); // 原子释放锁并等待
}
q.push(data);
std::cout << "生产者生产数据: " << data << std::endl;
NotifyConsume(); // 唤醒消费者
UnLockQueue();
}

// 消费者取出数据
void PopData(int &data) {
LockQueue();
while (IsEmpty()) { // 空则等待
std::cout << "队列为空，消费者等待…" << std::endl;
ConsumeWait(); // 原子释放锁并等待
}
data = q.front();
q.pop();
std::cout << "消费者消费数据: " << data << std::endl;
NotifyProduct(); // 唤醒生产者
UnLockQueue();
}
};

// 消费者线程函数
void *consumer(void *arg) {
BlockQueue *bq = (BlockQueue*)arg;
int data;
for (;;) {
bq->PopData(data);
sleep(1); // 模拟消费耗时
}
return NULL;
}

// 生产者线程函数
void *producter(void *arg) {
BlockQueue *bq = (BlockQueue*)arg;
srand((unsigned)time(NULL));
for (;;) {
int data = rand() % 1024;
bq->PushData(data);
sleep(1); // 模拟生产耗时
}
return NULL;
}

int main() {
BlockQueue bq;
pthread_t c, p;

pthread_create(&c, NULL, consumer, (void*)&bq);
pthread_create(&p, NULL, producter, (void*)&bq);

pthread_join(c, NULL);
pthread_join(p, NULL);

return 0;
}

3.3 代码关键点解读

• 双条件变量设计：full 和 empty 分别用于生产者和消费者的等待，各自独立，避免无效唤醒。
• while 循环：即使在 ProductWait() 返回后，依然再次检查队列是否满，防止虚假唤醒或竞争导致队列依然满。
• 唤醒时机：每次成功 push 后唤醒一个消费者，每次成功 pop 后唤醒一个生产者。这种设计保证了只要缓冲区状态变化，就通知可能等待的另一方。
• 锁的粒度：整个操作在锁保护下进行，条件变量的等待和通知都在锁内完成，符合规范。

四、常见问题与注意事项

4.1 条件变量与互斥锁的顺序

• 必须先锁定互斥锁，再调用 pthread_cond_wait。
• pthread_cond_signal 可以在锁定或解锁状态下调用，但一般推荐在解锁前调用，这样可以减少线程上下文切换（被唤醒的线程会尝试获得锁，而锁还未释放，可能被立即转移）。

4.2 signal 与 broadcast 的选择

• signal 只唤醒一个等待线程，适用于只有一个线程能够处理资源的场景（如单个消费者）。
• broadcast 唤醒所有等待线程，适用于多个线程都能处理但需要它们重新竞争的场景，或者当条件变化剧烈时，防止个别线程被遗漏。

4.3 避免死锁

• 确保 pthread_cond_wait 的互斥锁在使用前已锁定，并且在返回后正确解锁。
• 不要在一个线程中同时持有多把锁时进行条件等待，容易引发死锁。

4.4 性能考量

• 尽量减少条件变量唤醒的次数，不必要的唤醒会增加系统开销。
• 在生产者消费者模型中，可以考虑批量生产/消费来减少锁竞争。

五、总结

条件变量是 POSIX 多线程编程中最常用的同步工具之一，它与互斥锁配合，能够高效地实现线程间的协调。使用条件变量必须遵循规范：

生产者消费者模型是条件变量的经典应用场景，通过阻塞队列作为缓冲区，它完美地解决了生产者和消费者之间的同步与互斥问题，在并发编程中具有广泛的实际价值。