生产者/消费者(producer/consumer)问题,也叫作有界缓冲区(bounded buffer)问题。
假设有一个或多个生产者线程和一个或多个消费者线程。生产者把生成的数据项放入缓冲区;消费者从缓冲区取走数据项,以某种方式消费。
很多实际的系统中都会有这种场景。例如,在多线程的网络服务器中,一个生产者将HTTP 请求放入工作队列(即有界缓冲区),消费线程从队列中取走请求并处理。因为有界缓冲区是共享资源,所以我们必须通过同步机制来访问它,以免产生竞态条件。

实现环形缓冲区

首先需要一个共享缓冲区,让生产者放入数据,消费者取出数据。简单起见,我们使用一个数组来做缓冲区,put函数将值放入缓冲区,get函数从缓冲区取值。使用 fill变量记录放入缓冲区的位置,使用use变量记录get函数从缓冲区取值的位置,使用count变量记录缓冲区的使用情况。使用fill = (fill + 1) % MAX取余函数让缓冲区成环,节约空间使用。

int buffer[MAX]; 
int fill = 0; 
int use = 0; 
int count = 0;  //记录缓冲区的使用情况

void put(int value) { 
	buffer[fill] = value; 
	fill = (fill + 1) % MAX; // 当前位置是最后时,下次指向第一个位置
	count++; 
} 

int get() { 
	int tmp = buffer[use]; 
	use = (use + 1) % MAX; 
	count--; 
	return tmp; 
}

初始化锁和条件变量

初始化线程锁,条件变量empty和full.
当缓冲区满时,使用full唤醒消费者线程
当缓冲区空时,使用empty唤醒生产者线程

std::mutex mtx;
std::condition_variable empty;
std::condition_variable full; 

实现生产者线程函数

首先获取可操作缓冲区的锁,

  1. 当缓冲区满时,生产者线程进行等待,并释放缓冲区锁
  2. 当缓冲区不满的时候,生产者线程使用put函数向缓冲区存储一个数据,并通知消费者线程,现在缓冲区已经有数据了,可以进行消费了
void producer(std::string arg) { 
	int i=0; 
	while(1) { 
        {
            std::unique_lock<std::mutex> lock(mtx);
            while(count == MAX)
                empty.wait(lock);
            put(i); // p4 
            std::cout<<arg<<" "<<i++<<std::endl; 
            full.notify_all(); // p5 
        }
	} 
} 

实现消费者线程函数

首先获取可操作缓冲区的锁,

  1. 当缓冲区空时,消费者线程进行等待,并释放缓冲区锁
  2. 当缓冲区有数据时,生产者线程使用get函数从缓冲区取出一个数据,并通知生产者线程,现在缓冲区不是满的了,可以进行生产了
void consumer(std::string arg) { 
	while(1) {
        {
            std::unique_lock<std::mutex> lock(mtx);
            while (count == 0)
                full.wait(lock);
            int tmp = get(); 
            std::cout<<arg<<" "<<tmp<<std::endl; //帮助打印是哪个消费者信息
            empty.notify_all();
        }
		//sleep(1);
	} 
}

下面是使用示例

#include <iostream>
#include <thread>
#include <mutex> 
#include <condition_variable>
#include <string>

#define MAX 10
int buffer[MAX]; 
int fill = 0; 
int use = 0; 
int count = 0; 

void put(int value) { 
	buffer[fill] = value; 
	fill = (fill + 1) % MAX; 
	count++; 
} 

int get() { 
	int tmp = buffer[use]; 
	use = (use + 1) % MAX; 
	count--; 
	return tmp; 
}
std::mutex mtx;
std::condition_variable empty;
std::condition_variable full; 

void producer(std::string arg) { 
	int i=0; 
	while(1) { 
        {
            std::unique_lock<std::mutex> lock(mtx);
            while(count == MAX)
                empty.wait(lock);
            put(i); // p4 
            std::cout<<arg<<" "<<i++<<std::endl; 
            full.notify_all(); // p5 
        }
	} 
} 

void consumer(std::string arg) { 
	while(1) {
        {
            std::unique_lock<std::mutex> lock(mtx);
            while (count == 0)
                full.wait(lock);
            int tmp = get(); 
            std::cout<<arg<<" "<<tmp<<std::endl; //帮助打印是哪个消费者信息
            empty.notify_all();
        }
		//sleep(1);
	} 
}

int main(int argc,char *argv[])
{
	
    std::thread p1(producer, "producer1");
    std::thread c1(consumer, "consumer1");
    std::thread c2(consumer, "consumer2"); //帮助打印显示信息
    p1.join();
    c1.join();
    c2.join();
	return 0;
}

更多推荐