Not Only Algorithm,不仅仅是算法,关注数学、算法、数据结构、程序员笔试面试以及一切涉及计算机编程之美的内容 。。
你的位置:NoAlGo博客 » 程序设计 » 

生产者消费者问题

生产者消费者问题是一个经典的进程同步问题,本文讲解不同生产者消费者问题的解决思路,并提供一个完整的实现代码。

问题定义

生产者消费者问题(Producer-consumer problem,也称有限缓冲问题,Bounded-buffer problem),是一个多线程同步问题的经典案例。该问题描述了两个共享固定大小缓冲区的线程,即所谓的“生产者”和“消费者”,在实际运行时会发生的问题。生产者的主要作用是重复生成一定量的数据放到缓冲区中,与此同时,消费者也在缓冲区消耗这些数据。该问题的关键就是要保证生产者不会在缓冲区满时加入数据,消费者也不会在缓冲区中空时消耗数据。

解决这个问题要涉及到操作系统中进程的同步与互斥问题。

使用linux中的pthread库进行多线程编程,其中使用pthread_create创建新线程,pthread_join来等待线程结束。

使用信号量进行多线程的同步操作,信号量包括两个操作原语:

  • wait(P):将信号量的值减一,如果新的值大于或等于1,该进程继续执行,否该进程被阻塞,进入等待队列。
  • post(V):将信号量的值加一,如果新的值大于0,该进程继续执行,否则唤醒阻塞队列中的一个线程。

通过信号量可以对共享资源进行保护。把信号量的值初始化成共享资源的数量,当要使用共享资源时,可以使用wait操作进行申请,只有当有剩余资源的时候才能够申请成功,否则要进行等待。而当使用完资源时,需要使用post操作进行释放,使资源的计数值加一,如果释放时有人在等待资源,则将其唤醒。

在Linux中使用sem_t表示信号量,先使用sem_init初始化,然后使用sem_wait进行申请,使用sem_post进行释放,另外还有其他的API接口如下,含义可以根据名称得到:

int sem_init(sem_t *sem,int pshared,unsigned int value)
int sem_destroy(sem_t *sem)
int sem_wait(sem_t *sem)
int sem_trywait(sem_t *sem)
int sem_post(sem_t *sem)
int sem_getvalue(sem_t *sem)

互斥锁是一个特殊的信号量,即二元信号量,其只允许一个人访问共享资源。在Linux中使用pthread_mutex_t进行表示,其API接口如下所示,具体使用方法参照样例程序。

int pthread_mutex_init(pthread_mutex_t *mutex, const pthread_mutexattr_t *mutexattr)
int pthread_mutex_destroy(pthread_mutex_t *mutex)
int pthread_mutex_lock(pthread_mutex_t *mutex)
int pthread_mutex_unlock(pthread_mutex_t *mutex)
int pthread_mutex_trylock(pthread_mutex_t *mutex)

具体问题

下面讲解几种不同的生产者消费者模型。

一个生产者,一个消费者,一个缓冲区

只有一个生产者和消费者,我们定义两个信号量进行同步:

  • empty:缓冲区是否为空,初值为1。
  • full: 缓冲区是否为满,初值为0。

于是,当缓存区非满时才可以进行生产,当非空时才可以进行消费,其算法伪代码如下:

producerFunc:
while (true):
	sem_wait(empty);
	put staff to buf;
	sem_post(full);

consumerFunc:
while (true):
	sem_wait(full);
	get staff from buf;
	sem_post(empty);

一个生产者,一个消费者,n个环形缓冲区

由于有n个缓冲区,使用环形队列进行模拟,需要两个指针指明缓冲区当前的位置:

  • in:缓冲区头指针,放入物品时加一。
  • out:缓冲区为指针,取出物品时加一。

由于是环形缓冲区,当指针到达末尾时需要重新回到头部,利用取模操作解决。

producerFunc:
while (true):
	sem_wait(empty);
	put staff to buf[in];
	increase buf front pointer, in = (in + 1) % n;
	sem_post(full);

consumerFunc:
while (true):
	sem_wait(full);
	get staff from buf[out];
	decrease buf end pointer, out = (out + 1) % n;
	sem_post(empty);

若干生产者,若干消费者,n个环形缓冲区

由于生产者和消费者都不止一个,不同的生成者和消费者不能同时访问缓冲区,使用一个互斥锁mutex进行控制。

producerFunc:
while (true):
	sem_wait(empty);
	lock(mutex);
	put staff to buf[in];
	increase buf front pointer, in = (in + 1) % n;
	unlock(mutex);
	sem_post(full);
	
consumerFunc:
while (true):
	sem_wait(full);
	lock(mutex);
	get staff from buf[out];
	decrease buf end pointer, out = (out + 1) % n;
	unlock(mutex);
	sem_post(empty);

代码实现

下面是实现上述第三个问题的C++代码实现。

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <pthread.h>
#include <semaphore.h>
#include <signal.h>

const int NrProd = 3, NrCons = 2; //生产者、消费者数量
const int BufLen = 6;             //缓冲区长度

int buf[BufLen], in = 0, out = 0; //缓存区,首尾指针
sem_t empty, full;                //信号量
pthread_mutex_t mutex;            //互斥锁

int producerID = 0, consumerID = 0;//生产者、消费者ID

void handler(int signo)
{
	printf("Exit program.\n");
	exit(0);
}

void *producerFunc(void *arg) //生产者进程
{
	int id = producerID++;
	while (true)
	{
		sleep(2);

		sem_wait(&empty);
		pthread_mutex_lock(&mutex);
		
		buf[in] = 1;
		printf("Producer %d produces %d:", id, in);
		for (int i = 0; i < BufLen; ++i)
			if (buf[i]) printf(" %d", i);
		printf("\n");
		in = (in + 1) % BufLen;

		pthread_mutex_unlock(&mutex);
		sem_post(&full);
	}
}

void *consumerFunc(void * arg) //消费者进程
{
	int id = consumerID++;
	while (true)
	{
		sleep(2);

		sem_wait(&full);
		pthread_mutex_lock(&mutex);

		buf[out] = 0;
		printf("Consumer %d consumes %d:", id, out);
		for (int i = 0; i < BufLen; ++i)
			if (buf[i]) printf(" %d", i);
		printf("\n");
		out = (out + 1) % BufLen;

		pthread_mutex_unlock(&mutex);
		sem_post(&empty);
	}
}

int main()
{
	if (signal(SIGINT, handler) == SIG_ERR)
	{
		printf("Signal error.\n");
		exit(1);
	}

	if (sem_init(&empty, 0, BufLen) != 0)
	{
		printf("Init empty error.\n");
		exit(1);
	}

	if (sem_init(&full, 0, 0) != 0)
	{

		printf("Init full error.\n");
		exit(1);
	}

	pthread_t producer[NrProd];
	for (int i = 0; i < NrProd; ++i)
		if (pthread_create(&producer[i], NULL, producerFunc, NULL) != 0)
		{
			printf("Creating producer %d error.\n", i);
			exit(1);
		}
	
	pthread_t consumer[NrCons];
	for (int i = 0; i < NrCons; ++i)
		if (pthread_create(&consumer[i], NULL, consumerFunc, NULL) != 0)
		{
			printf("Creating consumer %d error.\n", i);
			exit(1);
		}
	
	for (int i = 0; i < NrProd; ++i) pthread_join(producer[i], NULL);
	for (int i = 0; i < NrCons; ++i) pthread_join(consumer[i], NULL);

	return 0;
}

下面是上述程序的输出结果。

Producer 0 produces 0: 0
Consumer 0 consumes 0:
Producer 1 produces 1: 1
Producer 2 produces 2: 1 2
Consumer 1 consumes 1: 2
Producer 0 produces 3: 2 3
Producer 2 produces 4: 2 3 4
Producer 1 produces 5: 2 3 4 5
Consumer 0 consumes 2: 3 4 5
Consumer 1 consumes 3: 4 5
Producer 0 produces 0: 0 4 5
Producer 1 produces 1: 0 1 4 5
Producer 2 produces 2: 0 1 2 4 5
Consumer 0 consumes 4: 0 1 2 5
Consumer 1 consumes 5: 0 1 2
^CExit program.
上一篇: 下一篇:

我的博客

NoAlGo头像编程这件小事牵扯到太多的知识,很容易知其然而不知其所以然,但真正了不起的程序员对自己程序的每一个字节都了如指掌,要立足基础理论,努力提升自我的专业修养。

站内搜索

最新评论