Linux 多线程 生产者&消费者 问题

在 Linux 系统中，生产者和消费者问题是一个经典的多线程同步问题，用于描述如何在多线程环境中协调多个线程对共享资源的访问。这个问题通常涉及两个类型的线程：生产者线程和消费者线程。生产者线程负责生成数据并将其放入缓冲区，而消费者线程则从缓冲区中取出数据进行处理。

生产者和消费者问题的基本概念

1. 生产者（Producer）：

   • 生成数据并将其放入共享缓冲区。
   • 生产者必须在缓冲区满时等待，直到有空间可用。

2. 消费者（Consumer）：

   • 从共享缓冲区中取出数据进行处理。
   • 消费者必须在缓冲区空时等待，直到有数据可用。

共享资源和同步机制

为了确保线程安全并避免竞态条件，生产者和消费者通常使用以下同步机制：

1. 互斥锁（Mutex）：

   • 用于确保只有一个线程可以在任何给定时间内访问共享缓冲区。
   • 通过锁定和解锁操作来实现。

2. 条件变量（Condition Variable）：

   • 用于让线程在某些条件下等待或唤醒。
   • 生产者在缓冲区不满时发信号唤醒消费者，消费者在缓冲区有数据时发信号唤醒生产者。

让我们用一个更生动的例子来解释生产者-消费者问题中的条件变量和互斥锁的工作机制。

例子：餐厅的厨师和服务员

想象一下，有一个餐厅，里面有一个厨房（缓冲区）和一个用餐区。餐厅有两种工作人员：厨师（生产者）和服务员（消费者）。他们之间的协调就像在处理生产者-消费者问题。

场景设置

• 厨房有一个有限大小的桌子（缓冲区），桌子上可以放一定数量的菜（数据）。
• 厨师做菜（生产数据）并把菜放到桌子上。
• 服务员从桌子上拿菜（消费数据）并把菜送到顾客那里。
• 如果桌子满了，厨师就不能放菜，需要等待服务员拿走菜。
• 如果桌子空了，服务员就不能拿菜，需要等待厨师放菜。

具体流程

1. 厨师做菜并放在桌子上：

   • 厨师首先检查桌子上有没有空位。
   • 如果桌子满了，厨师就等待（进入等待状态）。
   • 如果桌子有空位，厨师把菜放在桌子上，然后通知服务员桌子上有新菜了。
   • 厨师然后继续做下一道菜。

2. 服务员拿菜并送到顾客那里：

   • 服务员首先检查桌子上有没有菜。
   • 如果桌子空了，服务员就等待（进入等待状态）。
   • 如果桌子上有菜，服务员拿走菜并送到顾客那里，然后通知厨师桌子上有空位了。
   • 服务员然后继续拿下一道菜。

代码对应的具体场景

以下代码片段展示了这个过程：

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
 
#define BUFFER_SIZE 10
 
int buffer[BUFFER_SIZE];
int count = 0;
 
pthread_mutex_t mutex;
pthread_cond_t cond_producer;
pthread_cond_t cond_consumer;
 
void* producer(void* arg) {
    while (1) {
        // 生成数据（例如：随机数）
        int item = rand() % 100;
 
        pthread_mutex_lock(&mutex);
 
        // 等待缓冲区有空位
        while (count == BUFFER_SIZE) {
            pthread_cond_wait(&cond_producer, &mutex);
        }
 
        // 将数据放入缓冲区
        buffer[count++] = item;
        printf("Produced: %d\n", item);
 
        // 通知消费者有数据可用
        pthread_cond_signal(&cond_consumer);
        
        pthread_mutex_unlock(&mutex);
 
        // 模拟生产时间
        sleep(1);
    }
}
 
void* consumer(void* arg) {
    while (1) {
        pthread_mutex_lock(&mutex);
 
        // 等待缓冲区有数据
        while (count == 0) {
            pthread_cond_wait(&cond_consumer, &mutex);
        }
 
        // 从缓冲区取出数据
        int item = buffer[--count];
        printf("Consumed: %d\n", item);
 
        // 通知生产者有空位可用
        pthread_cond_signal(&cond_producer);
        
        pthread_mutex_unlock(&mutex);
 
        // 模拟消费时间
        sleep(1);
    }
}
 
int main() {
    pthread_t producer_thread, consumer_thread;
 
    pthread_mutex_init(&mutex, NULL);
    pthread_cond_init(&cond_producer, NULL);
    pthread_cond_init(&cond_consumer, NULL);
 
    pthread_create(&producer_thread, NULL, producer, NULL);
    pthread_create(&consumer_thread, NULL, consumer, NULL);
 
    pthread_join(producer_thread, NULL);
    pthread_join(consumer_thread, NULL);
 
    pthread_mutex_destroy(&mutex);
    pthread_cond_destroy(&cond_producer);
    pthread_cond_destroy(&cond_consumer);
 
    return 0;
}

运行结果：

解释

1. 互斥锁（pthread_mutex_t mutex）：

   • 互斥锁就像厨房的门，确保只有一个厨师或服务员可以进出厨房，以防止混乱。

2. 条件变量（pthread_cond_t cond_producer 和 pthread_cond_t cond_consumer）：

   • 条件变量就像门口的信号灯。
   • cond_producer 是厨师用来等待服务员拿走菜的信号灯。
   • cond_consumer 是服务员用来等待厨师放菜的信号灯。

3. 厨师（producer）：

   • 厨师检查厨房的桌子（缓冲区）是否满了。
   • 如果满了，厨师等待（等待cond_producer信号）。
   • 如果有空位，厨师放菜在桌子上，并通知服务员（发送cond_consumer信号）。
   • 然后，厨师离开厨房（解锁互斥锁），继续做下一道菜。

4. 服务员（consumer）：

   • 服务员检查厨房的桌子（缓冲区）是否有菜。
   • 如果桌子空了，服务员等待（等待cond_consumer信号）。
   • 如果有菜，服务员拿走菜，并通知厨师（发送cond_producer信号）。
   • 然后，服务员离开厨房（解锁互斥锁），继续送菜给顾客。