操作系统 — 生产者消费者模型

不念不忘少年蓝@ 2022-06-01 05:10 257阅读 0赞

# 生产者消费者模型 #

所谓的生产者消费者模型就是一个类似于队列一样的东西串起来，这个队列可以想像成一个存放产品的"仓库"，生产者只需要关心这个

"仓库"，并 不 需要关心具体的消费者，对于生产者而言甚至都不知道有这些消费者存在. 对于消费者而言他也不需要关心具体的生产

者，到底有多少生产者也 不是 他关心的事情，他只要关心这个"仓库"中还有没有东西. 其实这个模型的应用场景非常的广泛，生活中很

多例子都是这样！ 比如中午去吃东 西，那么 点餐台就是一个小的生产者消费者模型，你定的黄焖鸡如果没有做出来之前，那么你只能

阻塞式的等待. 那么今天我们尝试实现一个生产 者 消费者模 型.

不过我们的这个生产者消费者模型跟网上大多数的代码还是有一点不一样的，我们实现的是一个基于共享内存的功能简单的消息队列.

也就是我们 使 用将消费者和生产者 一起绑定到一个共享内存块中，我们提供的是一个自定义传输块大小，自定义队列的长度和大小，

自定义类型传输. 我们的 核心 思想其实就是实现消费者和生产者之间的通信，让他们可以互相发送数据块或者是信息. 不过可能实现起

来会有一点绕. 现在开始介绍我们的 结构：

(下面代码实现多出用到共享内存和信号量，如果不了解的人类可以戳进去了解一下: 共享内存 信号量)

所谓生产者和消费者模型当中有三个最关键的细节:

1.所谓的生产者和消费者之间的关系同步于互斥的关系

2.生产者和生产者之间为互斥的关系

3.消费者和消费者之间的互斥的关系.

既然我们实现一个基于生产者消费者模型和共享内存的类消息队列，此结构是一段连续存储相同数据块的空间，而在此空间的头部我们

拥有一个叫 做shmfifo 的结构体对他们进行管理，其实也就是管理我们的"仓库"，然后你可以自己指定仓库中存储 数据的类型，数据的

大小，然后我们最后使用 共享内存来传输这些数据，又因为消费者和生产者之间有同步和互斥的关系，生产者和生产者之间有 互斥关系

，消费者同理. 我们这里使用的是信 号量来解决他们的同步和互斥. 那么我们首先来认识shmfifo结构:

typedef struct shmfifo {
    	shmhead_t *p_shm; // 共享内存的头部地址 
    	char *p_payload;  // 有效负载起始地址
    	int shmid;        // 共享内存ID
    	int sem_mutex;    // 用来互斥的信号量
    	int sem_full;     // 用来控制共享内存满的信号量
    	int sem_empty;    // 用来控制共享内存空的信号量
    }shmfifo_t;

我对于我定义的这个结构体每一项都做了注释，这个结构体存储了有效的负载起始地址，共享内存的头部地址等等. 它的里面维护了

一个shmhead\_t 结构体，这个结构体还是我们定义的一个管理读写过程的结构体,它的结构为:

typedef struct shmhead {
    	unsigned int blksize;    // 块大小
    	unsigned int blocks;     // 块总数
    	unsigned int rd_index;   // 读索引
    	unsigned int wr_index;   // 写索引
    }shmhead_t;

我们用shmhead结构体来表示仓库中访问的细节，比如他的每一个类型块有多大，他总共有多少个内存块，消费者下次来读取应该从

哪个位置开始读 取， 生产者下次来放置数据应该放在哪个数据. 这样由这两个结构体就可以维护一段空间，这段空间还是你自己定

义大小的，然后你定义对应的函 数来进行读取和放置数据函数. 所以呢！ 我们这个基于共享内存的自定义消息队列就构造成功了！

我画一张图帮我们理解:

shmfifo_t *shmfifo_init(key_t key, int blksz, int blks);  //连接到同一个消息队列仓库！
    
    void shmfifo_put(shmfifo_t *fifo, const void *buf);       //往仓库当中放置数据
    
    void shmfifo_get(shmfifo_t *fifo, void *buf);             //从仓库当中读取数据
    
    void shmfifo_destroy(shmfifo_t *fifo);			  //销毁这个仓库

![SouthEast][]

接下来我们看到我们的实现代码，我会尽量详细的注释:

/*************************************************************************
    	> File Name: shmfifo.h
    	> Author: ma6174
    	> Mail: ma6174@163.com 
    	> Created Time: Sat 13 Jan 2018 05:37:48 AM PST
     ************************************************************************/
    #ifndef __SHMFIFO_H__
    #define __SHMFIFO_H__
    
    #include <unistd.h>
    #include <sys/ipc.h>
    #include <sys/shm.h>
    #include <sys/sem.h>
    
    #include <stdio.h>
    #include <stdlib.h>
    #include <string.h>
    #include <errno.h>
    
    extern int errno;
    
    #define ERR_EXIT(msg) \
    do { \
    	fprintf(stderr, "[%s][%d] %s : %s\n",__FILE__,__LINE__,\
    						msg,strerror(errno)); \
    	exit(EXIT_FAILURE); \
    }while ( 0 )
    
    typedef struct shmhead {
    	unsigned int blksize;    // 块大小
    	unsigned int blocks;     // 块总数
    	unsigned int rd_index;   // 读索引
    	unsigned int wr_index;   // 写索引
    }shmhead_t;
    
    typedef struct shmfifo {
    	shmhead_t *p_shm; // 共享内存的头部地址 
    	char *p_payload;  // 有效负载起始地址
    	int shmid;        // 共享内存ID
    	int sem_mutex;    // 用来互斥的信号量
    	int sem_full;     // 用来控制共享内存满的信号量
    	int sem_empty;    // 用来控制共享内存空的信号量
    }shmfifo_t;
    
    shmfifo_t *shmfifo_init(key_t key, int blksz, int blks);
    void shmfifo_put(shmfifo_t *fifo, const void *buf);
    void shmfifo_get(shmfifo_t *fifo, void *buf);
    void shmfifo_destroy(shmfifo_t *fifo);
    
    #endif //__SHMFIFO_H__

/*************************************************************************
    	> File Name: shmfifo.c
    	> Author: ma6174
    	> Mail: ma6174@163.com 
    	> Created Time: Sat 13 Jan 2018 05:37:48 AM PST
     ************************************************************************/
    
    #include <assert.h>
    #include "shmfifo.h"
    
    union semun { int val; };
    
    shmfifo_t *shmfifo_init(key_t key, int blksz, int blks)
    {
    	shmfifo_t *fifo = malloc(sizeof(shmfifo_t));
    	assert(fifo);
    	memset(fifo, 0x00, sizeof(shmfifo_t));
    	
    	int shmid = shmget(key, 0, 0);
    	//如果已经存在了，那么shmid返回值不为-1.
    	int size = sizeof(shmhead_t)+blksz*blks;
    	if ( shmid == -1 ) { // shmid不存在，新创建一个我们的基于共享内存的自定义消息队列.
    		shmid = shmget(key, size, IPC_CREAT|0666);
    		if ( shmid == -1 ) ERR_EXIT("shmget");
    		fifo->shmid = shmid;
    		fifo->p_shm = shmat(shmid, NULL, 0);
    		fifo->p_payload = (char*)(fifo->p_shm+1);
    
    		fifo->p_shm->blksize  = blksz;
    		fifo->p_shm->blocks   = blks;
    		fifo->p_shm->rd_index = 0;
    		fifo->p_shm->wr_index = 0;
    		
    		fifo->sem_mutex = semget(key, 1, 0644|IPC_CREAT);
    		fifo->sem_full  = semget(key+1, 1, 0644|IPC_CREAT);
    		fifo->sem_empty = semget(key+2, 1, 0644|IPC_CREAT);
    		//初始化shmfifo结构体内容.
    		
    		union semun su;
    		su.val = 1;
    		semctl(fifo->sem_mutex, 0, SETVAL, su);
    		su.val = 0;
    		semctl(fifo->sem_empty, 0, SETVAL, su);
    		su.val = blks;
    		semctl(fifo->sem_full, 0, SETVAL, su);
    		//初始化，每一个信号量的_op.   //信号量维护的结构体里面的内容
    		
    	} else { // shmid已经存在，连接到那个共享内存块即可.
    		fifo->shmid = shmid;
    		fifo->p_shm = shmat(shmid, NULL, 0);
    		fifo->p_payload = (char*)(fifo->p_shm+1);
    		fifo->sem_mutex = semget(key, 0, 0);
    		fifo->sem_full  = semget(key+1, 0, 0);
    		fifo->sem_empty = semget(key+2, 0, 0);
    	}
    
    	return fifo;
    }
    
    //p操作
    int sem_p(int semid)
    {
    	struct sembuf buf[1] = {0, -1, 0};
    	return semop(semid, buf, 1);
    }
    
    //v操作
    int sem_v(int semid)
    {
    	struct sembuf buf[1] = {0, 1, 0};
    	return semop(semid, buf, 1);
    }
    
    //放置数据
    void shmfifo_put(shmfifo_t *fifo, const void *buf)
    {
    	//防止自己在写的时候，别的生产者也过来写入内容，执行信号量sem_mutex写操作.
    	sem_p(fifo->sem_full);
    	sem_p(fifo->sem_mutex);
    	//每次放置数据之前，让信号量sem_full计数-1，等到该计数为0时，要放置数据时使用sem_P操作，就会被阻塞，等到有空间可以写为止.
    
    	memcpy(fifo->p_payload+fifo->p_shm->wr_index*fifo->p_shm->blksize, buf, fifo->p_shm->blksize);
    	fifo->p_shm->wr_index = (fifo->p_shm->wr_index+1) % fifo->p_shm->blocks;
    	//定义下次写位置.
    
    	//放置数据结束以后，让信号量sem_empty的计数+1，这样的话仓库中就多了一个可以读取的数据.
    	sem_v(fifo->sem_mutex);
    	sem_v(fifo->sem_empty);
    }
    
    void shmfifo_get(shmfifo_t *fifo, void *buf)
    {
    	//防止自己在写的时候，别的生产者也过来写入内容，执行信号量sem_mutex写操作.
    	sem_p(fifo->sem_empty);
    	sem_p(fifo->sem_mutex);
    	//每次读取数据之前，让信号量sem_empty计数-1,等到该计数为0时，要读取数据时使用sem_P操作，就会被阻塞，等到有数据可以读为止.
    
    	
    	memcpy(buf, fifo->p_payload+fifo->p_shm->rd_index*fifo->p_shm->blksize, fifo->p_shm->blksize);
    	fifo->p_shm->rd_index = (fifo->p_shm->rd_index+1) % fifo->p_shm->blocks;
    	//定位下次读位置.
    	
    	//读取完数据之后，让信号量sem_full+1，这样的话"仓库"中多了一个空间可以放置数据
    	sem_v(fifo->sem_mutex);
    	sem_v(fifo->sem_full);
    }
    
    void shmfifo_destroy(shmfifo_t *fifo)
    {
    	semctl(fifo->sem_mutex, 0, IPC_RMID, 0);
    	semctl(fifo->sem_empty, 0, IPC_RMID, 0);
    	semctl(fifo->sem_full,  0, IPC_RMID, 0);
    	shmdt(fifo->p_shm);
    	shmctl(fifo->shmid, IPC_RMID, 0);
    	free(fifo);
    }

/*************************************************************************
    	> File Name: put.c
    	> Author: ma6174
    	> Mail: ma6174@163.com 
    	> Created Time: Sat 20 Jan 2018 10:42:33 PM PST
     ************************************************************************/
    
    #include<stdio.h>
    #include "shmfifo.h"
    
    typedef struct stu {
    	int id;
    	char name[32];
    }stu_t;
    
    int main( void )
    {
    	int i;
    	shmfifo_t *p = shmfifo_init(1234, sizeof(stu_t), 3);
    	//创建一个共享内存，每一块数据大小为sizeof(stu_t)，仓库容量为3
    	stu_t buf[8] = 
    	{
    			{1, "aaa"},
    			{2, "bbb"},
    			{3, "ccc"},
    			{4, "ddd"},
    			{5, "eee"},
    			{6, "fff"},
    			{7, "ggg"},
    			{8, "hhh"}
    	};
    	for (i=0; i<8; i++) {
    
    		shmfifo_put(p, buf+i);
    		printf("完成! %d\n",i);
    		sleep(1);
    	}
    }

这是我写的一个非常小的测试用例，myput和myget应该非常容易的通读，那么我们来看看到底get.c能不能拿到数据呢？？

![SouthEast 1][]

果然他符合预期的完成了我们的任务，所以呢我们的生产者消费者模型就书写完毕了，我们这个程序是有难度的，他就是相当于你使用

共享内存实 现了一个实实在在的消息队列，拥有跟消息队列同样的基本功能，但是可以自定义数据大小，数据个数，换言之就是自定义

你的数据存放缓冲区的 大小，并且你是共享内存通信你的效率一定大于消息队列. 最后添加了同步于互斥的操作，消除资源的冲突.

我们书写这个程序不仅要了解生产者 消费者模型，并且需要了解共享内存的基本函数操作，和信号量的基本函数操作. 所以呢，对于我

们的知识点也是一个很大的总结，非常的有意义

[SouthEast]: /images/20220601/807913dbef104ae9800644c454982f7c.png
[SouthEast 1]: /images/20220601/9c70e5631ed44718b5d6b5f6025f1ef3.png