生产者消费者模式

举一个寄信的例子。假设你要寄一封平信,大致过程如下:
1、把信写好,相当于生产者制造数据
2、你把信放入邮筒,相当于生产者把数据放入缓冲区
3、邮递员把信从邮筒取出,相当于消费者把数据取出缓冲区
4、邮递员把信拿去邮局做相应的处理,相当于消费者处理数据

1、优点

1.解耦

  假设生产者和消费者分别是两个类。如果让生产者直接调用消费者的某个方法,那么生产者对于消费者就会产生依赖(也就是耦合)。将来如果消费者的代码发生变化,可能会影响到生产者。而如果两者都依赖于某个缓冲区,两者之间不直接依赖,耦合也就相应降低了。
  接着上述的例子,如果不使用邮筒(也就是缓冲区),必须得把信直接交给邮递员。那么直接给邮递员不是挺简单的?其实不简单,你必须得认识谁是邮递员,才能把信给他。这就产生和你和邮递员之间的依赖(相当于生产者和消费者的【强】耦合)。万一邮递员换人了,还要重新认识一下吗(相当于消费者变化导致修改生产者代码)。而邮筒相对邮递员来说比较固定,依赖它的成本也就比较低(相当于和缓冲区之间的【弱】耦合)。
2.支持并发(concurrency)
  生产者直接调用消费者的某个方法,还有另一个弊端。由于函数调用是同步的(或叫阻塞的),在消费者的方法没有返回之前,生产者只好一直等在那边。万一消费者处理数据很慢,生产者就会产生等待。
  使用了生产者/消费者模式之后,生产者和消费者可以是两个独立的并发主体。生产者把制造出来的数据往缓冲区一丢,就可以再去生产下一个数据。基本上不用依赖消费者的处理速度。
  而其实最初这个模式,主要就是用来处理并发问题的。
  从寄信的例子来看。如果没有邮筒,就得拿着信站在路口等邮递员过来收(相当于生产者阻塞);又或者邮递员得挨家挨户问,谁要寄信(相当于消费者轮询)。不管是哪种方法,都挺土的。

3.支持忙闲不均

  缓冲区还有另一个好处。如果制造数据的速度时快时慢,缓冲区的好处就体现出来了。当数据制造快的时候,消费者来不及处理,未处理的数据可以暂时存在缓冲区中。等生产者的制造速度慢下来,消费者再慢慢处理掉。
  再拿寄信的例子来说事儿。假设邮递员一次只能带走1000封信。万一某次碰上情人节送贺卡,需要寄出去的信超过1000封,这时邮筒这个缓冲区就派上用场了。邮递员把来不及带走的信暂存在邮筒中,等下次过来时再拿走。

2、数据单元

  简单地说,每次生产者放到缓冲区的,就是一个数据单元;每次消费者从缓冲区取出的,也是一个数据单元。我们可以把每一封单独的信件看成是一个数据单元。
  不过仅仅这么介绍,太过于简单。所以,后面看一下数据单元需要具备哪些特性。

1.关联到业务对象

  首先,数据单元必须关联到某种业务对象。在考虑该问题的时候,你必须深刻理解当前这个生产者/消费者模式所对应的【业务逻辑】,才能够作出合适的判断。
  由于“寄信”这个业务逻辑比较简单,所以很容易就可以判断出数据单元是什么。但现实生活中,往往没这么乐观。大多数业务逻辑都比较复杂,当中包含的业务对象是层次繁多、类型各异。在这种情况下,就不易作出决策了。
  虽说这一步有难度,但是很重要!如果选错了业务对象,会导致后续程序设计和编码实现的复杂度大为上升,增加了开发和维护成本。

2.完整性

  所谓完整性,就是在传输过程中,要保证该数据单元的完整。要么【整个】数据单元被传递到消费者,要么完全没有传递到消费者。不允许出现【部分】传递的情形。
  对于寄信来说,你【不能】把半封信放入邮筒;同样的,邮递员从邮筒中拿信,也【不能】只拿出信的一部分。

3.独立性

  所谓独立性,就是各个数据单元之间没有互相依赖,某个数据单元传输失败【不应该】影响已经完成传输的单元;也【不能】影响尚未传输的单元。
  为什么会出现传输失败?假如生产者的生产速度在一段时间内一直超过消费者的处理速度,那就会导致缓冲区不断增长并达到上限,之后就很不妙了(有些数据单元会被无情地抛弃)。如果数据单元相互独立,等到生产者的速度降下来之后,后续的数据单元继续处理,不会受到牵连;反之,如果数据单元之间有某种耦合,导致被丢弃的数据单元会影响到后续其它单元的处理,那就会使程序逻辑变得非常复杂。
  对于寄信来说,某封信弄丢了,不会影响后续信件的送达;当然更不会影响已经送达的信件。

4.颗粒度

  前面提到,数据单元需要关联到某种业务对象。那么数据单元和业务对象是否要一一对应?很多场合确实是一一对应的。
  不过,有时出于性能等因素的考虑,也可能会把N个业务对象打包成一个数据单元。那么,这个N该如何取值就是颗粒度的考虑了。颗粒度的大小是有讲究的。太大的颗粒度可能会浪费空间;太小的颗粒度可能会影响时间性能。颗粒度的权衡要基于多方面的因素,以及一些经验值的考量。
  还是拿寄信的例子。如果颗粒度过小(比如设定为1),那邮递员每次只取出1封信。如果信件多了,那就得来回跑好多趟,浪费了时间。
  如果颗粒度太大(比如设定为100),那寄信的人得等到凑满100封信才拿去放入邮筒。假如平时很少写信,就得等上很久,也不太爽。
  生产者和消费者的颗粒度能否设置成不同大小(比如对于寄信人设置成1,对于邮递员设置成100)。当然,理论上可以这么干,但是在某些情况下会增加程序逻辑和代码实现的复杂度。

4、线程方式的缓冲区类型

1.内存分配的性能

  在线程方式下,生产者和消费者各自是一个线程。生产者把数据写入队列头(以下简称 push),消费者从队列尾部读出数据(以下简称 pop)。当队列为空,消费者就稍息(稍事休息);当队列满(达到最大长度),生产者就稍息。整个流程并不复杂。
  那么,上述过程会有什么问题?一个主要的问题是关于内存分配的性能开销。对于常见的队列实现:在每次 push 时,可能涉及到【堆内存】的分配;在每次 pop 时,可能涉及【堆内存】的释放。假如生产者和消费者都很勤快,频繁地 push、pop,那内存分配的开销就很可观!对于内存分配的开销,用 C/C++ 的同学,想必对 OS 底层机制会更清楚,应该知道分配【堆内存】(new 或 malloc)会有加锁的开销和用户态/核心态切换的开销。
  那该怎么办?就要靠”环形缓冲区类型”。

2.同步和互斥的性能

  另外,由于两个线程共用一个队列,自然就会涉及到线程间诸如同步、互斥、死锁等等劳心费神的事情。
  在很多场合中,诸如信号量、互斥量等玩意儿的使用也是有不小的开销的(某些情况下,也可能导致用户态/核心态切换)。如果像刚才所说,生产者和消费者都很勤快,那这些开销也不容小觑。
  这又该怎么办?就要靠“双缓冲区类型”。

3.适用于队列的场合

  刚才尽管批判了队列的缺点,队列方式也并非一无是处。由于队列是很常见的数据结构,大部分编程语言都内置了队列的支持,有些语言甚至提供了线程安全的队列(比如JDK 1.5引入的 ArrayBlockingQueue)。因此,开发人员可以捡现成,避免了重新发明轮子。
  所以,假如数据流量不是很大,采用队列缓冲区的好处还是很明显的:逻辑清晰、代码简单、维护方便。比较符合 KISS 原则。

5、进程方式

  说完了线程的方式,再来介绍基于进程的并发。
  跨进程的生产者/消费者模式,非常依赖于具体的进程间通讯(IPC)方式。而IPC的种类名目繁多,不便于挨个列举。因此只挑选几种跨平台、且编程语言支持较多的IPC方式来说。

1.匿名管道

  管道是最像队列的IPC类型。生产者进程在管道的写端放入数据;消费者进程在管道的读端取出数据。整个的效果和线程中使用队列非常类似,区别在于使用管道就无需操心线程安全、内存分配等琐事(操作系统暗中都帮你搞定了)。
  管道又分“命名管道”和“匿名管道”两种,今天主要聊匿名管道。因为命名管道在不同的操作系统下差异较大(比如 Win32 和 POSIX,在命名管道的 API 接口和功能实现上都有较大差异;有些平台不支持命名管道,比如 Windows CE)。除了操作系统的问题,对于有些编程语言(比如 Java)来说,命名管道是无法使用的。
  其实匿名管道在不同平台上的 API 接口,也是有差异的(比如 Win32 的 CreatePipe 和 POSIX 的 pipe,用法就很不一样)。但是我们可以仅使用标准输入和标准输出(以下简称 stdio)来进行数据的流入流出。然后利用 shell 的管道符把生产者进程和消费者进程关联起来。实际上,很多操作系统(尤其是 POSIX 风格的)自带的命令都充分利用了这个特性来实现数据的传输(比如 more、grep 等)。
  这么干有如下几个好处:

1. 基本上所有操作系统都支持在 shell 方式下使用管道符。因此很容易实现跨平台。
2. 大部分编程语言都能够操作 stdio,因此跨编程语言也就容易实现。
3. 刚才已经提到,管道方式省却了线程安全方面的琐事。有利于降低开发、调试成本。

当然,这种方式也有自身的缺点:

1. 生产者进程和消费者进程必须得在同一台主机上,无法跨机器通讯。这个缺点比较明显。
2. 在一对一的情况下,这种方式挺合用。但如果要扩展到一对多或者多对一,那就有点棘手了。所以这种方式的扩展性要打个折扣。假如今后要考虑类似的扩展,这个缺点就比较明显。
3. 由于管道是 shell 创建的,对于两边的进程不可见(程序看到的只是 stdio)。在某些情况下,导致程序不便于对管道进行操纵(比如调整管道缓冲区尺寸)。这个缺点不太明显。
4. 最后,这种方式只能单向传数据。好在大多数情况下,消费者进程不需要传数据给生产者进程。万一你确实需要信息反馈(从消费者到生产者),那就费劲了。可能得考虑换种 IPC 方式。

  顺便补充几个注意事项:

1. 对 stdio 进行读写操作是以阻塞方式进行。比如管道中没有数据,消费者进程的读操作就会一直停在哪儿,直到管道中重新有数据。
2. 由于 stdio 内部带有自己的缓冲区(这缓冲区和管道缓冲区是两码事),有时会导致一些不太爽的现象(比如生产者进程输出了数据,但消费者进程没有立即读到)。

2.SOCKET(TCP 方式)

  基于 TCP 方式的 SOCKET 通讯是又一个类似于队列的 IPC 方式。它同样保证了数据的顺序到达;同样有缓冲的机制。而且这玩意儿也是跨平台和跨语言的,和刚才介绍的 shell 管道符方式类似。
  SOCKET 相比 shell 管道符的方式,有如下优点:

1. SOCKET 方式可以跨机器(便于实现分布式)。这是主要优点。
2. SOCKET 方式便于将来扩展成为多对一或者一对多。这也是主要优点。
3. SOCKET 可以设置阻塞和非阻塞方法,用起来比较灵活。这是次要优点。
4. SOCKET 支持双向通讯,有利于消费者反馈信息。

  当然有利就有弊。相对于上述 shell 管道的方式,使用 SOCKET 在编程上会更复杂一些。好在前人已经做了大量的工作,搞出很多 SOCKET 通讯库和框架(比如 C++ 的 ACE 库、Python 的 Twisted)。借助于这些第三方的库和框架,SOCKET 方式用起来还是比较爽的。
  虽然 TCP 在很多方面比 UDP 可靠,但鉴于跨机器通讯先天的不可预料性(比如网线可能被拔错了,网络的忙闲波动可能很大),在程序设计上我们还是要多留一手。具体该如何做?可以在生产者进程和消费者进程内部各自再引入基于线程的“生产者/消费者模式”。

1718644443163.jpg

  这么做的关键点在于把代码分为两部分:生产线程和消费线程属于和业务逻辑相关的代码(但和通讯逻辑无关);发送线程和接收线程属于通讯相关的代码(但和业务逻辑无关)。
  这样的好处是很明显的,具体如下:

1. 能够应对暂时性的网络故障。并且在网络故障解除后,能够继续工作。
2. 网络故障的应对处理方式(比如断开后的尝试重连),只影响发送和接收线程,不会影响生产线程和消费线程(业务逻辑部分)。
3. 具体的 SOCKET 方式(阻塞和非阻塞)只影响发送和接收线程,不影响生产线程和消费线程(业务逻辑部分)。
4. 不依赖 TCP 自身的发送缓冲区和接收缓冲区。(默认的 TCP 缓冲区的大小可能无法满足实际要求)
5. 业务逻辑的变化(比如业务需求变更)不影响发送线程和接收线程。

  针对上述的最后一条,再多啰嗦几句。如果整个业务系统中有多个进程是采用上述的模式,那或许可以重构一把:在业务逻辑代码和通讯逻辑代码之间切一刀,把业务逻辑无关的部分封装成一个通讯中间件(说“中间件”显得比较高大上)。

6、环形缓冲区 vs 队列缓冲区

1.外部接口相似

  介绍环形缓冲区之前,先来回顾一下普通的队列。普通的队列有一个写入端和一个读出端。队列为空的时候,读出端无法读取数据;当队列满(达到最大尺寸)时,写入端无法写入数据。
  对于使用者来讲,环形缓冲区和队列缓冲区是一样的。它也有一个写入端(用于 push)和一个读出端(用于 pop),也有缓冲区“满”和“空”的状态。所以,从队列缓冲区切换到环形缓冲区,对于使用者来说能比较平滑地过渡。

2.内部结构迥异

  虽然两者的对外接口差不多,但是内部结构和运作机制有很大差别。队列的内部结构此处就不多啰嗦了。重点介绍一下环形缓冲区的内部结构。
  可以把环形缓冲区的读出端(以下简称 R)和写入端(以下简称 W)想象成是两个人在体育场跑道上追逐。当 R 追上 W 的时候,就是缓冲区为空;当 W 追上 R 的时候(W 比 R 多跑一圈),就是缓冲区满。

1718644737748.jpg

  从上图可以看出,环形缓冲区所有的 push/pop 操作都是在一个【固定】的存储空间内进行。而队列缓冲区在 push 的时候,可能会分配存储空间用于存储新元素;在 pop 时,可能会释放废弃元素的存储空间。所以环形方式相比队列方式,少掉了对于缓冲区元素所用存储空间的分配、释放。这是环形缓冲区的一个主要优势。

7、环形缓冲区的实现

1.数组方式 vs 链表方式
  环形缓冲区的内部实现,即可基于数组(此处的数组,泛指连续存储空间)实现,也可基于链表实现。
  数组在物理存储上是一维的连续线性结构,可以在初始化时,把存储空间【一次性】分配好,这是数组方式的优点。但是要使用数组来模拟环,必须在逻辑上把数组的头和尾相连。在顺序遍历数组时,对尾部元素(最后一个元素)要作一下特殊处理。访问尾部元素的下一个元素时,要重新回到头部元素(第0个元素)。如下图所示:

1718644846253.jpg

  使用链表的方式,正好和数组相反:链表省去了头尾相连的特殊处理。但是链表在初始化的时候比较繁琐,而且在有些场合(比如后面提到的跨进程的 IPC)不太方便使用。

2.读写操作

  环形缓冲区要维护两个索引,分别对应写入端(W)和读取端(R)。写入(push)的时候,先确保环没满,然后把数据复制到 W 所对应的元素,最后 W 指向下一个元素;读取(pop)的时候,先确保环没空,然后返回 R 对应的元素,最后 R 指向下一个元素。

3.判断“空”和“满”

  上述的操作并不复杂,不过有一个小小的麻烦:空环和满环的时候,R 和 W 都指向同一个位置!这样就无法判断到底是“空”还是“满”。大体上有两种方法可以解决该问题。
  办法1:始终保持一个元素不用
  当空环的时候,R 和 W 重叠。当 W 比 R 跑得快,追到距离 R 还有一个元素间隔的时候,就认为环已经满。当环内元素占用的存储空间较大的时候,这种办法显得很土(浪费空间)。
  办法2:维护额外变量
  如果不喜欢上述办法,还可以采用额外的变量来解决。比如可以用一个整数记录当前环中已经保存的元素个数(该整数>=0)。当 R 和 W 重叠的时候,通过该变量就可以知道是“空”还是“满”。

4.元素的存储

  由于环形缓冲区本身就是要降低存储空间分配的开销,因此缓冲区中元素的类型要选好。尽量存储【值类型】的数据,而不要存储【指针(引用)类型】的数据。因为指针类型的数据又会引起存储空间(比如堆内存)的分配和释放,使得环形缓冲区的效果大打折扣。

7、应用场合

  如果你所使用的编程语言和开发库中带有现成的、成熟的环形缓冲区,强烈建议使用现成的库,不要重新制造轮子;确实找不到现成的,才考虑自己实现。

1.用于并发线程

  和线程中的队列缓冲区类似,线程中的环形缓冲区也要考虑线程安全的问题。除非你使用的环形缓冲区的库已经帮你实现了线程安全,否则你还是得自己动手搞定。线程方式下的环形缓冲区用得比较多,相关的网上资料也多,下面就大致介绍几个。
  对于 C++ 的程序员,强烈推荐使用 boost 提供的 circular_buffer 模板,该模板最开始是在 boost 1.35版本中引入的。鉴于 boost 在 C++ 社区中的地位,应该可以放心使用该模板。

2.用于并发进程

  适合进行环形缓冲的 IPC 类型,常见的有“共享内存和文件”。在这两种方式上进行环形缓冲,通常都采用数组的方式实现。程序事先分配好一个固定长度的存储空间,然后具体的读写操作、判断“空”和“满”、元素存储等细节就可参照前面所说的来进行。
  共享内存方式的性能很好,适用于数据流量很大的场景。但是有些语言(比如 Java)对于共享内存不支持。因此,该方式在多语言协同开发的系统中,会有一定的局限性。
  而文件方式在编程语言方面支持很好,几乎所有编程语言都支持操作文件。但它可能会受限于磁盘读写(Disk I/O)的性能。所以文件方式不太适合于快速数据传输;但是对于某些“数据单元”很大的场合,文件方式是值得考虑的。
对于进程间的环形缓冲区,同样要考虑好进程间的同步、互斥等问题。
“双缓冲区”是一个应用很广的手法。该手法用得最多的地方想必是屏幕绘制相关的领域(主要是为了减少屏幕闪烁)。另外,在设备驱动和工控方面,双缓冲也经常被使用。不过今天要聊的,并不是针对上述的某个具体领域,而是侧重于并发方面的同步/互斥开销。另外提醒一下,双缓冲方式和前面提到的队列缓冲、环形缓冲是可以结合使用的。

8、为什么要双缓冲区?

  在介绍队列缓冲区时,提及了普通队列缓冲区的两个性能问题:“内存分配的开销”和“同步/互斥的开销”。“内存分配的开销”已经在介绍环形缓冲区的时候解决了,而今天要介绍的双缓冲区,就是冲着同步/互斥的开销来的。
  只有当同步或互斥的开销非常明显的时候,才应该考虑双缓冲区的使用。否则还是用最基本、最简单的队列缓冲区。
9、双缓冲区的原理
=========。
  所谓“双缓冲区”,就是要有俩缓冲区(简称 A 和 B)。这俩缓冲区,总是一个用于生产者,另一个用于消费者。当俩缓冲区都操作完,再进行一次切换(先前被生产者写入的转为消费者读出,先前消费者读取的转为生产者写入)。由于生产者和消费者不会同时操作同一个缓冲区(不发生冲突),所以就不需要在读写每一个数据单元的时候都进行同步/互斥操作。(顺便提一下,这又一次展现了【空间换时间】的优化思路)
  但是光有俩缓冲区还不够。为了真正做到“不冲突”,还得再搞两个互斥锁(简称 La 和 Lb),分别对应俩缓冲区。生产者或消费者如果要操作某个缓冲区,必须先拥有对应的互斥锁。补充一句:要达到“不冲突”的效果,其实可以有多种搞法,今天只是挑一个简单的来讨论。

10、双缓冲区的几种状态

1.俩缓冲区都在使用的状态(并发读写)

  大多数情况下,生产者和消费者都处于并发读写状态。不妨设生产者写入 A,消费者读取 B。在这种状态下,生产者拥有锁 La;同样的,消费者拥有锁 Lb。由于俩缓冲区都是处于独占状态,因此每次读写缓冲区中的元素(数据单元)都【不需要】再进行加锁、解锁操作。这是节约开销的主要来源。

2.单个缓冲区空闲的状态

  由于两个并发实体的速度会有差异,必然会出现一个缓冲区已经操作完,而另一个尚未操作完。不妨假设生产者快于消费者。
  在这种情况下,当生产者把 A 写满的时候,生产者要先释放 La(表示它已经不再操作 A),然后尝试获取 Lb。由于 B 还没有被读空,Lb 还被消费者持有,所以生产者进入发呆(Suspend)状态。

3.缓冲区的切换

  过了若干时间,消费者终于把 B 读完。这时候,消费者也要先释放 Lb,然后尝试获取 La。由于 La 刚才已经被生产者释放,所以消费者能立即拥有 La 并开始读取 A 的数据。而由于 Lb 被消费者释放,所以刚才发呆的生产者会缓过神来(Resume)并拥有 Lb,然后生产者继续往 B 写入数据。
  经过上述几个步骤,俩缓冲区完成了对调,变为:生产者写入 B,消费者读取 A。

11、可能的并发问题

  本来单个缓冲区的生产者/消费者问题就已经是教科书的经典问题了,现在搞出俩缓冲区,所以就更加耗费脑细胞了。一不小心,就会搞出些并发的Bug,而且并发的Bug还很难调试和测试(这也就是不要轻易使用的原因)。

1.死锁的问题

  假如把前面介绍的操作步骤调换一下顺序:生产者或消费者在操作完当前的缓冲区之后,先去获取另一个缓冲区的锁,再来释放当前缓冲区的锁。那会怎样?
  一旦两个并发实体【同时】处理完各自缓冲区,然后【同时】去获取对方拥有的锁,那就会出现典型的死锁场景。它俩从此陷入万劫不复的境地。

12、应用场景

1.用于并发线程

  在线程方式下,首先要考虑的是缓冲区的类型:到底用队列方式还是环形方式。这方面的选择依据在介绍环形缓冲区的时候已经阐述过了。
  另一个需要注意的是,某些编程语言或者程序库提供了的线程安全的缓冲区(比如 JDK 1.5 引入的 ArrayBlockingQueue)。由于这种缓冲区会自动为每次的读写进行同步/互斥,所以就把双缓冲的优势抵消掉了。因此,在进行缓冲区选型的时候要避开这类缓冲区。

2.用于并发进程

  在进程间使用双缓冲,先得考察不同 IPC 类型的特点。由于今天讨论双缓冲的目的是降低同步/互斥的开销,对于那些已经封装了同步/互斥的 IPC 类型,就没太大必要再去搞双缓冲了(单凭这条就足以让好多种 IPC 出局)。剩下的 IPC 类型中,比较适合用双缓冲的主要是:共享内存和文件。


扫描二维码,在手机上阅读!

发表评论

电子邮件地址不会被公开。 必填项已用*标注