多线程程序设计（五）——Producer-Consumer

news/2025/3/31 17:23:38/文章来源:https://www.cnblogs.com/green-cnblogs/p/18795973

本文摘要了《Java多线程设计模式》一书中提及的 Producer-Consumer 模式的适用场景，并针对书中例子（若干名称有微调）给出一份 C++ 参考实现及其 UML 逻辑图，也列出与之相关的模式。

◆ 适用场景

为了匹配数据的生产者（Producer）线程与消费者（Consumer）线程之间的处理速度。提高整体吞吐量（throughput）的同时，并保证数据的安全性。

◆ 解决方案

在 Producer 与 Consumer 线程之间加上一个可存放多条数据的缓冲。使用 std::mutex 锁机制保证对缓冲的临界区的互斥访问；使用 std::condition_variable 对象让线程等待至缓冲区可访问，并在可访问后唤醒在等待中的线程。

◆ 参考实现

例子模拟了厨师把做好的蛋糕给食客吃过程。厨师（Maker）把做好的蛋糕放在桌子（Table）上，食客（Eater）从桌上拿蛋糕吃。桌上最多放 3 块蛋糕，如果桌上还有 3 块蛋糕，厨师要等到桌上有空后才能接着放；如果桌上没有蛋糕了，食客要等到桌上有蛋糕后才能接着拿。

class Table
{...vector<string>__buffer__;                             #1intconst__capacity__;int__head__;int__tail__;int__count__;mutex__mtx__;condition_variable__cv__;...voidput(string cake, string name){unique_lock<mutex> lk(__mtx__);std::printf("%s puts %s\n", name.c_str(), cake.c_str());__cv__.wait(lk,                                               #3[&] {bool full = __count__ == __capacity__;          #2if (full)std::printf("Table is full. %s is waiting...\n", name.c_str());return !full;});__buffer__[__tail__] = cake;                        #4__tail__ = (__tail__ + 1) % __capacity__;++__count__;__cv__.notify_all();                                    #5}stringtake(string name){unique_lock<mutex> lk(__mtx__);__cv__.wait(lk,                                         #7[&] {bool idle = __count__ == 0;                 #6if (idle)std::printf("Table is empty. %s is waiting...\n", name.c_str());return !idle;});string cake(__buffer__[__head__]);                  #8__head__ = (__head__ + 1) % __capacity__;--__count__;__cv__.notify_all();                                    #9std::printf("%s takes %s\n", name.c_str(), cake.c_str());return cake;}};

存放蛋糕（string）的缓冲区（#1）由 Table 管理。做好的蛋糕通过 put 函数被放入缓冲区中。当蛋糕被放到桌上前，要判断缓冲区内是否已满（#2）。如果已放满（full 为 true），做蛋糕的厨师要等待（#3）；如果未放满，蛋糕被放入缓冲区后（#4），并会唤醒等待的其他厨师或食客（#5）。缓冲区中的蛋糕通过 take 函数被取出。当蛋糕被取出前，要判断缓冲区内是否还有蛋糕（#6）。如果没有蛋糕（idle 为 true），取蛋糕的食客要等待（#7）；如果有蛋糕，蛋糕从缓冲区中被取出（#8），并会唤醒等待的其他厨师或食客（#9）。

class Maker
{...string__name__;...voidrun(Table & table){while (true) {std::this_thread::sleep_for(milliseconds(std::rand()%1000));...cake += ...table.put(cake, __name__);                      #1}}};class Eater
{...string__name__;...voidrun(Table & table){while (true) {string cake = table.take(__name__);                 #2std::this_thread::sleep_for(milliseconds(std::rand()%3000));}}};

厨师会在 0 ~ 1 秒时间内，将做好的蛋糕放到桌上（#1）。食客会在 0 ~ 3 秒时间内，将做好的蛋糕从桌上拿走（#2）。

以下类图展现了代码主要逻辑结构,

class

以下顺序图展现了线程并发中的交互。

sequence

◆ 验证测试

笔者在实验环境一中编译代码（-std=c++11）成功后运行可执行文件，

$ g++ -std=c++11 -lpthread producer_consumer.cpp
$ ./a.out

运行结果如下：

...
Table is empty. Eater-1 is waiting...
Table is empty. Eater-2 is waiting...
Table is empty. Eater-3 is waiting...
Maker-3 puts [Cake No.1 by Maker-3]
Eater-1 takes [Cake No.1 by Maker-3]
Table is empty. Eater-2 is waiting...
Table is empty. Eater-3 is waiting...
Maker-2 puts [Cake No.2 by Maker-2]
Eater-2 takes [Cake No.2 by Maker-2]
Table is empty. Eater-3 is waiting...
Maker-3 puts [Cake No.3 by Maker-3]
Eater-3 takes [Cake No.3 by Maker-3]
Table is empty. Eater-2 is waiting...
Maker-1 puts [Cake No.4 by Maker-1]
Eater-2 takes [Cake No.4 by Maker-1]
Maker-1 puts [Cake No.5 by Maker-1]
Maker-3 puts [Cake No.6 by Maker-3]
Maker-2 puts [Cake No.7 by Maker-2]
Maker-3 puts [Cake No.8 by Maker-3]
Table is full. Maker-3 is waiting...
Maker-2 puts [Cake No.9 by Maker-2]
Table is full. Maker-2 is waiting...
Eater-3 takes [Cake No.5 by Maker-1]
Table is full. Maker-2 is waiting...
Maker-1 puts [Cake No.10 by Maker-1]
...

可以看到，当 Table 上没有蛋糕时，食客需要等待；而当 Table 上已放满蛋糕时，厨师需要等待。