文章目录
- 二、线程同步
- 条件变量
- 1、条件变量的概念
- 2、同步概念与竞态条件
- 3、条件变量函数初始化
- 4、条件变量函数销毁
- 5、条件变量函数等待
- 6、条件变量函数唤醒等待
- 生产者消费者模型
- 1、理论部分
- 2、“3 2 1”原则
- 3、基于阻塞队列的生产者消费者模型
- POSIX信号量
- 1、信号量的概念
- 2、信号量操作函数
- ①初始化信号量
- ②销毁信号量
- ③等待信号量 P( )
- ④发布信号量 V( )
- 3、基于环形队列的生产者消费者模型
- ①认识环形队列
- ②多线程下的环形队列
- ③具体代码实现
- ④改进：改成多生产者、多消费者模型
二、线程同步
例如：当一个线程访问队列时，发现队列为空时只能等待，直到其他线程将一个节点添加到队列中，显然，只有互斥锁的情况下，我们比较困难的知道临界资源的状态，此时我们需要一种机制或者策略来知道临界资源的状态——条件变量！
条件变量
1、条件变量的概念
条件变量类似于if语句，他有“真”、“假”两个状态，在条件变量使用的过程中，一个线程等待条件为“真”，另一个线程在使用完临界资源后将条件设置为“真”，唤醒阻塞在等待条件变量为“真”的线程，执行其任务。在这个过程中，必须保证在并行或者并发的条件下使得条件变量在“真”“假”两个状态之间正确转换，所以条件变量一般需要和互斥锁配合使用，由此实现对于临界资源的互斥访问！
2、同步概念与竞态条件
- 同步：在保证数据安全的前提下，让线程能够按照某种特定的顺序访问临界资源，从而有效避免饥饿问题，叫做同步。
- 竞态条件：因为时序问题而导致程序异常，我们常称为竞态条件。在线程场景下，这种问题也不难理解。
3、条件变量函数初始化
如果你想要的创建条件变量，只需要在类中的或者全局创建一个pthread_cond_t类型的变量，这样原生线程库就会自动帮你创建好条件变量，然后你就可以开始一系列的操作。
| 函数名称 | pthread_cond_init |
|---|---|
| 函数功能 | 初始化条件变量 |
| 头文件 | #include<pthread.h> |
| 函数原型 | int pthread_cond_init(pthread_cond_t *restrict cond, const pthread_condattr_t *restrict attr); |
| 参数 | cond：指向要初始化的条件变量的指针 attr：指向条件变量属性的指针，来指定条件变量的一些属性，但是我们这里给NULL，表明使用默认属性来初始化该条件变量 |
| 返回值 | 0：成功 !0：失败 |
4、条件变量函数销毁
| 函数名称 | pthread_cond_destroy |
|---|---|
| 函数功能 | 销毁条件变量 |
| 头文件 | #include<pthread.h> |
| 函数原型 | int pthread_cond_destroy(pthread_cond_t *cond); |
| 参数 | cond：需要销毁的条件变量 |
| 返回值 | 0：成功 !0：失败 |
5、条件变量函数等待
| 函数名称 | pthread_cond_wait |
|---|---|
| 函数功能 | 等待条件变量被设置 |
| 头文件 | #include<pthread.h> |
| 函数原型 | int pthread_cond_wait( pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex); |
| 参数 | cond：需要等待的条件变量 mutex：与条件变量相关的互斥锁 |
| 返回值 | 0：成功 !0：失败 |
6、条件变量函数唤醒等待
| 函数名称 | pthread_cond_signal |
|---|---|
| 函数功能 | 唤醒一个等待的线程 |
| 头文件 | #include<pthread.h> |
| 函数原型 | int pthread_cond_signal(pthread_cond_t *cond); |
| 参数 | cond：需要通知的条件变量的指针 |
| 返回值 | 0：成功 !0：失败 |
问题：pthread_cond_signal等待的是哪一个线程？
在条件变量下的等待队列里等待的第一个线程！
| 函数名称 | pthread_cond_broadcast |
|---|---|
| 函数功能 | 唤醒所有等待的线程 |
| 头文件 | #include<pthread.h> |
| 函数原型 | int pthread_cond_broadcast(pthread_cond_t *cond); |
| 参数 | cond：需要广播通知的条件变量的指针 |
| 返回值 | 0：成功 !0：失败 |
修改部分代码：
结果展示：
我们现在想要实现一个简单的逻辑，boss发出指令去控制工作线程工作：
#include <iostream>
#include <string>
#include <pthread.h>
#include <unistd.h>
pthread_mutex_t mtx;
pthread_cond_t cond;
// ctrl线程控制 work线程，定期完成任务
void *ctrl(void *args)
{
std::string name=(char*)(args);
while(true)
{
std::cout<<"master say:begin work!"<<std::endl;
// 唤醒在条件变量下等待的一个线程
pthread_cond_signal(&cond);
sleep(5);
}
}
void *work(void *args)
{
int number = *(int *)args;
delete (int *)args;
while (true)
{
// 等待条件变量被设置
pthread_cond_wait(&cond,&mtx);
std::cout << "work: " << number << " is working..." << std::endl;
}
}
// 一个主线程控制另外3个线程
int main()
{
#define NUM 3
pthread_mutex_init(&mtx,nullptr);
pthread_cond_init(&cond,nullptr);
pthread_t master;
pthread_t worker[NUM];
pthread_create(&master, nullptr, ctrl, (void*)"boss");
for (int i = 0; i < NUM; i++)
{
int *number = new int(i);
pthread_create(worker + i, nullptr, work, (void *)number);
}
for (int i = 0; i < NUM; i++)
{
pthread_join(worker[i], nullptr);
}
pthread_join(master, nullptr);
pthread_mutex_destroy(&mtx);
pthread_cond_destroy(&cond);
return 0;
}
结果展示：
现象：老板发出命令，线程才开始工作，而且每个线程都是按序工作(2->1->0)。
当2号线程，1号线程他们跑完自己的任务，去干什么了？
答案：他又跑去当前条件变量下等待去了
结论：每个线程运行完后都去死循环等待了，条件变量内部一定存在一个等待队列。
生产者消费者模型
1、理论部分
超市——生产者消费者模型
主要有两大优点
- 提高效率
- 将生产环节与消费环节解耦
为什么要有超市？
收集需求，减少交易成本
2、“3 2 1”原则
- 3种关系
生产者与生产者：竞争、互斥关系
消费者与消费者：竞争、互斥关系
生产者与消费者：互斥、同步关系 - 2个角色
生产者与消费者，通常n ：n - 1个交易场所
超市->(交易场所，通常就是一段缓冲区，可以是一段内存空间、STL容器等)
3、基于阻塞队列的生产者消费者模型
当前我们的策略：
1、当生产满时，就不要生产了(不要竞争锁了)，而应该让消费者来消费
2、当消费空了，就不应该消费(不要竞争锁了)，而应该让生产者来生产
所以我们的条件变量有两个：
template <class T>
class BlockQueue
{
private:
std::queue<T> bq_; // 阻塞队列
int cap_;
pthread_mutex_t mtx_; // 保护临界资源的锁
pthread_cond_t is_full_; // bq_满的，消费者在该条件变量下等待
pthread_cond_t is_empty; // bq_空的，生产者在该条件变量下等待
puulic:
// code
};
const &：输入型参数
*：输出型参数
&：输入输出型参数
void Push(const T &in)
{
LockQueue();
if (IsFull())
{
//等待会将线程挂起，且当前持有锁(别人来申请会造成死锁问题)
ProducterWait();
}
bq_.push(in);
if(bq_.size()>cap_/2)WakeupConsumer();
UnLockQueue();
//WakeupConsumer(); // 在内在外都可以
}
void Pop(T *out)
{
LockQueue();
if (IsEmpty())
{
//让消费者等待
ConsumerWait();
}
*out = bq_.front();
bq_.pop();
if(bq_.size()<cap_/2)WakeupProducter();
UnLockQueue();
//WakeupProducter();
}
pthread_cond_wait函数的功能：
1、调用时，会首先自动释放mtx_，然后再挂起自己！
2、返回时，会首先自动竞争锁，获得锁之后，才能返回！
void ProducerWait()
{
pthread_cond_wait(&is_empty_, &mtx_);
}
void ConsumerWait()
{
pthread_cond_wait(&is_full_, &mtx_);
}
下面我们来实现我们全部的生产消费逻辑代码：
#pragma once
#include <iostream>
#include <string>
#include <queue>
#include <unistd.h>
#include <pthread.h>
namespace ns_blockqueue
{
const int default_cap = 5;
template <class T>
class BlockQueue
{
private:
std::queue<T> bq_; // 阻塞队列
int cap_;
pthread_mutex_t mtx_; // 保护临界资源的锁
pthread_cond_t is_full_; // bq_满的，消费者在该条件变量下等待
pthread_cond_t is_empty_; // bq_空的，生产者在该条件变量下等待
private:
// 判断队列是否为满
bool IsFull()
{
return bq_.size() == cap_;
}
bool IsEmpty()
{
return bq_.size() == 0;
}
void LockQueue()
{
pthread_mutex_lock(&mtx_);
}
void UnlockQueue()
{
pthread_mutex_unlock(&mtx_);
}
void ProducerWait()
{
pthread_cond_wait(&is_empty_, &mtx_);
}
void ConsumerWait()
{
pthread_cond_wait(&is_full_, &mtx_);
}
void WakeupConsumer()
{
pthread_cond_signal(&is_full_);
}
void WakeupProducer()
{
pthread_cond_signal(&is_empty_);
}
public:
// 构造函数中对锁和条件变量进行初始化
BlockQueue(int cap = default_cap)
: cap_(cap)
{
pthread_mutex_init(&mtx_, nullptr);
pthread_cond_init(&is_empty_, nullptr);
pthread_cond_init(&is_full_, nullptr);
}
~BlockQueue()
{
pthread_mutex_destroy(&mtx_);
pthread_cond_destroy(&is_empty_);
pthread_cond_destroy(&is_full_);
}
public:
// 向队列中放数据
void Push(const T &in)
{
LockQueue();
if (IsFull()) // bug?
{
// 条件不满足时，让生产线程等待
ProducerWait();
}
bq_.push(in);
if (bq_.size() > cap_ / 2)
WakeupConsumer();
UnlockQueue();
}
// 从队列中出数据
void Pop(T *out)
{
LockQueue();
if (IsEmpty())
{
ConsumerWait();
}
*out = bq_.front();
bq_.pop();
if (bq_.size() < cap_ / 2)
WakeupProducer();
UnlockQueue();
}
};
}
//
#include "BlockQueue.hpp"
#include <time.h>
#include <stdlib.h>
#include <unistd.h>
using namespace ns_blockqueue;
void *consumer(void *args)
{
BlockQueue<int> *bq = (BlockQueue<int> *)(args);
while (true)
{
sleep(2); //控制同步节奏
int data=0;
bq->Pop(&data);
std::cout<<"消费者消费了一个数据："<<data<<std::endl;
}
}
void *producer(void *args)
{
BlockQueue<int> *bq = (BlockQueue<int> *)(args);
while (true)
{
// 制造数据
//sleep(2); //控制同步节奏
int data = rand() % 20 + 1;
std::cout << "生产者生了一个数据：" << data << std::endl;
bq->Push(data);
}
}
int main()
{
// 随机数种子
srand((unsigned int)time(nullptr));
BlockQueue<int> *bq = new BlockQueue<int>();
pthread_t c;
pthread_t p;
pthread_create(&c, nullptr, consumer, (void *)bq); // 将bq传过去，两个线程就看到了同一份资源（缓冲区队列）
pthread_create(&p, nullptr, producer, (void *)bq);
pthread_join(c, nullptr);
pthread_join(p, nullptr);
return 0;
}
结果展示：
此处有个bug！！！！
使用if来进行条件判断是不太完善的！
如果出现了如图所示的两种情况，当函数ProducterWait返回时，那么if语句是默认顺序向下执行的，并不满足生产函数的条件，所以我们需要再次进行判断，将if换成while。
我们需要进行条件检测的时候，这里需要使用循环的方式，来保证退出循环一定是因为条件不满足导致的！
LockQueue();
// if (IsFull()) // bug?
while (IsEmpty())
{
// 条件不满足时，让生产线程等待
ProducerWait();
}
bq_.push(in);
if (bq_.size() > cap_ / 2)
WakeupConsumer(); // 只有生产者知道，消费者什么时候可以消费
UnlockQueue();
改进
生产和消费，传输数据只是第一步
我们还需要解决两个问题
1、数据怎么来的？耗时吗？
2、数据怎么处理？耗时吗？
所以我们还需要添加一个场景–>任务处理
// "task.hpp"
#pragma once
#include <iostream>
#include <string>
#include <pthread.h>
namespace ns_task
{
class Task
{
private:
int x_;
int y_;
char op_; // +-*/%
public:
Task() {}
Task(int x, int y, char op) : x_(x), y_(y), op_(op)
{
}
int Run()
{
int res = 0;
switch (op_)
{
case '+':
res = x_ + y_;
break;
case '-':
res = x_ - y_;
break;
case '*':
res = x_ * y_;
break;
case '/':
res = x_ / y_;
break;
case '%':
res = x_ % y_;
break;
default:
std::cout << "错误的运算" << std::endl;
break;
}
std::cout << "当前任务正在被[" << pthread_self() << "]处理" << x_ << op_ << y_ << "=" << res << std::endl;
std::cout<<"-----------------------------"<<std::endl;
return res;
}
int operator()()
{
return Run();
}
~Task() {}
};
} // namespace ns_task
///
#include "task.hpp"
using namespace ns_task;
void *consumer(void *args)
{
BlockQueue<Task> *bq = (BlockQueue<Task> *)(args);
while (true)
{
Task t;
bq->Pop(&t);
// 任务处理
t();
}
}
void *producer(void *args)
{
BlockQueue<Task> *bq = (BlockQueue<Task> *)(args);
std::string ops = "+-*/%";
while (true)
{
// 制造数据
int x = rand() % 20 + 1;
int y = rand() % 10 + 1;
char op = ops[rand() % 5];
Task t(x, y, op);
std::cout << "生产者派发了一个任务: " << x << op << y << "=?" << std::endl;
bq->Push(t);
sleep(1);
}
}
int main()
{
// 随机数种子
srand((unsigned int)time(nullptr));
BlockQueue<Task> *bq = new BlockQueue<Task>();
pthread_t c,p;
pthread_t c1,c2,c3,c4;
pthread_create(&c, nullptr, consumer, (void*)bq);
pthread_create(&c1, nullptr, consumer, (void*)bq);
pthread_create(&c2, nullptr, consumer, (void*)bq);
pthread_create(&c3, nullptr, consumer, (void*)bq);
pthread_create(&c4, nullptr, consumer, (void*)bq);
pthread_create(&p, nullptr, producer, (void*)bq);
pthread_join(c, nullptr);
pthread_join(c1, nullptr);
pthread_join(c2, nullptr);
pthread_join(c3, nullptr);
pthread_join(c4, nullptr);
pthread_join(p, nullptr);
return 0;
}
运行结果：
POSIX信号量
1、信号量的概念
信号量的本质是一把计数器，描述临界资源中资源数目的大小(最多能有多少资源分配给线程)
2、信号量操作函数
同样和互斥锁和条件变量的使用相同，需要定义一个sem_t类型的变量，就可以用户信号量的操作函数
①初始化信号量
int sem_init(sem_t *sem, int pshared, unsigned int value);
参数：
pshared：默认给0即可，0表示线程间共享 ，非0表示进行间共享
value：信号量的初值
②销毁信号量
int sem_destroy(sem_t *sem);
③等待信号量 P( )
int sem_wait(sem_t *sem);
④发布信号量 V( )
int sem_post(sem_t *sem);
3、基于环形队列的生产者消费者模型
①认识环形队列
问题：环形队列什么时候为空？什么时候为满？
刚刚开始的时候为空，拿和放在同一个位置；之后满了，拿和放还是在同一个位置。
言下之意：有数据在队列中就是——放!=拿
我们判断队列为空还是为满有两种方法：
- 1、计数器
- 2、镂空一个位置：放前先做判断，当前位置+1 != 拿，可以放；否则不放
此处的环形结构采用数组通过模运算来模拟环形队列
②多线程下的环形队列
目标：实现一个基于环形队列的生产者消费者模型！在多线程的情况下，来进行环形队列的并发访问！
①基本原理
-
生产者和消费者刚开始的时候，队列为空，指向同一个位置！——应该让生产者先访问临界资源
-
生产者和消费者在队列为满时，也是指向同一个位置！——应该让消费者先访问临界资源
-
言下之意：队列不为空，或者不为满的时候，生产者和消费者一定指向的不是同一个位置！！！！
结论：
前两点说明，队列为空或者为满时，不能让生产者和消费者同时访问临界资源(满足互斥和局部同步特性)。
最后一点说明，生产和消费可以并发执行。
②基本实现的思想：
生产者最关心什么资源？——队列中的空位置也是资源！
消费者最关心什么资源？——队列中的数据就是资源！
- 规则1：生产者不能把消费者围成一个圈(意思就是不能超过，超过就是覆盖)
- 规则2：消费者不能超过生产者
- 规则3：当指向同一个位置时，要根据空、满状态判断，谁先执行！
- 其他规则：除此之外，生产和消费都可以并发执行！
③具体代码实现
基于信号量的循环队列的单生产和单消费的生产者消费者模型
// ring_queue.hpp
#pragma once
#include <iostream>
#include <string>
#include <vector>
#include <semaphore.h>
namespace ns_ring_queue
{
const int g_cap_default = 10;
template <class T>
class RingQueue
{
private:
std::vector<T> ring_queue_;
int cap_;
sem_t blank_sem_; // 生产者关心空格资源
sem_t data_sem_; // 消费者关心数据资源
// 表明生产、消费的位置
int c_step_;
int p_step_;
public:
RingQueue(int cap = g_cap_default)
: ring_queue_(cap), cap_(cap)
{
sem_init(&blank_sem_, 0, cap);
sem_init(&data_sem_, 0, 0);
c_step_ = p_step_ = 0;
}
~RingQueue()
{
sem_destroy(&blank_sem_);
sem_destroy(&data_sem_);
}
public:
// 目前高优先级的先实现单生产和单消费
void Push(const T &in)
{
// 生产接口
// 1.申请信号量
sem_wait(&blank_sem_); // P(空位置)
ring_queue_[p_step_] = in;
sem_post(&data_sem_); // V(数据)
p_step_++;
p_step_ %= cap_;
}
void Pop(T *out)
{
// 消费接口
sem_wait(&data_sem_);
*out = ring_queue_[c_step_];
sem_post(&blank_sem_);
c_step_++;
c_step_ %= cap_;
}
};
}
/
#include <iostream>
#include <pthread.h>
#include <time.h>
#include "ring_queue.hpp"
#include <unistd.h> // sleep()
using namespace ns_ring_queue;
void *consumer(void *args)
{
RingQueue<int> *rq = (RingQueue<int> *)args;
while (true)
{
int data = 0;
rq->Pop(&data);
std::cout << "消费了一个数据：" << data << std::endl;
sleep(1); //让消费者慢一点
}
}
void *producer(void *args)
{
RingQueue<int> *rq = (RingQueue<int> *)args;
while (true)
{
// 生产随机数
int data = rand() % 20 + 1;
std::cout << "生产了一个数据：" << data << std::endl;
rq->Push(data);
}
}
int main()
{
srand((unsigned int)time(0));
RingQueue<int> *rq = new RingQueue<int>();
pthread_t c, p;
pthread_create(&c, nullptr, consumer, (void *)rq);
pthread_create(&p, nullptr, producer, (void *)rq);
pthread_join(c, nullptr);
pthread_join(p, nullptr);
return 0;
}
结果展示：
生产和消费函数最大的亮点：申请自己关心的资源；释放对方关心的资源！
对于生产函数，他关心的是空格子，所以先要申请空格子资源，申请成功，代码顺序往后执行，空格上就放上了数据，释放时，虽然格子被占据了，但是数据却多了，所以要释放数据资源！！
同理消费函数亦如此。
④改进：改成多生产者、多消费者模型
// task.hpp
#pragma once
#include <iostream>
#include <string>
#include <pthread.h>
namespace ns_task
{
class Task
{
private:
int x_;
int y_;
char op_; // +-*/%
public:
Task() {}
Task(int x, int y, char op) : x_(x), y_(y), op_(op)
{
}
std::string Show()
{
std::string message = std::to_string(x_);
message += op_;
message += std::to_string(y_);
message +="=?";
return message;
}
int Run()
{
int res = 0;
switch (op_)
{
case '+':
res = x_ + y_;
break;
case '-':
res = x_ - y_;
break;
case '*':
res = x_ * y_;
break;
case '/':
res = x_ / y_;
break;
case '%':
res = x_ % y_;
break;
default:
std::cout << "错误的运算" << std::endl;
break;
}
std::cout << "当前任务正在被[" << pthread_self() << "]处理" << x_ << op_ << y_ << "=" << res << std::endl;
std::cout << "-----------------------------" << std::endl;
return res;
}
int operator()()
{
return Run();
}
~Task() {}
};
}
/
// ring_queue.hpp
#pragma once
#include <iostream>
#include <string>
#include <vector>
#include <semaphore.h>
#include <pthread.h>
namespace ns_ring_queue
{
const int g_cap_default = 10;
template <class T>
class RingQueue
{
private:
std::vector<T> ring_queue_;
int cap_;
sem_t blank_sem_; // 生产者关心空格资源
sem_t data_sem_; // 消费者关心数据资源
// 表明生产、消费的位置
int c_step_;
int p_step_;
// 引入两把互斥锁，分别维护生产者内部和消费者内部的关系
pthread_mutex_t c_mtx_;
pthread_mutex_t p_mtx_;
public:
RingQueue(int cap = g_cap_default)
: ring_queue_(cap), cap_(cap)
{
sem_init(&blank_sem_, 0, cap);
sem_init(&data_sem_, 0, 0);
c_step_ = p_step_ = 0;
pthread_mutex_init(&c_mtx_, nullptr);
pthread_mutex_init(&p_mtx_, nullptr);
}
~RingQueue()
{
sem_destroy(&blank_sem_);
sem_destroy(&data_sem_);
pthread_mutex_destroy(&c_mtx_);
pthread_mutex_destroy(&p_mtx_);
}
public:
// 目前高优先级的先实现单生产和单消费
void Push(const T &in)
{
// 生产接口
// 申请信号量
// 多生产和多消费的优势并不在这里，而是在于并发的获取和处理任务
sem_wait(&blank_sem_); // P(空位置)
pthread_mutex_lock(&p_mtx_);
ring_queue_[p_step_] = in;
// 它也变成了临界资源
p_step_++;
p_step_ %= cap_;
pthread_mutex_unlock(&p_mtx_);
sem_post(&data_sem_); // V(数据)
}
void Pop(T *out)
{
// 消费接口
sem_wait(&data_sem_);
pthread_mutex_lock(&p_mtx_);
*out = ring_queue_[c_step_];
c_step_++;
c_step_ %= cap_;
pthread_mutex_unlock(&p_mtx_);
sem_post(&blank_sem_);
}
};
}
/**
* @file ring_cp.cc
* @author sjj
* @brief 改进基于环形队列的生产者消费者模型--->多生产者多消费者模型
* @version 0.2
* @date 2022-09-11
*
* @copyright Copyright (c) 2022
*
*/
#include <iostream>
#include <pthread.h>
#include <time.h>
#include "ring_queue.hpp"
#include <unistd.h> // sleep()
#include "task.hpp"
using namespace ns_ring_queue;
using namespace ns_task;
void *consumer(void *args)
{
RingQueue<Task> *rq = (RingQueue<Task> *)args;
while (true)
{
Task t;
rq->Pop(&t);
t();
sleep(1); //让消费者慢一点
}
}
void *producer(void *args)
{
RingQueue<Task> *rq = (RingQueue<Task> *)args;
const std::string ops="+-*/%";
while (true)
{
// 生产随机数
int x = rand() % 20 + 1;
int y=rand()%10+1;
char op=ops[rand()%5];
Task t(x,y,op);
std::cout << "生产了一个任务：" << t.Show() << " 我是[" << pthread_self() << "]" << std::endl;
rq->Push(t);
}
}
int main()
{
srand((unsigned int)time(0));
RingQueue<Task> *rq = new RingQueue<Task>();
pthread_t c0, c1, c2, c3, p0, p1, p2;
pthread_create(&c0, nullptr, consumer, (void *)rq);
pthread_create(&c1, nullptr, consumer, (void *)rq);
pthread_create(&c2, nullptr, consumer, (void *)rq);
pthread_create(&c3, nullptr, consumer, (void *)rq);
pthread_create(&p0, nullptr, producer, (void *)rq);
pthread_create(&p1, nullptr, producer, (void *)rq);
pthread_create(&p2, nullptr, producer, (void *)rq);
pthread_join(c0, nullptr);
pthread_join(c1, nullptr);
pthread_join(c2, nullptr);
pthread_join(c3, nullptr);
pthread_join(p0, nullptr);
pthread_join(p1, nullptr);
pthread_join(p2, nullptr);
return 0;
}
结果展示：
