首先，生产者-消费者模型
生产者-消费者模式是解决多个模块间数据通信问题的有效机制。通过在数据生产者和数据消费者之间设置数据缓冲区，实现了低耦合度的数据通信。
图1生产者/消费者模式的结构
这样的结构就像流水线上的两道工序，以及它们之间的一个架子。前道工序有一批工人，他们会把这道工序的产品上架，然后马上回到自己的生产工作中；同样，后一道工序的若干工人可以直接从货架上领取前一道工序的产品，直接开始自己的生产工作。
与直接调用数据通信模式相比，生产者/消费者模式有一个额外的数据缓冲区，但它的优势也非常明显:
1.支持模块并发
使用直接调用通信最明显的缺点是调用关系被阻塞，调用者必须中断自己的任务，等待被调用者返回后才能继续执行，大大降低了程序的效率。尤其是当被调用的模块涉及到网络通信、文件读写等耗时的操作时，如果主音一直阻塞等待，将是一种无法接受的性能损失。使用生产者/消费者模式，生产者模块(即原音函数)只需要将数据放入缓冲区，本周期的任务完成，就可以立即返回执行下一周期的任务。类似地，消费者模块(对应于最初被调用的函数)并不 不必等到最后一个进程结束，只要数据缓冲区不为空，就可以立即开始消费操作。
2.支持忙闲不均。
在很多业务场景中，数据生产者的生产速度和数据消费者的消费速度可能在一定范围内波动。如果使用直接呼叫，无论主叫方或被叫方是否忙，双方都必须减速以协调数据同步的速度。但是，如果一方通过数据缓冲区进入繁忙状态，它可以使用缓冲区中的数据或空间进行调整(如果消费者的处理速度变慢，生产者不必减速，生成的数据可以使用缓冲区的剩余部分进行存储)。
也可以对应流水线的例子。如果前道工序的工人上厕所，如果直接交接产品，那么后道工序的工人就要暂时停工。而如果双方通过货架交接，货架上的产品就足够前道工序的工人在前道工序的工人离开的这段时间操作。
3.降低耦合度
如果使用直拨，如果以后双方的编码发生变化(比如收发数据的频率，数据单元的大小)，双方的编码都需要变化。在生产者/消费者模式中，如果一方 的数据访问方法发生变化时，他只需要使用缓冲区的一部分。
对应于流量以水线为例，如果下一道工序上来一个新人，这个新人有自己的工作节奏，习惯一次拿两块，那么上一道工序的人就得改变节奏，给隔壁送两块；如果双方通过货架交接，无论隔壁的人有什么工作习惯，都可以按照自己的想法从货架上拿走任何零件，前道工序的生产节奏不需要调整。
二、环形缓冲区结构
图2环形缓冲区的结构
循环缓冲区使用两个指针分别用于读写，两者旋转方向相同，就像两个人在操场上互相竞争。在读指针前面，写指针后面，这部分被划分成数据块，这部分的状态是可读和未写；而写指针在读指针的前面和后面，这部分被划分成空闲块，这部分块的状态是可写和不可读。当写指针追上读指针时，缓冲区已满，写指针需要暂停写入；当读指针追上写指针时，缓冲区为空，读指针需要暂停读取。
与普通的队列结构(fifo)相比，环形缓冲区的所有读写操作都是在相对固定的存储区域内完成的，这样如果程序涉及到频繁的读写，就可以节省大量的空间应用释放操作。
三、生产者-消费者模式下环形缓冲区的应用(代码实现)
以下代码是在linux环境下编译运行的，其他环境可能略有不同。
# include stdio . h # include stdlib . h # include unistd . h # include semaphore . h # include pthread . h # define size 10int product _ idx = 0；//生产者指针int consume _ idx = 0；//消费者指针int data = 0；int ring[size]；//环形缓冲区sem _ t blanksem//控制空闲块的信号量sem _ t datasem//控制数据块的信号量pthread _ mutex _ t product _ lock = pthread _ mutex _ initializer；//互斥锁pthread _ mutex _ t consume _ lock = pthread _ mutex _ initializer；//互斥锁定消费者中的pthread _ tproduct _ 1 _ tid、product _ 2 _ tid、consume _ 1 _ tid、consume _ 2 _ tidvoid* product_1(void *arg) //生产者线程1{ while(1) {pthread_mutex_lock(product _ lock)；//抢占生产者互斥锁sem _ wait(blank sem)；//获取一个空闲的块资源环[product _ idx]= data；sem _ post(datasem)；//释放一个数据块资源printf( product _ 1 putdata : % d \ n ，环[product _ idx])；product _ idx = product _ idx % size//环形缓冲区pthread _ mutex _ unlock(product _ lock)的地址方法；//释放生产者互斥锁sleep(1)；}返回null}void* product_2(void *arg) //生产者线程2 { while(1){ pthread _ mutex _ lock(product _ lock)；sem _ wait(blank sem)；ring[product _ idx]= data；s: % d \ n ，环[product _ idx])；product _ idx = product _ idx % sizepthread _ mutex _ unlock(product _ lock)；睡眠(1)；}返回null}void* consume_1(void *arg) //消费者线程1 { int consume _ data = 0；while(1){ pthread _ mutex _ lock(consume _ lock)；//抢占消费者互斥体sem _ wait(datasem)；//获取一个数据块资源consume _ data = ring[consume _ idx]；sem _ post(blank sem)；//释放一个自由块资源printf( consume _ 1 get data : % d \ n ，consume _ data)；睡眠(1)；consume _ idxconsume _ idx = consume _ idx % size//环形缓冲区的地址方法pthread _ mutex _ unlock(consume _ lock)；//释放消费者互斥睡眠(2)；}返回null}void* consume_2(void *arg) //消费者线程2 { int consume _ data = 0；while(1){ pthread _ mutex _ lock(consume _ lock)；sem _ wait(datasem)；consume _ data = ring[consum: % d \ n ，consume _ data)；睡眠(1)；consume _ idxconsume _ idx = consume _ idx % sizepthread _ mutex _ unlock(consume _ lock)；睡眠(2)；}返回null}int main{ sem_init(blanksem，0，size)；//初始化信号量sem_init(datasem，0，0)；int ret = 0；ret = pthread _ create(product _ 1 _ tid，null，(void *) product_1，null)；//创建线程if(ret){ printf( pthread _ createproduct _ 1错误);退出(0)；} ret = pthread _ create(product _ 2 _ tid，null，(void *) product_2，null)；if(ret){ printf( pthread_create product_2错误);退出(0)；} ret = pthread _ create(consume _ 1 _ tid，null，(void *) consume_1，null)；if(ret){ printf( pthread_create consume_1错误);退出(0)；} ret = pthread _ create(consume _ 2 _ tid，null，(void *) consume_2，null)；if(ret){ printf( pthread_create consume_2错误);退出(0)；} pthread_join(product_1_tid，null)；//让主线程等待每个子线程完成pthread _ join (product _ 2 _ tid，null)；pthread_join(consume_1_tid，null)；pthread_join(consume_2_tid，null)；sem _ destroy(blank sem)；//销毁信号量sem _ destroy(datasem)；}
代码编译运行后，运行效果如下图所示:
图3演示程序的运行效果
可以看出，生产者产生的数总是大于消费者取走的数，而且不会比消费者取走的数大10，这说明生产者 s指针既不能被消费者超越，也不能被消费者围剿。
四、定速生产者模式下的设计思路
生产者/消费者模型中一个很重要的原则是，需要时刻监控缓冲区的状态，当缓冲区满了，生产者就应该停止生产操作；当缓冲区为空时，应该停止使用方。
但在实际开发中，可能会出现生产者或消费者处于固定速度的状态(比如生产者是一个数据采集模块，他采集的很多数据都需要消费)。在这种情况下，流水线的例子似乎并不适用。我们可以用一个新的混凝土模型:天花板漏水。
天花板以固定的速度向地板漏水，地上放一个小水桶。偶尔有住户从卧室出来倒水。在这里，漏水的天花板充当生产者，住户是消费者，这个小水桶就是他们之间的缓冲器。在这种情况下，生产者以固定的速度连续产生数据，即使缓冲区满了，他也不会停滴。这种情况下如何保护缓冲区不溢出？我的想法是在缓冲器上再加一个缓冲器，就是在水桶下面再放一个盆。在实际开发中，采用的是读写文件的，在缓冲区满了的情况下写入文件。在这种情况下，消费者策略也应该更改为在缓冲区不为空时读取缓冲区，在缓冲区为空时读取文件。
在代码层面，可以考虑用sem_trywait(sem_t *sem _ t * sem)代替sem_wait(sem_t*sem)。当sem_wait想要获取的资源量为0时，线程会一直阻塞在这里，等待其他线程释放资源，而sem _ wait不会等待。当资源量为0时，继续向下执行，通过返回值判断是否成功获取资源。如果成功获取，则返回0，信号量减1。无法获取-1，信号量未更改。
通过这种，我们可以对上述文章中的生产者和消费者代码进行以下更改:void* product(void *arg) //生产者线程{ while(1){ pthread _ mutex _ lock(product _ lock)；//抓住生产者互斥锁if (sem _ try (blanksem)！= 0) //非阻塞尝试获取空闲块资源{/*此处添加写文件代码*/} else //获取成功{ ring[product _ idx]= data；sem _ post(datasem)；//释放一个数据块资源printf( product _ 1 putdata : % d \ n ，环[product _ idx])；product _ idx = product _ idx % size} pthread _ mutex _ unlock(product _ lock)；//释放生产者互斥锁sleep(1)；}返回null}void* consume(void *arg) //消费者线程{ int consume _ data = 0；while(1){ pthread _ mutex _ lock(consume _ lock)；//抢占消费者互斥if (sem _ trywaite (datasem)！= 0) //非阻塞尝试获取一个数据块资源{/*这里是补充读取文件代码*/} else //获取成功{ consume _ data = ring[consume _ idx]；sem _ post(blank sem)；//释放一个自由块资源printf( consume _ 1 get data : % d \ n ，consume _ data)；睡眠(1)；consume _ idxconsume _ idx = consume _ idx % size} pthread _ mutex _ unlock(consume _ lock)；//释放消费者互斥睡眠(2)；}返回null}
标签:
缓冲生产者
了解更多环形缓冲区中有缓冲区类型(环形缓冲区是一种类型)相关内容请关注本站点。