• 基于阻塞队列的生产消费模型

    目录

    一、线程同步

    1.生产消费模型（或生产者消费者模型）

    2.认识同步

    （1）生产消费模型中的同步

    （2）生产者消费者模型的特点

    二、条件变量

    1.认识条件变量

    2.条件变量的使用

    3.代码改造

    三、基于阻塞队列的生产消费模型

    1.阻塞队列类

    （1）阻塞队列

    （2）实现生产者的生产函数

    （3）实现消费者的消费函数

    2.pthread_cond_wait为什么要传入锁

    3.生产者和消费者线程的执行函数

    （1）执行函数

    （2）试运行

    4.部分细节处理

    （1）伪唤醒问题

    （2）解锁与唤醒的顺序

    5.处理任务的生产消费模型

    （1）代码改造

    （2）运行

    6.生产消费模型为何高效

    四、双阻塞队列的生产消费模型

    1.编写类代码

    2.增加处理保存任务的函数

    3.更改线程函数

    4.更改main函数

    一、线程同步

    1.生产消费模型（或生产者消费者模型）

    我们肯定有在超市买东西的经历，比如买水。超市的瓶装水是供应商提供的，所以供应商是生产者；超市从供应商进货，超市就是一个交易场所；我们从超市买水，我们就是消费者。

    这些概念也可以转换到线程中，我们将读取数据的线程叫做消费者线程（消费者），将产生数据的线程叫做生产者线程（供应商），将共享的特定数据结构叫做缓冲区（交易场所）。

    超市中售卖的瓶装水品牌很多，所以供应商肯定不止一个，这些不同牌子瓶装水的生产者之间的关系就是竞争关系。所以，多线程中各个生产者线程之间是互斥关系，同一时刻只有一个生产者线程能访问缓冲区。

    对于消费者线程也是一样的，一个消费者买到了水，其他人可能就买不到了。所以消费者线程和消费者线程之间也是互斥关系，同一时刻也只有一个消费者线程能访问缓冲区。

    超市也总要有补货的时间，如果当前仓库里有货，但是工作人员还没有将货物摆在货架上，那到来的顾客就以为这里没有水。

    为了避免出问题，应当在超市补货时阻止消费者进入。由于缓冲区数据又被错误覆盖的风险，所以最好在消费者线程访问缓冲区时，不允许生产者线程访问缓冲区，反之亦然。

    消费者线程和生产者线程之间也是互斥关系，在同一时间内只有一个线程可以访问缓冲区。

    2.认识同步

    （1）生产消费模型中的同步

    在保证数据安全的前提下，让线程能够按照某种特定的顺序访问临界资源，从而有效避免饥饿问题，叫做同步。

    首先什么叫饥饿状态？

    我以多线程抢票代码为例，如果每个线程抢完票后都没有进行其他处理动作时，第一个申请到锁的线程更容易申请到锁。最终，大部分票都被一个线程抢走。而其他线程竞争能力弱，缺乏调度，这些线程就处于饥饿状态。

    那什么是同步？

    而同步就是让所有线程按照一定顺序来抢票，尽可能做到人人有份，避免线程饥饿问题产生。

    再次回到超市买水，供应商不能没完没了地向超市供货，一方面超市一直关着，消费者无法消费，另一方面，超市又不是四次元口袋，总是要装满的。

    同样，消费者也不能没完没了地买水，一方面超市不关门，供货商不能进货，另一方面，瓶装水又肯定会卖完。

    所以最好生产者先供货，货架摆满了就不进货了。消费者来买，当水卖完了再让供应商进货，让消费者和生产者协同起来。

    所以，消费者线程和生产者线程之间也是同步关系。生产者线程和消费者线程需要按照一定顺序去访问缓冲区。

    所以，我们可以将生产消费模型总结为321原则：3种关系、2种角色、1个交易场所。

    • 3种关系：生产者和生产者(互斥关系)，消费者和消费者(互斥关系)，生产者和消费者(互斥和同步关系)
    • 2种角色：生产者和消费者
    • 1个交易场所：一段特定结构的缓冲区

    生产消费模型的运行本质就是321原则。

    （2）生产者消费者模型的特点

    对供货商而言，只需要给超市供大量的货即可，不用关心消费者什么时候来买。

    对消费者而言，只需要直接去超市买方便面就行，不用等待方便面的生产运输。

    对超市而言，只需要在水卖完时，告诉供货商进货，进完货后告诉消费者来买。

    生产消费模型中消费者和生产者各自只需要关心自己所做的事情，生产与消费线程线程之间完全独立，在计算机科学的角度，我们称其实现了消费者线程和生产者线程之间的解耦。

    我们大部分人在周一到周五都是上班上学，所以这些时间去超市买水的人会相对少，这个时候超市就可以适时多进货。而周末大家都放假了，去超市买水的人变多，因为之前进货也很多了，就不需要进货了。

    就像上面所说的策略，生产消费模型解决了生产者线程和消费者线程忙闲不均的问题。

    而超市作为交易场所，能够储存更多的物品，同样缓冲区也能储存更多的数据。如果消费者直接去找供货商，供货商一般都不会零售。纵使能够零售，直接去找生产者还要等待生成者完成商品生产，消耗时间成本高，效率低。

    生产者消费者模型提高了了生产者线程和消费者线程的执行效率。

    二、条件变量

    1.认识条件变量

    条件变量是用来描述某种临界资源是否就绪的一种数据化描述。

    比如说存在一个共享的容器，生产者线程负责生产数据到容器内，消费者线程负责从容器中中读取数据。消费者线程发现容器为空时，就不应当去竞争锁，而是阻塞等待，直到生产者线程将数据生成到容器中。

    要想让消费者线程等待，那就必须使用条件变量标识容器的状态，那么就需要用到条件变量。

    那条件变量到底是什么呢？

    假设超市的架子进货一次只放一瓶水，只有这瓶水被买走后，供货商才会进货。

    此时又有很多消费者来买水，只有竞争能力强的消费者才能买到水，甚至他们会不停地买。竞争能力弱的消费者，买不到水。放在线程中也是一样的，竞争能力弱的消费者线程始终抢不到锁，产生了饥饿问题。

    为了解决这个问题，超市的工作人员设置了一个柜台，所有消费者都在这里排队，有一瓶水摆上货架，工作人员就允许一个消费者进去买，没有水所有人就需要在外面等待。而如果消费者想买第二瓶，就只能重新排队。而这个柜台和工作人员就相当于条件变量。

    多线程互斥访问临界资源时，为了让这些线程按一定顺序访问。通常会将这些线程都放在条件变量的等待队列中，当其他线程让条件变量符合线程的唤醒条件时，队列中的第一个线程就会去访问临界资源。

    2.条件变量的使用

    条件变量同样是一个类（pthread_cond_t），由POSIX线程库维护，使用的是POSIX标准。它也可以构造对象pthread_cond_t cond，cond就是条件变量的对象。

    以下是条件变量的一些成员函数和使用代码：

    int pthread_cond_init(pthread_cond_t *restrict cond,const pthread_condattr_t *restrict attr);

    头文件：pthread.h

    功能：初始化条件变量。

    参数：pthread_cond_t *restrict cond表示需要被初始化的条件变量的地址，const pthread_condattr_t *restrict attr表示条件变量的属性，一般都为nullptr。

    返回值：取消成功返回0，取消失败返回错误码。

    int pthread_cond_destroy(pthread_cond_t *cond);

    头文件：pthread.h

    功能：销毁互斥条件变量。

    参数：pthread_cond_t *cond表示需要被销毁的条件变量的地址。

    返回值：销毁成功返回0，失败返回错误码。

    pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

    如果是全局或static修饰的条件变量，使用上面语句初始化。

    int pthread_cond_wait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex);

    头文件：pthread.h

    功能：将调用该接口的线程放入传入的条件变量等待队列中。

    参数：pthread_cond_t *restrict cond创建的条件变量地址。pthread_mutex_t *restrict mutex互斥锁的地址（为什么传锁以后会解释）。

    返回值：放入等待队列成功返回0，失败返回错误码。

    int pthread_cond_signal(pthread_cond_t *cond);

    头文件：pthread.h

    功能：由另一个线程(通常是主线程)唤醒指定条件变量等待队列中的一个线程。

    参数：pthread_cond_t *cond表示需要唤醒的线程所在的等待队列的条件变量地址。

    返回值：唤醒成功返回0，失败返回错误码。

    int pthread_cond_broadcast(pthread_cond_t *cond);

    头文件：pthread.h

    功能：由另一个线程(通常是主线程)唤醒指定条件变量等待队列中的所有线程。

    参数：pthread_cond_t *cond表示需要唤醒的线程所在的等待队列的条件变量地址。

    返回值：唤醒成功返回0，失败返回错误码。

    3.代码改造

    我们使用条件变量使所有进程可以以一定顺序抢票。

    1. #include
    2. #include
    3. #include
    4. #include
    5. #include
    6. using namespace std;
    7.  
    8. #define NUM 5
    9.  
    10. pthread_mutex_t mutx = PTHREAD_MUTEX_INITIALIZER;//构建一个全局锁
    11. pthread_cond_t cond = PTHREAD_COND_INITIALIZER;//构建一个全局条件变量
    12. int tickets = 10;
    13.  
    14. class pthread_data
    15. {
    16. public:
    17.     pthread_t tid;
    18.     char buffer[64];
    19. };
    20.  
    21. void* start_routine(void* args)
    22. {
    23.     pthread_data* p = (pthread_data*)args;
    24.     string s;
    25.     s += p->buffer;
    26.     s += "Remaining tickets:";
    27.     while(1)
    28.     {
    29.         pthread_mutex_lock(&mutx);//加锁
    30.         pthread_cond_wait(&cond, &mutx);//线程进入等待队列
    31.         if(tickets > 0)
    32.         {
    33.             --tickets;
    34.             pthread_mutex_unlock(&mutx);//解锁
    35.             printf("%s%d\n", s.c_str(), tickets);//不修改临界资源，可以不包含在内
    36.         }
    37.         else
    38.         {
    39.             pthread_mutex_unlock(&mutx);//解锁
    40.             break;
    41.         }
    42.     }
    43.     pthread_exit(nullptr);
    44. }
    45.  
    46. int main()
    47. {
    48.     vector vpd;
    49.     //创建多个线程
    50.     for(int i = 0; i
    51.     {
    52.         pthread_data* pd = new pthread_data;
    53.         snprintf(pd->buffer, sizeof(pd->buffer), "thread%d buy ticket:",i+1);
    54.         pthread_create(&(pd->tid), nullptr, start_routine, (void*)pd);
    55.         vpd.push_back(pd);
    56.     }
    57.     //主线程唤醒其他线程
    58.     for(;;)
    59.     {
    60.         sleep(1);
    61.         pthread_cond_signal(&cond);
    62.         printf("main thread wake up a thread\n");
    63.     }
    64.     //线程回收
    65.     for(int i = 0; i
    66.     {
    67.         pthread_join(vpd[i]->tid, nullptr);
    68.         delete vpd[i];
    69.     }
    70.     return 0;
    71. }

    条件变量、票数和锁都是全局变量，每个线程申请锁成功就进入条件变量等待队列。主线程每个一秒钟唤醒一个等待的线程抢票。

    运行结果：

    可以发现线程按12345的顺序循环抢票。

    使用pthread_cond_broadcast()接口可以一次唤醒条件变量等待队列中的的所有线程，每隔一秒唤醒一次。

    运行结果：

    仍然是按照一定顺序抢票，只是进行抢票的线程是5个同时进行。

    三、基于阻塞队列的生产消费模型

    既然讲了这么半天的生产消费模型，那我们不妨实现一个。

    1.阻塞队列类

    首先需要搭建模型的框架，也就是实现包括生产者线程、消费者线程，还有一个储存数据的阻塞队列（缓冲区）的简单执行代码。

    （1）阻塞队列

    阻塞队列的实现有以下注意事项：

    阻塞队列可使用C++STL中的queue实现。

    由于阻塞队列是公共资源，所以必须保证它是线程安全的。生产者线程和消费者线程需要互斥访问，其实也只需要一把互斥锁就能实现生产者和消费者间的互斥。以后的生产者和生产者，消费者和消费者之间的互斥也是这样实现的。

    只有阻塞队列中有数据消费者才能读取，消费者读取时，生产者不能生产，必须在等待队列中。

    阻塞队列中没有数据或者数据未填满时，生产者才能生产，消费者在生产的时候，消费者不能读取，必须在等待队列中。

    1. template
    2. class Blockqueue
    3. {
    4. public:
    5.     //构造函数
    6.     Blockqueue(size_t capcity = MAX_NUM)
    7.         :_capcity(capcity)
    8.     {
    9.         pthread_mutex_init(&_mutx, nullptr);
    10.         pthread_cond_init(&_pcond, nullptr);
    11.         pthread_cond_init(&_ccond, nullptr);
    12.     }
    13.  
    14.     //析构函数
    15.     ~Blockqueue()
    16.     {
    17.         pthread_mutex_destroy(&_mutx);
    18.         pthread_cond_destroy(&_pcond);
    19.         pthread_cond_destroy(&_ccond);
    20.     }
    21.  
    22.     //生产数据
    23.     void push(const T& data);
    24.     //消费数据
    25.     void pop(T* data);
    26. private:
    27.     //检测队列是否装满
    28.     size_t Isfull() const
    29.     {
    30.         return (_q.size() == _capcity);
    31.     }
    32.  
    33.     std::queue _q;
    34.     pthread_mutex_t _mutx;
    35.     pthread_cond_t _pcond;
    36.     pthread_cond_t _ccond;
    37.     size_t _capcity;
    38. };

    为了保持生产者和消费者的互斥，我们对生产者和消费者各使用一个条件变量，用一个锁控制对阻塞队列的访问。

    需要给阻塞队列储存的数据量设置一个上限，生产者线程不能无限制地生产数据。

    构造函数中初始化锁和条件变量，在析构函数中释放互斥锁和条件变量。阻塞队列的容量设置一个合适的缺省值。

    （2）实现生产者的生产函数

    生产数据我们一般使用push作为函数名。

    1. void push(const T& data)
    2. {
    3.     //下面的判断就开始使用共享资源，需要加锁
    4.     pthread_mutex_lock(&_mutx);
    5.        
    6.     //如果当前队列是满的，那就需要将生产者线程加入等待队列挂起
    7.     if(Isfull())
    8.     {
    9.         pthread_cond_wait(&_pcond, &_mutx)
    10.     }
    11.     _q.push(data);
    12.     //唤醒消费者线程消费
    13.     pthread_cond_signal(&_ccond, &_mutx);
    14.     //解锁
    15.     pthread_mutex_unlock(&_mutx);
    16. }

    生产者线程调用生产数据接口时，先申请锁进入临界区。

    当阻塞队列满时，在生产者条件变量将线程放入等待队列中挂起。

    当阻塞队列不满时，生产者生产数据到阻塞队列，由于消费者线程全部在等待，所以需要唤醒消费者线程消费数据，否则生产者会一直生产至满。

    （3）实现消费者的消费函数

    消费数据我们一般使用pop作为函数名。

    1. void pop(T* data)
    2. {
    3.     //下面的判断就开始使用共享资源，需要加锁
    4.     pthread_mutex_lock(&_mutx);
    5.     //如果当前队列是空的，那就需要将消费者线程加入等待队列挂起
    6.     if(_q.empty())
    7.     {
    8.         pthread_cond_wait(&_ccond, &_mutx)
    9.     }
    10.     //将数据输出到data中并删除
    11.     *data = _q.front();
    12.     _q.pop();
    13.     //唤醒生产者线程生产
    14.     pthread_cond_signal(&_pcond, &_mutx);
    15.     //解锁
    16.     pthread_mutex_unlock(&_mutx);
    17. }

    消费者线程同样先申请到锁后进入临界区。

    当阻塞队列为空时，没有数据可以消费，消费者挂起等待。

    当阻塞队列为不为空时，消费者消费数据，由生产者线程全部在等待，所以需要唤醒生产者线程消费数据，否则消费者会一直消费至空。

    2.pthread_cond_wait为什么要传入锁

    使用pthread_cond_wait接口时，必须传如一个锁，这一点我们没有解释。

    线程在条件变量的等待队列中排队等待，其目的就是要拿到要访问临界资源的那把锁，申请到锁，线程就可以进入临界区。

    如果一个线程拿到了锁，而又发现自己不满足条件需要挂起等待。按照之前的知识，该线程应该继续拿着锁进入条件变量的等待队列。即使其他线程被唤醒了，因为申请不到到锁，无法访问共享资源，只能被挂起。

    为了解决这个问题，pthread_cond_wiat的实现大致分三个步骤：挂起该线程->释放锁->记录锁。

    也就是说，只要是持有锁的线程进入等待队列，就自动释放自己持有的锁，而释放锁也是原子性操作，不会引起线程安全问题。

    在最后，接口还会记录当前进程挂起时释放的锁，这也就解释了为什么唤醒线程的时候，pthread_cond_signal(pthread_cond_t* cond)只有一个参数，被唤醒线程根据记录就知道它该去申请哪把锁。

    我们最终得到以下结论：

    • 参数传递的这个锁必须是正在使用的锁。
    • 调用pthread_cond_wait函数的进程，会以原子性的方式，将锁释放，并将自己挂起。
    • 线程在被唤醒的时候，会自动重新获取挂起时传入的锁。

    3.生产者和消费者线程的执行函数

    （1）执行函数

    生产者不断生成随机数再将数据插入阻塞队列。消费者不断将随机数再从阻塞队列中拿出来。

    1. //生产者
    2. void* Produce(void* args)
    3. {
    4.     Blockqueue<int>* bq = (Blockqueue<int>*)args;
    5.     while(1)
    6.     {
    7.         sleep(1);
    8.         int data = rand()%10;
    9.         bq->push(data);
    10.         printf("生产数据完成，数据为：%d\n", data);
    11.     }
    12.     return nullptr;
    13. }
    14.  
    15. //消费者
    16. void* Consume(void* args)
    17. {
    18.     Blockqueue<int>* bq = (Blockqueue<int>*)args;
    19.     while(1)
    20.     {
    21.         sleep(1);
    22.         int data = 0;
    23.         bq->pop(&data);
    24.         printf("消费数据完成，数据为：%d\n", data);
    25.     }
    26.     return nullptr;
    27. }

    （2）试运行

    我们只创建一个生产者线程和一个消费者线程，并且让生产者每一秒生产一个数据，所以最终代码如下：

    1. #include
    2. #include
    3. #include
    4. #include
    5. #include
    6. using namespace std;
    7.  
    8. #define MAX_NUM 10
    9.  
    10. template<class T>
    11. class Blockqueue
    12. {
    13. public:
    14.     //构造函数
    15.     Blockqueue(size_t capcity = MAX_NUM)
    16.         :_capcity(capcity)
    17.     {
    18.         pthread_mutex_init(&_mutx, nullptr);
    19.         pthread_cond_init(&_pcond, nullptr);
    20.         pthread_cond_init(&_ccond, nullptr);
    21.     }
    22.  
    23.     //析构函数
    24.     ~Blockqueue()
    25.     {
    26.         pthread_mutex_destroy(&_mutx);
    27.         pthread_cond_destroy(&_pcond);
    28.         pthread_cond_destroy(&_ccond);
    29.     }
    30.  
    31.     //生产数据
    32.     void push(const T& data)
    33.     {
    34.         //下面的判断就开始使用共享资源，需要加锁
    35.         pthread_mutex_lock(&_mutx);
    36.         //如果当前队列是满的，那就需要将生产者线程加入等待队列挂起
    37.         if(Isfull())
    38.         {
    39.             pthread_cond_wait(&_pcond, &_mutx);
    40.         }
    41.         _q.push(data);
    42.         //唤醒消费者线程消费
    43.         pthread_cond_signal(&_ccond);
    44.         //解锁
    45.         pthread_mutex_unlock(&_mutx);
    46.     }
    47.  
    48.     //消费数据
    49.     void pop(T* data)
    50.     {
    51.         //下面的判断就开始使用共享资源，需要加锁
    52.         pthread_mutex_lock(&_mutx);
    53.         //如果当前队列是空的，那就需要将消费者线程加入等待队列挂起
    54.         if(_q.empty())
    55.         {
    56.             pthread_cond_wait(&_ccond, &_mutx);
    57.         }
    58.         //将数据输出到data中并删除
    59.         *data = _q.front();
    60.         _q.pop();
    61.         //唤醒生产者线程生产
    62.         pthread_cond_signal(&_pcond);
    63.         //解锁
    64.         pthread_mutex_unlock(&_mutx);
    65.     }
    66.  
    67. private:
    68.     //检测队列是否装满
    69.     size_t Isfull() const
    70.     {
    71.         return (_q.size() == _capcity);
    72.     }
    73.  
    74.     std::queue _q;
    75.     pthread_mutex_t _mutx;
    76.     pthread_cond_t _pcond;
    77.     pthread_cond_t _ccond;
    78.     size_t _capcity;
    79. };
    80.  
    81. //生产者
    82. void* Produce(void* args)
    83. {
    84.     Blockqueue<int>* bq = (Blockqueue<int>*)args;
    85.     while(1)
    86.     {
    87.         sleep(1);
    88.         int data = rand()%10;
    89.         bq->push(data);
    90.         printf("生产数据完成，数据为：%d\n", data);
    91.     }
    92.     return nullptr;
    93. }
    94.  
    95. //消费者
    96. void* Consume(void* args)
    97. {
    98.     Blockqueue<int>* bq = (Blockqueue<int>*)args;
    99.     while(1)
    100.     {
    101.         //sleep(1);
    102.         int data = 0;
    103.         bq->pop(&data);
    104.         printf("消费数据完成，数据为：%d\n", data);
    105.     }
    106.     return nullptr;
    107. }
    108.  
    109. int main()
    110. {
    111.     srand((unsigned int)time(nullptr));
    112.     Blockqueue<int>* bq = new Blockqueue<int>;
    113.  
    114.     pthread_t tids[2];
    115.  
    116.     pthread_create(&tids[0], nullptr, Produce, (void*)bq);
    117.     pthread_create(&tids[1], nullptr, Consume, (void*)bq);
    118.  
    119.     pthread_join(tids[0], nullptr);
    120.     pthread_join(tids[1], nullptr);
    121.  
    122.     return 0;
    123. }

    运行结果：

    4.部分细节处理

    对于一个消费者和一个生产者的模型而言，上面的代码的确足够了。但是生产消费模型的生产者和消费者都应当有多个，此时我们就需要对其进行修改。

    （1）伪唤醒问题

    如果现在有多个生产者线程在进行数据生产。当阻塞队列满了以后，所有生产者线程都会在条件变量的等待队列中等待。

    第一种情况，某个生产者线程调用挂起接口失败。

    pthread_cond_wait即使调用失败，它也只会返回错误码，并不能阻断执行流继续向下执行。所以，即使出错，生产者还是会生成数据到阻塞队列中。

    第二种情况，一个消费者线程一次唤醒所有生产者线程。

    比如，现在阻塞队列满了，消费者线程消费了一个数据并且唤醒了所有的生产者线程，那么很多个生产者向阻塞队列的一个空位置生成数据，同样会出现上述问题。

    上面这种情况被叫做伪唤醒，再生产者和消费者线程中都存在这个问题。所以需要让执行流只要不满足队列满或空的条件就循环执行pthread_cond_wait。执行成功了，该线程就被挂起了，执行流不再运行；执行失败了，执行流会一直调用pthread_cond_wait，执行流也不会向下操作数据。

    生产数据

    消费数据

    （2）解锁与唤醒的顺序

    在生产者生产完数据后，它需要做两件事：唤醒消费者线程和归还锁。

    对于消费者进程也是一样的：唤醒生产者线程和归还锁。

    由于唤醒线程是不对共享资源进行操作的，所以对于唤醒线程和解锁的顺序谁先谁后都可以。只是更建议先唤醒再解锁。

    5.处理任务的生产消费模型

    （1）代码改造

    我们使用生产消费模型可不是用来保存随机数的，而是用它处理任务的。

    我们可以写一个保存计算方法的任务类，从而实现一个随机数计算器。

    任务类

    1. //任务类
    2. class Task
    3. {
    4.     typedef std::function<int(int,int,char)> func_t;
    5. public:
    6.     //默认构造
    7.     Task()
    8.     {}
    9.     //构造函数
    10.     Task(int a, int b, char op, func_t func)
    11.         :_a(a)
    12.         ,_b(b)
    13.         ,_op(op)
    14.         ,_func(func)
    15.     {}
    16.  
    17.     //仿函数
    18.     string operator()()
    19.     {
    20.         int result = _func(_a, _b, _op);
    21.         char buffer[64];
    22.         snprintf(buffer, sizeof(buffer), "%d %c %d = %d\n", _a, _op, _b, result);
    23.         string s(buffer);
    24.         return s;
    25.     }
    26.  
    27.     //显示任务
    28.     string show_task()
    29.     {
    30.         char buffer[64];
    31.         snprintf(buffer, sizeof(buffer), "%d %c %d = ?\n", _a, _op, _b);
    32.         string s(buffer);
    33.         return s;
    34.     }
    35. private:
    36.     func_t _func;
    37.     int _a;
    38.     int _b;
    39.     char _op;
    40. };

    设置处理任务的函数，修改生产者和消费者线程执行的函数。

    1. //计算器函数
    2. const string ops = "+-*/%";
    3. int calculate(int a, int b, char op)
    4. {
    5.     int result = 0;
    6.     switch(op)
    7.     {
    8.         case '+':
    9.             result = a + b;
    10.             break;
    11.         case '-':
    12.             result = a - b;
    13.             break;
    14.         case '*':
    15.             result = a * b;
    16.             break;
    17.         case '/':
    18.         {
    19.             if(b == 0)
    20.                 cerr << "除数不能为0\n";
    21.             else
    22.                 result = a / b;
    23.         }
    24.             break;
    25.         case '%':
    26.         {
    27.             if(b == 0)
    28.                 cerr << "取模的数字不能为0\n";
    29.             else
    30.                 result = a % b;
    31.         }
    32.             break;
    33.         default:
    34.             break;
    35.     }
    36.     return result;
    37. }
    38.  
    39. //生产者
    40. void* Produce(void* args)
    41. {
    42.     Blockqueue* bq = (Blockqueue*)args;
    43.     while(1)
    44.     {
    45.         sleep(1);
    46.         int a = rand()%10;
    47.         int b = rand()%10;
    48.         int opnum = rand()%ops.size();
    49.         Task data(a, b, ops[opnum], calculate);
    50.         string s = "数据生产完成，需要计算：";
    51.         bq->push(data);
    52.         s += data.show_task().c_str();
    53.         cout << s;
    54.     }
    55.     return nullptr;
    56. }

    （2）运行

    在主线程中多创建几个线程就可以运行了。

    总代码如下：

    1. #include
    2. #include
    3. #include
    4. #include
    5. #include
    6. #include
    7. using namespace std;
    8.  
    9. #define MAX_NUM 10
    10.  
    11. //任务类
    12. class Task
    13. {
    14.     typedef std::function<int(int,int,char)> func_t;
    15. public:
    16.     //默认构造
    17.     Task()
    18.     {}
    19.     //构造函数
    20.     Task(int a, int b, char op, func_t func)
    21.         :_a(a)
    22.         ,_b(b)
    23.         ,_op(op)
    24.         ,_func(func)
    25.     {}
    26.  
    27.     //仿函数
    28.     string operator()()
    29.     {
    30.         int result = _func(_a, _b, _op);
    31.         char buffer[64];
    32.         snprintf(buffer, sizeof(buffer), "%d %c %d = %d\n", _a, _op, _b, result);
    33.         string s(buffer);
    34.         return s;
    35.     }
    36.  
    37.     //显示任务
    38.     string show_task()
    39.     {
    40.         char buffer[64];
    41.         snprintf(buffer, sizeof(buffer), "%d %c %d = ?\n", _a, _op, _b);
    42.         string s(buffer);
    43.         return s;
    44.     }
    45. private:
    46.     func_t _func;
    47.     int _a;
    48.     int _b;
    49.     char _op;
    50. };
    51.  
    52. template<class T>
    53. class Blockqueue
    54. {
    55. public:
    56.     //构造函数
    57.     Blockqueue(size_t capcity = MAX_NUM)
    58.         :_capcity(capcity)
    59.     {
    60.         pthread_mutex_init(&_mutx, nullptr);
    61.         pthread_cond_init(&_pcond, nullptr);
    62.         pthread_cond_init(&_ccond, nullptr);
    63.     }
    64.  
    65.     //析构函数
    66.     ~Blockqueue()
    67.     {
    68.         pthread_mutex_destroy(&_mutx);
    69.         pthread_cond_destroy(&_pcond);
    70.         pthread_cond_destroy(&_ccond);
    71.     }
    72.  
    73.     //生产数据
    74.     void push(const T& data)
    75.     {
    76.         //下面的判断就开始使用共享资源，需要加锁
    77.         pthread_mutex_lock(&_mutx);
    78.         //如果当前队列是满的，那就需要将生产者线程加入等待队列挂起
    79.         while(Isfull())
    80.         {
    81.             pthread_cond_wait(&_pcond, &_mutx);
    82.         }
    83.         _q.push(data);
    84.         //唤醒消费者线程消费
    85.         pthread_cond_signal(&_ccond);
    86.         //解锁
    87.         pthread_mutex_unlock(&_mutx);
    88.     }
    89.  
    90.     //消费数据
    91.     void pop(T* data)
    92.     {
    93.         //下面的判断就开始使用共享资源，需要加锁
    94.         pthread_mutex_lock(&_mutx);
    95.         //如果当前队列是空的，那就需要将消费者线程加入等待队列挂起
    96.         while(_q.empty())
    97.         {
    98.             pthread_cond_wait(&_ccond, &_mutx);
    99.         }
    100.         //将数据输出到data中并删除
    101.         *data = _q.front();
    102.         _q.pop();
    103.         //唤醒生产者线程生产
    104.         pthread_cond_signal(&_pcond);
    105.         //解锁
    106.         pthread_mutex_unlock(&_mutx);
    107.     }
    108.  
    109. private:
    110.     //检测队列是否装满
    111.     size_t Isfull() const
    112.     {
    113.         return (_q.size() == _capcity);
    114.     }
    115.  
    116.     std::queue _q;
    117.     pthread_mutex_t _mutx;
    118.     pthread_cond_t _pcond;
    119.     pthread_cond_t _ccond;
    120.     size_t _capcity;
    121. };
    122.  
    123. //计算器函数
    124. const string ops = "+-*/%";
    125. int calculate(int a, int b, char op)
    126. {
    127.     int result = 0;
    128.     switch(op)
    129.     {
    130.         case '+':
    131.             result = a + b;
    132.             break;
    133.         case '-':
    134.             result = a - b;
    135.             break;
    136.         case '*':
    137.             result = a * b;
    138.             break;
    139.         case '/':
    140.         {
    141.             if(b == 0)
    142.                 cerr << "除数不能为0\n";
    143.             else
    144.                 result = a / b;
    145.         }
    146.             break;
    147.         case '%':
    148.         {
    149.             if(b == 0)
    150.                 cerr << "取模的数字不能为0\n";
    151.             else
    152.                 result = a % b;
    153.         }
    154.             break;
    155.         default:
    156.             break;
    157.     }
    158.     return result;
    159. }
    160.  
    161. //生产者
    162. void* Produce(void* args)
    163. {
    164.     Blockqueue* bq = (Blockqueue*)args;
    165.     while(1)
    166.     {
    167.         sleep(1);
    168.         int a = rand()%10;
    169.         int b = rand()%10;
    170.         int opnum = rand()%ops.size();
    171.         Task data(a, b, ops[opnum], calculate);
    172.         string s = "数据生产完成，需要计算：";
    173.         bq->push(data);
    174.         s += data.show_task().c_str();
    175.         cout << s;
    176.     }
    177.     return nullptr;
    178. }
    179.  
    180. //消费者
    181. void* Consume(void* args)
    182. {
    183.     Blockqueue* bq = (Blockqueue*)args;
    184.     while(1)
    185.     {
    186.         //sleep(1);
    187.         Task data;
    188.         string s = "数据消费完成，计算结果为：";
    189.         bq->pop(&data);
    190.         string result = data();
    191.         s += result;
    192.         cout << s;
    193.     }
    194.     return nullptr;
    195. }
    196.  
    197. #define NUM_PRODUCE 3
    198. #define NUM_CONSUME 3
    199.  
    200. int main()
    201. {
    202.     srand((unsigned int)time(nullptr));
    203.     Blockqueue* bq = new Blockqueue;
    204.  
    205.     pthread_t ptids[NUM_PRODUCE];
    206.     pthread_t ctids[NUM_CONSUME];
    207.  
    208.     //创建多个生产者线程
    209.     for(int i = 0; i
    210.     {
    211.         pthread_create(&ptids[i], nullptr, Produce, (void*)bq);
    212.     }
    213.  
    214.     //创建多个消费者线程
    215.     for(int i = 0; i
    216.     {
    217.         pthread_create(&ctids[i], nullptr, Consume, (void*)bq);
    218.     }
    219.  
    220.     //回收所有线程
    221.     for(int i = 0; i
    222.     {
    223.         pthread_join(ptids[i], nullptr);
    224.     }
    225.     for(int i = 0; i
    226.     {
    227.         pthread_join(ctids[i], nullptr);
    228.     }
    229.  
    230.     return 0;
    231. }

    运行结果：

    有一个地方我一直没说，像cout 所以我们在打印信息时尽量使用单句printf或者cout

    6.生产消费模型为何高效

    该模型中，多个生产者线程向阻塞队列生成数据，多个消费者线程也从阻塞队列中消费数据。

    各生产消费者线程之间互斥关系，各线程对于阻塞队列的访问是串行的。同一时间访问阻塞队列的线程只有一个，拿这样又何来高效呢？

    注意观察程序的运行逻辑，我们能发现只有临界区的代码是串行的，其他代码所有线程都是并发执行的。这些非临界区的代码通常耗时长，而它们是并发的，所以该模型的效率就变得很高。

    结论：生产消费模型的高效不体现在对临界资源的访问上，而是体现在对非临界区代码的并发执行上。

    四、双阻塞队列的生产消费模型

    我么们可以使用上面的生产者消费者模型，将消费者处理完的计算任务保存成日志，并储存到磁盘上。

    所以该模型有两个阻塞队列，一个阻塞队列用于保存计算任务，另一个阻塞队列用于保存保存任务。原来的生产者线程还是生产者，原来的消费者作为中间的线程，既是消费者也是生产者，保存线程是消费者。

    1.编写类代码

    我们需要构建四个类：计算任务类、保存任务类、阻塞队列类和多队列集合类。

    • 计算任务类就是之前的Task类，我们其他代码不用动，对它改个CalTask的名字就可以。
    • 保存任务类与计算任务类很相似，需要传入一个string表示需要打印的信息，还有一个函数对象表示保存数据的具体执行函数。
    • 阻塞队列类也不用改
    • 多队列集合类是为了线程执行函数的void* args传参设计的，包含了两个阻塞队列指针。
    1. //计算任务类
    2. class CalTask
    3. {
    4.     typedef std::function<int(int,int,char)> func_t;
    5. public:
    6.     //默认构造
    7.     CalTask()
    8.     {}
    9.     //构造函数
    10.     CalTask(int a, int b, char op, func_t func)
    11.         :_a(a)
    12.         ,_b(b)
    13.         ,_op(op)
    14.         ,_func(func)
    15.     {}
    16.  
    17.     //仿函数
    18.     string operator()()
    19.     {
    20.         int result = _func(_a, _b, _op);
    21.         char buffer[64];
    22.         snprintf(buffer, sizeof(buffer), "%d %c %d = %d\n", _a, _op, _b, result);
    23.         string s(buffer);
    24.         return s;
    25.     }
    26.  
    27.     //显示任务
    28.     string show_task()
    29.     {
    30.         char buffer[64];
    31.         snprintf(buffer, sizeof(buffer), "%d %c %d = ?\n", _a, _op, _b);
    32.         string s(buffer);
    33.         return s;
    34.     }
    35. private:
    36.     func_t _func;
    37.     int _a;
    38.     int _b;
    39.     char _op;
    40. };
    41.  
    42. //保存任务类
    43. class SaveTask
    44. {
    45.     typedef function<void(const string&)> func_t;
    46. public:
    47.     //默认构造
    48.     SaveTask()
    49.     {}
    50.     //构造函数
    51.     SaveTask(string message, func_t func)
    52.         :_message(message)
    53.         ,_func(func)
    54.     {}
    55.  
    56.     //仿函数
    57.     void operator()()
    58.     {
    59.         _func(_message);
    60.     }
    61. private:
    62.     string _message;
    63.     func_t _func;
    64. };
    65.  
    66. //阻塞队列类
    67. template<class T>
    68. class Blockqueue
    69. {
    70. public:
    71.     //构造函数
    72.     Blockqueue(size_t capcity = MAX_NUM)
    73.         :_capcity(capcity)
    74.     {
    75.         pthread_mutex_init(&_mutx, nullptr);
    76.         pthread_cond_init(&_pcond, nullptr);
    77.         pthread_cond_init(&_ccond, nullptr);
    78.     }
    79.  
    80.     //析构函数
    81.     ~Blockqueue()
    82.     {
    83.         pthread_mutex_destroy(&_mutx);
    84.         pthread_cond_destroy(&_pcond);
    85.         pthread_cond_destroy(&_ccond);
    86.     }
    87.  
    88.     //生产数据
    89.     void push(const T& data)
    90.     {
    91.         //下面的判断就开始使用共享资源，需要加锁
    92.         pthread_mutex_lock(&_mutx);
    93.         //如果当前队列是满的，那就需要将生产者线程加入等待队列挂起
    94.         while(Isfull())
    95.         {
    96.             pthread_cond_wait(&_pcond, &_mutx);
    97.         }
    98.         _q.push(data);
    99.         //唤醒消费者线程消费
    100.         pthread_cond_signal(&_ccond);
    101.         //解锁
    102.         pthread_mutex_unlock(&_mutx);
    103.     }
    104.  
    105.     //消费数据
    106.     void pop(T* data)
    107.     {
    108.         //下面的判断就开始使用共享资源，需要加锁
    109.         pthread_mutex_lock(&_mutx);
    110.         //如果当前队列是空的，那就需要将消费者线程加入等待队列挂起
    111.         while(_q.empty())
    112.         {
    113.             pthread_cond_wait(&_ccond, &_mutx);
    114.         }
    115.         //将数据输出到data中并删除
    116.         *data = _q.front();
    117.         _q.pop();
    118.         //唤醒生产者线程生产
    119.         pthread_cond_signal(&_pcond);
    120.         //解锁
    121.         pthread_mutex_unlock(&_mutx);
    122.     }
    123.  
    124. private:
    125.     //检测队列是否装满
    126.     size_t Isfull() const
    127.     {
    128.         return (_q.size() == _capcity);
    129.     }
    130.  
    131.     std::queue _q;
    132.     pthread_mutex_t _mutx;
    133.     pthread_cond_t _pcond;
    134.     pthread_cond_t _ccond;
    135.     size_t _capcity;
    136. };
    137.  
    138. //多队列集合类
    139. template<class A,class B>
    140. class Blockqueues
    141. {
    142. public:
    143.     Blockqueue* _q1;
    144.     Blockqueue* _q2;
    145. };

    2.增加处理保存任务的函数

    我们之前对计算任务有一个处理函数calculate，那处理保存任务也同样需要一个函数Savedata。

    1. //保存函数
    2. void Savedata(const string& message)
    3. {
    4.     //需要保存信息的文件
    5.     char buffer[64] = "log.txt";
    6.     //按追加方式打开文件
    7.     FILE* fp = fopen(buffer, "a");
    8.     if(fp == nullptr)
    9.     {
    10.         cerr << "文件打开失败" << endl;
    11.         return;
    12.     }
    13.     fprintf(fp, "%s", message.c_str());
    14.     fclose(fp);
    15. }

    3.更改线程函数

    生产者线程此时依旧生产，只不过生产的位置是第一个阻塞队列。

    消费者线程此时也成为了后一个模型的生产者，那就需要添加向第二个队列生产的代码。

    保存者线程此时是消费者，从第二个队列中取任务执行。

    1. //生产者
    2. void* Produce(void* args)
    3. {
    4.     Blockqueues* bqs = (Blockqueues*)args;
    5.     //生产数据
    6.     while(1)
    7.     {
    8.         sleep(1);
    9.         int a = rand()%10;
    10.         int b = rand()%10;
    11.         int opnum = rand()%ops.size();
    12.         CalTask data(a, b, ops[opnum], calculate);
    13.         string s = "数据生产完成，需要计算：";
    14.         bqs->_q1->push(data);
    15.         s += data.show_task().c_str();
    16.         cout << s;
    17.     }
    18.     return nullptr;
    19. }
    20.  
    21. //消费者
    22. void* Consume(void* args)
    23. {
    24.     Blockqueues* bqs = (Blockqueues*)args;
    25.     while(1)
    26.     {
    27.         //消费数据
    28.         //sleep(1);
    29.         CalTask data;
    30.         string s1 = "数据消费完成，计算结果为：";
    31.         bqs->_q1->pop(&data);
    32.         string result = data();
    33.         s1 += result;
    34.         cout << s1;
    35.         //生成待保存的数据
    36.         string s2 = "数据保存任务推送完毕\n";
    37.         SaveTask task = SaveTask(result, Savedata);
    38.         bqs->_q2->push(task);
    39.          cout << s2;
    40.     }
    41.     return nullptr;
    42. }
    43.  
    44. void* Save(void* args)
    45. {
    46.     Blockqueues* bqs = (Blockqueues*)args;
    47.     while(1)
    48.     {
    49.         //sleep(1);
    50.         SaveTask data;
    51.         string s = "数据保存完成\n";
    52.         bqs->_q2->pop(&data);
    53.         data();
    54.         cout << s;
    55.     }
    56.     return nullptr;
    57. }

    4.更改main函数

    最后我们在main函数中创建两个队列，创建三种线程最后加上回收代码就可以了。

    所以总代码如下：

    produce_consume.h

    1. #include
    2. #include
    3. #include
    4. #include
    5. #include
    6. #include
    7. #include
    8. #define MAX_NUM 10
    9. using namespace std;
    10. //计算任务类
    11. class CalTask
    12. {
    13.     typedef std::function<int(int,int,char)> func_t;
    14. public:
    15.     //默认构造
    16.     CalTask()
    17.     {}
    18.     //构造函数
    19.     CalTask(int a, int b, char op, func_t func)
    20.         :_a(a)
    21.         ,_b(b)
    22.         ,_op(op)
    23.         ,_func(func)
    24.     {}
    25.  
    26.     //仿函数
    27.     string operator()()
    28.     {
    29.         int result = _func(_a, _b, _op);
    30.         char buffer[64];
    31.         snprintf(buffer, sizeof(buffer), "%d %c %d = %d\n", _a, _op, _b, result);
    32.         string s(buffer);
    33.         return s;
    34.     }
    35.  
    36.     //显示任务
    37.     string show_task()
    38.     {
    39.         char buffer[64];
    40.         snprintf(buffer, sizeof(buffer), "%d %c %d = ?\n", _a, _op, _b);
    41.         string s(buffer);
    42.         return s;
    43.     }
    44. private:
    45.     func_t _func;
    46.     int _a;
    47.     int _b;
    48.     char _op;
    49. };
    50.  
    51. //保存任务类
    52. class SaveTask
    53. {
    54.     typedef function<void(const string&)> func_t;
    55. public:
    56.     //默认构造
    57.     SaveTask()
    58.     {}
    59.     //构造函数
    60.     SaveTask(string message, func_t func)
    61.         :_message(message)
    62.         ,_func(func)
    63.     {}
    64.  
    65.     //仿函数
    66.     void operator()()
    67.     {
    68.         _func(_message);
    69.     }
    70. private:
    71.     string _message;
    72.     func_t _func;
    73. };
    74.  
    75. //阻塞队列类
    76. template<class T>
    77. class Blockqueue
    78. {
    79. public:
    80.     //构造函数
    81.     Blockqueue(size_t capcity = MAX_NUM)
    82.         :_capcity(capcity)
    83.     {
    84.         pthread_mutex_init(&_mutx, nullptr);
    85.         pthread_cond_init(&_pcond, nullptr);
    86.         pthread_cond_init(&_ccond, nullptr);
    87.     }
    88.  
    89.     //析构函数
    90.     ~Blockqueue()
    91.     {
    92.         pthread_mutex_destroy(&_mutx);
    93.         pthread_cond_destroy(&_pcond);
    94.         pthread_cond_destroy(&_ccond);
    95.     }
    96.  
    97.     //生产数据
    98.     void push(const T& data)
    99.     {
    100.         //下面的判断就开始使用共享资源，需要加锁
    101.         pthread_mutex_lock(&_mutx);
    102.         //如果当前队列是满的，那就需要将生产者线程加入等待队列挂起
    103.         while(Isfull())
    104.         {
    105.             pthread_cond_wait(&_pcond, &_mutx);
    106.         }
    107.         _q.push(data);
    108.         //唤醒消费者线程消费
    109.         pthread_cond_signal(&_ccond);
    110.         //解锁
    111.         pthread_mutex_unlock(&_mutx);
    112.     }
    113.  
    114.     //消费数据
    115.     void pop(T* data)
    116.     {
    117.         //下面的判断就开始使用共享资源，需要加锁
    118.         pthread_mutex_lock(&_mutx);
    119.         //如果当前队列是空的，那就需要将消费者线程加入等待队列挂起
    120.         while(_q.empty())
    121.         {
    122.             pthread_cond_wait(&_ccond, &_mutx);
    123.         }
    124.         //将数据输出到data中并删除
    125.         *data = _q.front();
    126.         _q.pop();
    127.         //唤醒生产者线程生产
    128.         pthread_cond_signal(&_pcond);
    129.         //解锁
    130.         pthread_mutex_unlock(&_mutx);
    131.     }
    132.  
    133. private:
    134.     //检测队列是否装满
    135.     size_t Isfull() const
    136.     {
    137.         return (_q.size() == _capcity);
    138.     }
    139.  
    140.     std::queue _q;
    141.     pthread_mutex_t _mutx;
    142.     pthread_cond_t _pcond;
    143.     pthread_cond_t _ccond;
    144.     size_t _capcity;
    145. };
    146.  
    147. //多队列集合类
    148. template<class A,class B>
    149. class Blockqueues
    150. {
    151. public:
    152.     Blockqueue* _q1;
    153.     Blockqueue* _q2;
    154. };

    produce_consume.cc

    1. #include"produce_consume.h"
    2. using namespace std;
    3.  
    4. //计算器函数
    5. const string ops = "+-*/%";
    6. int calculate(int a, int b, char op)
    7. {
    8.     int result = 0;
    9.     switch(op)
    10.     {
    11.         case '+':
    12.             result = a + b;
    13.             break;
    14.         case '-':
    15.             result = a - b;
    16.             break;
    17.         case '*':
    18.             result = a * b;
    19.             break;
    20.         case '/':
    21.         {
    22.             if(b == 0)
    23.                 cerr << "除数不能为0\n";
    24.             else
    25.                 result = a / b;
    26.         }
    27.             break;
    28.         case '%':
    29.         {
    30.             if(b == 0)
    31.                 cerr << "取模的数字不能为0\n";
    32.             else
    33.                 result = a % b;
    34.         }
    35.             break;
    36.         default:
    37.             break;
    38.     }
    39.     return result;
    40. }
    41.  
    42. //保存函数
    43. void Savedata(const string& message)
    44. {
    45.     //需要保存信息的文件
    46.     char buffer[64] = "log.txt";
    47.     //按追加方式打开文件
    48.     FILE* fp = fopen(buffer, "a");
    49.     if(fp == nullptr)
    50.     {
    51.         cerr << "文件打开失败" << endl;
    52.         return;
    53.     }
    54.     fprintf(fp, "%s", message.c_str());
    55.     fclose(fp);
    56. }
    57.  
    58. //生产者
    59. void* Produce(void* args)
    60. {
    61.     Blockqueues* bqs = (Blockqueues*)args;
    62.     //生产数据
    63.     while(1)
    64.     {
    65.         sleep(1);
    66.         int a = rand()%10;
    67.         int b = rand()%10;
    68.         int opnum = rand()%ops.size();
    69.         CalTask data(a, b, ops[opnum], calculate);
    70.         string s = "数据生产完成，需要计算：";
    71.         bqs->_q1->push(data);
    72.         s += data.show_task().c_str();
    73.         cout << s;
    74.     }
    75.     return nullptr;
    76. }
    77.  
    78. //消费者
    79. void* Consume(void* args)
    80. {
    81.     Blockqueues* bqs = (Blockqueues*)args;
    82.     while(1)
    83.     {
    84.         //消费数据
    85.         //sleep(1);
    86.         CalTask data;
    87.         string s1 = "数据消费完成，计算结果为：";
    88.         bqs->_q1->pop(&data);
    89.         string result = data();
    90.         s1 += result;
    91.         cout << s1;
    92.         //生成待保存的数据
    93.         string s2 = "数据保存任务推送完毕\n";
    94.         SaveTask task = SaveTask(result, Savedata);
    95.         bqs->_q2->push(task);
    96.          cout << s2;
    97.     }
    98.     return nullptr;
    99. }
    100.  
    101. void* Save(void* args)
    102. {
    103.     Blockqueues* bqs = (Blockqueues*)args;
    104.     while(1)
    105.     {
    106.         //sleep(1);
    107.         SaveTask data;
    108.         string s = "数据保存完成\n";
    109.         bqs->_q2->pop(&data);
    110.         data();
    111.         cout << s;
    112.     }
    113.     return nullptr;
    114. }
    115.  
    116. #define NUM_PRODUCE 3
    117. #define NUM_CONSUME 3
    118. #define NUM_SAVE 3
    119.  
    120. int main()
    121. {
    122.     srand((unsigned int)time(nullptr));
    123.     Blockqueues* bqs = new Blockqueues();
    124.     bqs->_q1 = new Blockqueue();
    125.     bqs->_q2 = new Blockqueue();
    126.  
    127.     pthread_t ptids[NUM_PRODUCE];
    128.     pthread_t ctids[NUM_CONSUME];
    129.     pthread_t stids[NUM_SAVE];
    130.  
    131.     //创建多个生产者线程
    132.     for(int i = 0; i
    133.     {
    134.         pthread_create(&ptids[i], nullptr, Produce, (void*)bqs);
    135.     }
    136.  
    137.     //创建多个消费者线程
    138.     for(int i = 0; i
    139.     {
    140.         pthread_create(&ctids[i], nullptr, Consume, (void*)bqs);
    141.     }
    142.  
    143.     //创建多个保存者线程
    144.     for(int i = 0; i
    145.     {
    146.         pthread_create(&stids[i], nullptr, Save, (void*)bqs);
    147.     }
    148.  
    149.     //回收所有线程
    150.     for(int i = 0; i
    151.     {
    152.         pthread_join(ptids[i], nullptr);
    153.     }
    154.     for(int i = 0; i
    155.     {
    156.         pthread_join(ctids[i], nullptr);
    157.     }
    158.     for(int i = 0; i
    159.     {
    160.         pthread_join(stids[i], nullptr);
    161.     }
    162.     return 0;
    163. }

    运行结果：

    目录中确实多了一个log.txt，也能看出线程的具体执行轨迹。

    基于阻塞队列的生产者消费者模型是线程同步与互斥的充分应用，现实中很多场景都可以应用，是线程中的一大杀器。

  • 相关阅读:
    合合信息大模型加速器亮相WAIC大会：文档解析与文本识别新突破
    多线程的概念（多线程的代码实现）
    练习敲代码速度/提高打字速度
    10大数据恢复软件可以帮助您恢复电脑数据
    基于springboot外委员工后台管理系统毕业设计源码101157
    云上的云服务器、裸金属以及容器可以为 Lakehouse 提供海量的计算资源
    基于Java web的校园滴滴代驾管理系统 毕业设计-附源码260839
    linux内核中修改TCP MSS值
    批量调整视频饱和度和色度，提升你的视频剪辑效率！
    无需公网IP，教学系统如何实现远程一站式管理维护？
  • 原文地址：https://blog.csdn.net/qq_65285898/article/details/132654397