前言

介绍了进程如何基于面向对象的封装，本章我们基于封装好的Process类来实现一种无锁版的生产者和消费者框架，用它实现了高性能文件拷贝功能。读这篇文章之前大家可以想一下如果是你，你会怎么设计这样的框架？在这个模型中作为每个生产者，如何把读取到内容发送给消费者。而作为消费者，如何把接收到的内容写入文件，并且最终保证文件内容是一致的。

好了，废话不多说，我们来通过代码看下是怎么设计的吧？

生产者消费者模型

我们回顾一下什么是生产者消费者模型。这个模型是为了解决在整个程序过程中既要不断产生数据，又要处理数据的一种方案:生产者专注产生数据，消费者专注处理数据。
但如果模型中生产者产生数据很快，消费者处理数据很慢，那么必然会出现生产者要等待消费者处理完以后，才能继续生产。或者反之也是类似，消费者则会等到生产者生产完才去处理。这样两者就形成了依赖关系，无法做到高性能，需要在两者之间加入缓冲的区域，让它们不再有直接相互依赖的情况，如下图所示：

生产者消费者进程

结合上述对于多进程的封装以及生产者消费者的模型，该框架的设计关键点如下：

1.采用多个生产者和消费者都为进程。将Process作为基类，设计Process的继承类生产者进程类（ProducerProcess）和消费者进程类（ConsumerProcess）

2.两者之间的数据缓冲的通道则基于对无名管道的封装设计为一个缓存通道类，让生产者进程和消费者进程共用，并且生产者只往通道里放入数据，消费者只从通道里获取数据。

3.考虑到具体过程是文件处理，需要设计一种数据结构，用于生产者和消费者之间的数据传递。

发送数据结构设计

1.需要发送到管道的数据用Producer结构体来存放，并且通过内存拷贝来传输和接受数据。因为多进程之间指针是很难共享的，这里采用相对简单的方式是通过进程间通信的办法，把指针指向的内容传递过去。

2.结构体内要有一个读/写数据长度的成员变量。因为无法预设要拷贝的文件会有多大，所以不能认为只通过一两次读取文件就能拿到它的所有内容了。那么就得把读取文件的过程设计为每次不断读取固定长度的数据，直到读到文件末尾为止。

具体代码如下：


typedef  struct Producer{size_t length;size_t readoffset;char *readbuf;}Producer;typedef struct Consumer{size_t length;size_t writeoffset;char *writebuf;}Consumer;

缓冲通道设计

1.如果单次传输数据量大于了管道的大小，就会出问题，因此对于上诉数据传递结构体中的数据长度要小于管道最大存储值（默认是4k）。

2.需要封装管道的读端写端的打开/关闭，判断是否已打开等。

具体代码如下：

class Tunnel{public:Tunnel();~Tunnel();bool isOpen() const;void open();void closeWriteDescriptor();void closeReadDescriptor();ssize_t read(char *buf, size_t count);ssize_t write(char *buf, size_t count);inline int getReadDescriptor(){return m_pipeFd[0];}inline int getWriteDescriptor(){return m_pipeFd[1];}private:int m_pipeFd[2];};

读文件设计

这里有个问题，既然每个生产者进程的业务逻辑都是先读了源文件的一部分后再向管道发送读到的数据。那么到下一次读源文件的时候，要偏移多少呢？在这里，多个生产者进程读取文件的设计上为了减少同步机制造成的操作系统开销，采取每个进程从不同初始偏移量开始读取的方式。初始偏移量=index✖️m_maxlength。index为进程序列号，比如共有5个进程，index为0、1、2、3、4 ，m_maxlength为最大读取长度。到下一次再去读取源文件的时候，偏移量offset又从原有的offset增加了m_maxProducer✖️m_maxlength。其中的m_maxProducer是类ConfigManager的成员变量，从输入参数中获取。

此外，还采用pread接口读取文件，其实际操作是lseek+read,即先将offset调整到指定值，再调用read读取数据。它同read的不同之处在于它是原子操作，即pread操作前后的offset是一致的。

读文件大小超过管道大小遇到的问题

若管道大小不是结构体Producer大小的整数倍，必然会出现生产者进程写入管道的数据大于管道大小。那么对于消费者进程来说，读取的数据大小是固定长度的，那么必然会出现它没有读完，只读取到管道大小数据就直接返回。按照Linux公平调度原则，这个消费者没办法马上读取剩余数据，只能通过read的返回值知道还有多少没有读。当管道再次未满时，未发送完数据的生产者进程从阻塞态转为运行态，把剩余数据继续发送到管道并返回。而其它生产者进程会继续把自己的Producer数据继续写管道。这样又会造成一个问题。消费者按照固定长度读取，必然会把前一个生产者写入管道的剩余数据和后面生产者写入管道的数据一同读取。从而造成了内容错乱问题。怎么办呢？设计中我们把Producer的大小定义为管道大小的1/2。

源码链接

源码链接

多进程生产者消费者框架设计相关推荐

Python并发编程之多进程(生产者消费者模型)
十二.生产者消费者模型 1.什么是生产者消费者模型生产者 : 程序中负责产生数据的一方消费者 : 程序中负责处理数据的一方 2.为什么引入生产者消费者模型在并发编程中, 生产者消费者模式通过一个 ...
图文并茂的生产者消费者应用实例demo
前面的几篇文章<<.NET 中的阻塞队列BlockingCollection的正确打开方式>><<项目开发中应用如何并发处理的一二事>>从代码以及理论角 ...
多进程实现生产者消费者
1 # 多进程实现生产者消费者模型 2 import multiprocessing 3 import random 4 import time 5 6 7 class Producer(multip ...
Linux多进程实现生产者消费者问题
1. 任务简介生产者消费者问题(Producer-consumer problem),也称有限缓冲问题(Bounded-buffer problem),是一个著名的进程同步问题的经典案例.它描述的是 ...
基于队列的生产消费设计java_生产者-消费者设计模式
一.生产者消费者设计模式 1.中间队列一段内存空间,且可存取: 2.两种角色 (1)生产者:生产数据: (2)消费者:消费数据. 3.三种关系 (1)生产者与生产者的互斥关系: (2)消费者与消费者 ...
linux 生产者消费者多进程,Linux多线程，生产者消费者算法和条件变量的使用
接着上一篇博文,原来双线程,现在为了实现暂停/继续功能,又加了一个线程.第三线程使用条件信号量,当用户按下S键,第三线程将检测到,并且将ifpause置为1,然后输出线程将在if语句成立后被条件信 ...
并发编程（五）python实现生产者消费者模式多线程爬虫
并发编程专栏系列博客并发编程(一)python并发编程简介并发编程(二)怎样选择多线程多进程和多协程并发编程(三)Python编程慢的罪魁祸首.全局解释器锁GIL 并发编程(四)如何使用多线程, ...
java多线程抽奖_java 线程池、多线程并发实战(生产者消费者模型 1 vs 10) 附案例源码...
导读前二天写了一篇<Java 多线程并发编程>点我直达,放国庆,在家闲着没事,继续写剩下的东西,开干! 线程池为什么要使用线程池例如web服务器.数据库服务器.文件服务器或邮件服务器 ...
生产者/消费者问题的多种Java实现方式--转
实质上,很多后台服务程序并发控制的基本原理都可以归纳为生产者/消费者模式,而这是恰恰是在本科操作系统课堂上老师反复讲解,而我们却视而不见不以为然的.在博文<一种面向作业流(工作流)的轻量级可复用 ...

多进程生产者消费者框架设计

前言