Linux系统：线程同步与生产消费模型

---

前言

线程同步是指在多线程环境中，通过特定机制控制线程的执行顺序和资源访问，避免因线程并发执行导致的数据混乱或逻辑错误。简单来说，当多个线程共享资源（如全局变量、文件等）时，如果不加以控制，可能会出现多个线程同时修改同一资源或线程访问到未准备好的资源等问题。线程同步就是解决这类问题的技术。

---

一，条件变量

条件变量是多线程编程中用于线程间同步的一种机制，主要解决线程需要等待某个条件成立才能继续执行的场景。它通常与互斥锁Mutex配合使用，实现线程间的等待 - 通知逻辑。

---

1-1 条件变量的核心作用

想象这样一个场景：

• 消费者线程需要等待队列中有数据才能消费（否则就该休眠）
• 生产者线程生产数据后，需要通知消费者可以来消费了

条件变量正是为这种场景设计的，它能让线程：

• 在条件不满足时进入等待状态（释放 CPU 资 源，不占用空转）
• 当条件满足时被其他线程唤醒（继续执行后续操作）

条件变量必须与互斥锁配合使用，原因是：检查条件和等待操作需要是原子操作（避免线程在检查条件后、进入等待前被其他线程打断）

以生产者 - 消费者模型为例，关键步骤如下：

消费者线程：

pthread_mutex_lock(&mutex);       // 加锁
while (条件不满足) {              // 循环检查条件（避免虚假唤醒）pthread_cond_wait(&cond, &mutex);  // 释放锁并进入等待，被唤醒时重新获取锁
}
// 执行操作（如消费数据）
pthread_mutex_unlock(&mutex);     // 解锁

生产者线程：

pthread_mutex_lock(&mutex);       // 加锁
// 执行操作（如生产数据，使条件满足）
pthread_cond_signal(&cond);       // 唤醒一个等待的线程
// 或 pthread_cond_broadcast(&cond); 唤醒所有等待的线程
pthread_mutex_unlock(&mutex);     // 解锁

---

1-2 条件变量初始化

在 POSIX 线程（pthread）中，条件变量（pthread_cond_t）的初始化有两种常用方式，适用于不同场景：

• 静态初始化（推荐简单场景使用）
  直接通过宏 PTHREAD_COND_INITIALIZER 初始化，语法简单，无需手动销毁。

用法示例：

#include <pthread.h>
// 静态初始化条件变量
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

特点：

• 适用于全局变量或静态变量（编译时初始化）

• 无需调用 pthread_cond_destroy 销毁（但调用也不会出错）

• 内部会将条件变量初始化为默认状态

• 动态初始化（推荐复杂场景使用）
  通过函数 pthread_cond_init 初始化，可自定义属性，灵活性更高。

用法示例：

#include <pthread.h>
#include <stdio.h>int main() {pthread_cond_t cond;pthread_condattr_t attr;  // 条件变量属性（可选）// 初始化属性（默认属性可省略此步）if (pthread_condattr_init(&attr) != 0) {perror("condattr_init failed");return 1;}// 动态初始化条件变量（第二个参数为属性，NULL 表示使用默认属性）if (pthread_cond_init(&cond, &attr) != 0) {perror("cond_init failed");return 1;}// 使用完毕后，必须销毁条件变量和属性pthread_cond_destroy(&cond);pthread_condattr_destroy(&attr);return 0;
}

特点：

• 适用于局部变量或需要自定义属性的场景（如进程间共享）
• 必须与 pthread_cond_destroy 配对使用（否则可能导致资源泄漏）
• 通过属性对象 pthread_condattr_t 可设置特殊行为（如进程共享）

演示代码：

#include<pthread.h>
#include<stdio.h>
pthread_mutex_t mutex=PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond=PTHREAD_COND_INITIALIZER;
int queue[10];
int count=0;
void* comsumer(void*arg)
{while(1){pthread_mutex_lock(&mutex);while(count==0){pthread_cond_wait(&cond,&mutex);}int data=queue[--count];printf("消费数据:%d\n",data);pthread_mutex_unlock(&mutex);}return NULL;
}
void *producer(void* arg)
{for(int i=0;i<5;i++){pthread_mutex_lock(&mutex);queue[count++]=i;printf("生产数据:%d\n",i);pthread_cond_signal(&cond);pthread_mutex_unlock(&mutex);}return NULL;
}
int main()
{pthread_t c,p;pthread_create(&c,NULL,comsumer,NULL);pthread_create(&p,NULL,producer,NULL);pthread_join(c,NULL);pthread_join(p,NULL);return 0;
}

演示结果：

gch@hcss-ecs-f59a:/gch/code/HaoHao/learn2/day6$ ./exe
生产数据:0
消费数据:0
生产数据:1
消费数据:1
生产数据:2
消费数据:2
生产数据:3
消费数据:3
生产数据:4
消费数据:4

这里我们每生产一个资源就发信号pthread_cond_signal(&cond)给消费者消费。消费者消费完了之后会用pthread_cond_wait(&cond,&mutex)进行等待

---

二，生产者消费者模型

生产者消费者模型是多线程 / 多进程编程中最经典的同步协作模型，核心解决 “生产数据” 和 “消费数据” 两类角色的解耦与高效协作问题，广泛用于缓存、消息队列、任务池等场景（比如 Linux 内核的工作队列、Redis 的任务处理）。

---

2-1 核心概念：3 个角色

模型本质是通过一个 “中间缓冲区”（也叫 “仓库”），让生产者（负责生成数据）和消费者（负责处理数据）完全解耦 —— 生产者不直接给消费者传数据，消费者也不直接找生产者要数据，二者仅与缓冲区交互。

生产者： 生成数据，放入缓冲区 缓冲区满时，必须等待，不能继续生产
消费者： 从缓冲区取数据，处理数据 缓冲区空时，必须等待，不能继续消费
中间缓冲区： 存储生产者的 “待消费数据”，通常是数组 / 链表 保证线程安全，同一时间只有一个角色操作

---

2-2 为什么需要这个模型

解决 “生产速度” 和 “消费速度不匹配” 的问题，同时实现解耦：

• 解耦：生产者和消费者无需知道对方的存在，只需关注 “自己与缓冲区的交互”（比如生产者只需关心 “能不能往缓冲区放数据”，不用管 “谁来消费”）。
• 削峰填谷：若生产者突然生产大量数据（峰），缓冲区可暂存，避免消费者处理不过来导致数据丢失；若生产者生产慢（谷），消费者可从缓冲区取之前暂存的数据，避免空闲。
• 提高效率：生产者和消费者可并行执行（比如生产者生产时，消费者同时处理之前的缓存数据），比 “生产者生产一个、消费者立即处理一个” 的串行模式效率更高。

---

2-3 如何保证线程安全

生产者和消费者都会操作 “中间缓冲区”（临界资源），必须通过同步机制解决两个核心问题：

• 互斥：同一时间，只有一个线程能操作缓冲区（比如生产者放数据时，消费者不能取；反之亦然）—— 通常用互斥锁（pthread_mutex_t） 实现。
• 同步：
  • 缓冲区满时，生产者必须 “等待”，直到消费者取走数据（缓冲区有空闲）；
  • 缓冲区空时，消费者必须 “等待”，直到生产者放入数据（缓冲区有数据）；
  • 通常用条件变量（pthread_cond_t） 实现 “等待 - 唤醒” 逻辑。

---

2-4 Linux 下的代码示例

以 “单生产者 + 单消费者” 为例，用数组作为缓冲区，配合互斥锁和条件变量实现安全协作。
代码逻辑：

• 定义缓冲区（大小固定为 5）、互斥锁（保护缓冲区）、两个条件变量（分别通知 “缓冲区非空”“缓冲区非满”）；
• 生产者线程：循环生成数据，若缓冲区满则等待，否则放入数据并唤醒消费者；
• 消费者线程：循环取数据，若缓冲区空则等待，否则取数据并唤醒生产者；

#include <stdio.h>
#include <pthread.h>
#include <unistd.h>// 1. 定义模型核心组件
#define BUF_SIZE 5  // 缓冲区大小
int buffer[BUF_SIZE];  // 中间缓冲区（存储数据）
int count = 0;         // 缓冲区中当前数据个数（0=空，BUF_SIZE=满）pthread_mutex_t mutex;                // 互斥锁：保护缓冲区操作
pthread_cond_t cond_not_full;         // 条件变量：通知“缓冲区非满”（生产者等）
pthread_cond_t cond_not_empty;       // 条件变量：通知“缓冲区非空”（消费者等）// 2. 生产者线程：生成1-10的数字，放入缓冲区
void* producer(void* arg) {int data;for (data = 1; data <= 10; data++) {  // 生产10个数据后退出// 加锁：操作缓冲区前必须先获取互斥锁pthread_mutex_lock(&mutex);// 若缓冲区满，等待“消费者取数据”（释放锁，阻塞等待）while (count == BUF_SIZE) {printf("生产者：缓冲区满，等待...\n");// 等待时会释放mutex，被唤醒后重新获取mutexpthread_cond_wait(&cond_not_full, &mutex);}// 放入数据到缓冲区（此时已持有锁，安全）buffer[count] = data;count++;printf("生产者：生产数据 %d，缓冲区剩余空间：%d\n", data, BUF_SIZE - count);// 解锁前，唤醒消费者（告知“缓冲区非空，可以取数据了”）pthread_cond_signal(&cond_not_empty);pthread_mutex_unlock(&mutex);sleep(1);  // 模拟生产耗时（比如读取文件、计算数据）}return NULL;
}// 3. 消费者线程：从缓冲区取数据并处理
void* consumer(void* arg) {int data;while (1) {  // 循环消费（实际场景可加退出条件）// 加锁：操作缓冲区前必须先获取互斥锁pthread_mutex_lock(&mutex);// 若缓冲区空，等待“生产者放数据”（释放锁，阻塞等待）while (count == 0) {printf("消费者：缓冲区空，等待...\n");pthread_cond_wait(&cond_not_empty, &mutex);}// 从缓冲区取数据（此时已持有锁，安全）count--;data = buffer[count];  // 从末尾取（简化逻辑）printf("消费者：消费数据 %d，缓冲区剩余数据：%d\n", data, count);// 解锁前，唤醒生产者（告知“缓冲区非满，可以放数据了”）pthread_cond_signal(&cond_not_full);pthread_mutex_unlock(&mutex);sleep(2);  // 模拟消费耗时（比如处理数据、写入数据库）}return NULL;
}int main() {pthread_t prod_tid, cons_tid;// 初始化互斥锁和条件变量pthread_mutex_init(&mutex, NULL);pthread_cond_init(&cond_not_full, NULL);pthread_cond_init(&cond_not_empty, NULL);// 创建生产者和消费者线程pthread_create(&prod_tid, NULL, producer, NULL);pthread_create(&cons_tid, NULL, consumer, NULL);// 等待线程结束（生产者生产10个数据后退出，消费者可手动终止）pthread_join(prod_tid, NULL);pthread_join(cons_tid, NULL);  // 实际运行时，可按Ctrl+C终止// 销毁资源pthread_mutex_destroy(&mutex);pthread_cond_destroy(&cond_not_full);pthread_cond_destroy(&cond_not_empty);return 0;
}

---

演示结果：

gch@hcss-ecs-f59a:/gch/code/HaoHao/learn2/day6$ ./exe
生产者：生产数据 1，缓冲区剩余空间：4
消费者：消费数据 1，缓冲区剩余数据：0
生产者：生产数据 2，缓冲区剩余空间：4
消费者：消费数据 2，缓冲区剩余数据：0
生产者：生产数据 3，缓冲区剩余空间：4
生产者：生产数据 4，缓冲区剩余空间：3
消费者：消费数据 4，缓冲区剩余数据：1
生产者：生产数据 5，缓冲区剩余空间：3
生产者：生产数据 6，缓冲区剩余空间：2
消费者：消费数据 6，缓冲区剩余数据：2
生产者：生产数据 7，缓冲区剩余空间：2
生产者：生产数据 8，缓冲区剩余空间：1
消费者：消费数据 8，缓冲区剩余数据：3
生产者：生产数据 9，缓冲区剩余空间：1
生产者：生产数据 10，缓冲区剩余空间：0
消费者：消费数据 10，缓冲区剩余数据：4
消费者：消费数据 9，缓冲区剩余数据：3
消费者：消费数据 7，缓冲区剩余数据：2
消费者：消费数据 5，缓冲区剩余数据：1
消费者：消费数据 3，缓冲区剩余数据：0
消费者：缓冲区空，等待...

这里一共会生产10个数据，如果数据不超过5个，每隔一秒生产一个数据，如果缓存区不为空，消费者每两秒消费一个，这两个线程是同步线程

---

2-5 为什么 pthread_cond_wait 需要互斥量

• pthread_cond_wait 需要互斥量mutex，核心原因是为了解决 “条件检查” 和 “等待操作” 的原子性问题，避免多线程环境下的竞态条件。

• 用一个生活场景类比：
  你去餐厅排队点餐，需要先确认 “当前有没有空位”（检查条件），如果没有就 “排队等待”（进入等待状态）。这两个动作必须连续完成 —— 如果刚确认 “没位置”，还没来得及排队，就被别人插队占了最后一个空位，你就会错误地进入等待（实际已经有位置了）。

• 对应到代码中，pthread_cond_wait 必须和互斥量配合，就是为了防止这种 “检查条件后、等待前被其他线程插队” 的问题

具体原因拆解：

• 条件检查需要互斥：线程调用 pthread_cond_wait 前，必须先检查条件（比如 count == 0）。这个检查操作需要在互斥锁保护下进行，否则：
  • 线程 A 刚检查完 count == 0（准备等待），还没进入 pthread_cond_wait；
  • 线程 B 此时生产了数据（count 变为 1）并调用 pthread_cond_signal（但此时 A 还没开始等待，信号会 “丢失”）
  • 线程 A 随后进入等待，永远收不到信号，导致 “假等”。
• 等待时必须释放锁，唤醒后必须重新获锁
  pthread_cond_wait 的内部逻辑是：
  • 先自动释放互斥锁（让其他线程有机会修改条件，比如生产者可以加数据）；
  • 然后阻塞等待信号；
  • 被唤醒后，自动重新获取互斥锁（保证后续操作仍在互斥保护下）。如果没有互斥锁，就无法实现 “释放锁→等待→重获锁” 的原子操作链，会导致线程永久阻塞或条件检查失效。
• 避免 “虚假唤醒”：操作系统可能会在无信号的情况下唤醒线程（称为 “虚假唤醒”）。因此，被唤醒后必须重新检查条件，而重新检查需要在互斥锁保护下进行（防止其他线程同时修改条件）。

代码示例：

pthread_mutex_lock(&mutex);       // 加锁
while (count == 0) {             // 循环检查条件（被唤醒后也要要检查）pthread_cond_wait(&cond, &mutex);  // 释放锁→等待→被唤醒后重新获锁
}
// 操作共享资源（此时已持有锁，安全）
pthread_mutex_unlock(&mutex);     // 解锁

pthread_cond_wait 需要互斥量，是为了确保 “条件检查” 和 “等待操作” 的原子性，防止信号丢失和虚假唤醒导致的逻辑错误，最终保证多线程环境下条件同步的安全性。