机械加工网名大全,百度seo站长,手表网站上没有价格,wordpress nginx 302文章目录 一、POSIX 信号量二、POSIX 信号量的接口2.1 sem_init——初始化信号量2.2 sem_destroy——销毁信号量2.3 sem_wait——等待信号量2.4 sem_post——发布信号量 三、基于环形队列的生产消费者模型3.1 单生产单消费模型3.2 多生产多消费模型3.3 基于任务的多生产多消费模… 文章目录 一、POSIX 信号量二、POSIX 信号量的接口2.1 sem_init——初始化信号量2.2 sem_destroy——销毁信号量2.3 sem_wait——等待信号量2.4 sem_post——发布信号量 三、基于环形队列的生产消费者模型3.1 单生产单消费模型3.2 多生产多消费模型3.3 基于任务的多生产多消费模型 四、结语 一、POSIX 信号量
共享资源也可以被看成多份只要规定好每个线程的访问区域即可此时就可以让多线程去并发的访问临界资源。
POSIX 信号量和 SystemV 信号量作用相同都是用于同步操作达到无冲突的访问共享资源的目的。但 POSIX 可以用于线程间同步。信号量本质是一把计数器用来描述可用资源数目的申请信号量时其实就已经在间接的做判断看资源是否就绪了只要申请到信号量那么说明资源一定是就绪的。
信号量只能保证不让多余的线程来访问共享资源即当前共享资源有十份信号量不会允许同时有十一个线程来访问临界资源。但是具体的资源分配是通过程序员编码去实现的。如果出现一个共享资源同时被两个线程访问就属于程序员的编码 Bug。
二、POSIX 信号量的接口
2.1 sem_init——初始化信号量
#include semaphore.h
int sem_init(sem_t *sem, int pshared, unsigned int value);sem要初始化的信号量 pshared0表示线程间共享非0表示进程间共享。 value信号量初始值
2.2 sem_destroy——销毁信号量
int sem_destroy(sem_t *sem);2.3 sem_wait——等待信号量
int sem_wait(sem_t *sem); //P()功能会将信号量的值减1
2.4 sem_post——发布信号量
int sem_post(sem_t *sem);//V() 功能发布信号量表示资源使用完毕可以归还资源了。将信号量的值加1
三、基于环形队列的生产消费者模型
只要生产和消费不访问同一个格子那么生产和消费就可以同时进行。那生产和消费什么时候会指向同一个数据呢答案是队列为空和为满的时候。 基于环形队列的生产消费者模型必须遵守以下三个原则 当生产和消费指向同一个资源的时候只能一个人访问。为空的时候由生产者去访问为满的时候由消费者去访问 消费者不能超过生产者 生产者不能把消费者套圈因为这样会导致数据被覆盖
生产者最关心还剩多少空间空间数量消费者最关系还剩多少数据数据数量。因为有两种资源所以需要定义两个信号量。
3.1 单生产单消费模型
// RingQueue.hpp
#pragma once#include pthread.h
#include vector
#include semaphore.htemplateclass T
class RingQueue
{
private:static const int defaultcap 5;void P(sem_t *sem) // 申请一个信号量{sem_wait(sem); }void V(sem_t *sem) // 归还一个信号量{sem_post(sem);}
public:RingQueue(int cap defaultcap):ringqueue_(cap), cap_(cap), c_step(0), p_step(0){sem_init(cdata_sem, 0, 0);sem_init(pspace_sem, 0, cap_);}void Push(const T data) // 生产行为{P(pspace_sem);ringqueue_[p_step] data;V(cdata_sem);p_step;p_step % cap_;}void Pop(T *out) // 消费行为{P(cdata_sem);*out ringqueue_[c_step];V(pspace_sem);c_step;c_step % cap_;}~RingQueue(){sem_destroy(cdata_sem);sem_destroy(pspace_sem);}
private:std::vectorT ringqueue_; // 环形队列int cap_; // 容量int c_step; // 消费者下一个要消费的位置int p_step; // 生产者下一个要生产的位置sem_t cdata_sem; // 数据资源sem_t pspace_sem; // 空间资源
};// main.cc
#include RingQueue.hpp
#include iostream
#include unistd.husing namespace std;void *Consumer(void *args)
{RingQueueint *rq static_castRingQueueint*(args);while(true){int data 0;rq-Pop(data);cout Consumer is running... get a data: data endl;// 模拟处理数据usleep(1000000);}
}void *Productor(void *args)
{RingQueueint *rq static_castRingQueueint*(args);while(true){// 获取数据usleep(10000); // 模拟获取数据int data rand() % 10;rq-Push(data);cout Productor is running... produce a data: data endl;}
}int main()
{srand((unsigned int)time(nullptr));pthread_t c, p;RingQueueint *rq new RingQueueint();pthread_create(c, nullptr, Consumer, rq);pthread_create(p, nullptr, Productor, rq);pthread_join(c, nullptr);pthread_join(p, nullptr);return 0;
}互斥与同步的体现当生产下标和消费下标相同的时候只允许一个来访问这就是互斥性的体现。当队列为空的时候让生产者去访问资源当队列为满的时候让消费者去访问资源这就是在指向同一个位置时让生产和消费具有一定的顺序性这就是同步性的体现。当队列不为空或不为满的时候生产下标和消费下标不同此时两个线程并发执行并没有体现出很强的互斥特性。
3.2 多生产多消费模型
此时需要对下标资源进行保护。因为生产下标和消费下标各自只有一份不允许同时有多个生产线程去访问生产下标消费线程也一样。因此需要通过加锁来实现生产线程之间的互斥和消费线程之间的互斥。
先加锁还是先申请信号量答案是先申请信号量以生产线程为例这样可以让所有生产线程并发的去执行什么意思呢如果是先加锁再申请信号量的话因为始终只有一个生产者线程能够申请到锁所以也就只有一个生产者线程能去申请信号量其他生产者线程只能干巴巴的等待锁被释放。这时申请锁和申请信号量的动作是串行的。而先申请信号量的话可以保证虽然只有一个线程能够申请到锁但是其他没有锁的线程也可以不用闲着可以先去申请信号量因为信号量的申请是原子的因此也不需要加锁进行保护只要能申请到信号量就说明资源还有此时那些申请到信号量的线程就可能等待锁被释放拿到锁之后就可以去执行相应的代码了。
// RingQueue.hpp
#pragma once#include pthread.h
#include vector
#include semaphore.h
#include stringtemplateclass T
class RingQueue
{
private:static const int defaultcap 5;void P(sem_t *sem) // 申请一个信号量{sem_wait(sem); }void V(sem_t *sem) // 归还一个信号量{sem_post(sem);}void Lock(pthread_mutex_t *mutex){pthread_mutex_lock(mutex);}void Unlock(pthread_mutex_t *mutex){pthread_mutex_unlock(mutex);}
public:RingQueue(int cap defaultcap):ringqueue_(cap), cap_(cap), c_step(0), p_step(0){sem_init(cdata_sem, 0, 0);sem_init(pspace_sem, 0, cap_);pthread_mutex_init(c_mutex, nullptr);pthread_mutex_init(p_mutex, nullptr);}void Push(const T data) // 生产行为{P(pspace_sem);Lock(p_mutex);ringqueue_[p_step] data;p_step;p_step % cap_;Unlock(p_mutex);V(cdata_sem);}void Pop(T *out) // 消费行为{P(cdata_sem); // 信号量资源是不需要保护的因为它的操作是原子的临界区中的代码要尽可能的少所以不需要把信号量的申请放在加锁之后Lock(c_mutex);*out ringqueue_[c_step];c_step;c_step % cap_;Unlock(c_mutex);V(pspace_sem);}~RingQueue(){sem_destroy(cdata_sem);sem_destroy(pspace_sem);pthread_mutex_destroy(c_mutex);pthread_mutex_destroy(p_mutex);}
private:std::vectorT ringqueue_; // 环形队列int cap_; // 容量int c_step; // 消费者下一个要消费的位置int p_step; // 生产者下一个要生产的位置sem_t cdata_sem; // 数据资源sem_t pspace_sem; // 空间资源pthread_mutex_t c_mutex; // 对消费下标的保护pthread_mutex_t p_mutex; // 对生产下标的保护
};template class T
class Message
{
public:Message(std::string thread_name, RingQueueT *ringqueue):thread_name_(thread_name), ringqueue_(ringqueue){}std::string get_thread_name(){return thread_name_;}RingQueueT *get_ringqueue(){return ringqueue_;}
private:std::string thread_name_;RingQueueT *ringqueue_;
};// main.cc
#include RingQueue.hpp
#include iostream
#include unistd.h
#include vectorusing namespace std;void *Consumer(void *args)
{Messageint *message static_castMessageint *(args);RingQueueint *rq message-get_ringqueue();string name message-get_thread_name();while (true){int data 0;rq-Pop(data);printf(%s is running... get a data: %d\n, name.c_str(), data);// 模拟处理数据// usleep(1000000);}
}void *Productor(void *args)
{Messageint *message static_castMessageint *(args);RingQueueint *rq message-get_ringqueue();string name message-get_thread_name();while (true){// 获取数据// usleep(1000000); // 模拟获取数据int data rand() % 10;rq-Push(data);printf(%s is running... produce a data: %d\n, name.c_str(), data);usleep(1000000);}
}int main()
{srand((unsigned int)time(nullptr));pthread_t c[3], p[5];RingQueueint *rq new RingQueueint(); vectorMessageint* messages; for (int i 0; i 5; i){Messageint *message new Messageint(Produttor Thread to_string(i), rq);pthread_create(p i, nullptr, Productor, message);messages.push_back(message);}for (int i 0; i 3; i){Messageint *message new Messageint(Consumer Thread to_string(i), rq);pthread_create(c i, nullptr, Consumer, message);messages.push_back(message);}for (int i 0; i 3; i){pthread_join(c[i], nullptr);}for (int i 0; i 5; i){pthread_join(p[i], nullptr);}for (auto message : messages){delete message;}delete rq;return 0;
}3.3 基于任务的多生产多消费模型
RingQueue 的内容不变
// Task.h
#include iostream
#include stringenum
{DIVERROR 1,MODERROR,UNKNOWERRROR
};class Task
{
public:Task(int a 0, int b 0, char op ):data1_(a), data2_(b), op_(op), result_(0), exitcode_(0){}void run(){switch(op_){case :result_ data1_ data2_;break;case -:result_ data1_ - data2_;break;case *:result_ data1_ * data2_;break;case /:if(data2_ 0) exitcode_ DIVERROR;else result_ data1_ / data2_;break;case %:if(data2_ 0) exitcode_ MODERROR;else result_ data1_ % data2_;break;default:exitcode_ UNKNOWERRROR;break;}}std::string result_to_string(){std::string ret std::to_string(data1_);ret ;ret op_;ret ;ret std::to_string(data2_);ret ;ret ;ret ;ret std::to_string(result_);ret [exitcode: ;ret std::to_string(exitcode_);ret ];return ret;}std::string get_task(){std::string ret std::to_string(data1_);ret ;ret op_;ret ;ret std::to_string(data2_);ret ;ret ;ret ;ret ?;return ret;}
private:int data1_;int data2_;char op_;int result_;int exitcode_;
};// main.cc
#include RingQueue.hpp
#include iostream
#include unistd.h
#include vector
#include Task.husing namespace std;const std::string opers -*/%;void *Consumer(void *args)
{MessageTask *message static_castMessageTask *(args);RingQueueTask *rq message-get_ringqueue();string name message-get_thread_name();while (true){// 获取任务// int data 0;Task task;rq-Pop(task);// 对任务做处理task.run();printf(%s is running... get a data: %s\n, name.c_str(), task.result_to_string().c_str());// 模拟处理数据// usleep(1000000);}
}void *Productor(void *args)
{MessageTask *message static_castMessageTask *(args);RingQueueTask *rq message-get_ringqueue();string name message-get_thread_name();int len opers.size();while (true){// 获取数据// usleep(1000000); // 模拟获取数据// int data rand() % 10;// 模拟获取数据int data1 rand() % 10 1; // [1, 10]usleep(10);int data2 rand() % 13; // [0, 13]usleep(10);char op opers[rand() % len];Task task(data1, data2, op);// 生产数据rq-Push(task);// printf(%s is running... produce a data: %d\n, name.c_str(), data);printf(%s is running... produce a Task: %s\n, name.c_str(), task.get_task().c_str());usleep(1000000);}
}int main()
{srand((unsigned int)time(nullptr));pthread_t c[3], p[2];RingQueueTask *rq new RingQueueTask(); vectorMessageTask* messages; for (int i 0; i 5; i){MessageTask *message new MessageTask(Produttor Thread to_string(i), rq);pthread_create(p i, nullptr, Productor, message);messages.push_back(message);}for (int i 0; i 3; i){MessageTask *message new MessageTask(Consumer Thread to_string(i), rq);pthread_create(c i, nullptr, Consumer, message);messages.push_back(message);}// 等待子线程for (int i 0; i 3; i){pthread_join(c[i], nullptr);}for (int i 0; i 5; i){pthread_join(p[i], nullptr);}// 释放资源for (auto message : messages){delete message;}delete rq;return 0;
}四、结语
今天的分享到这里就结束啦如果觉得文章还不错的话可以三连支持一下春人的主页还有很多有趣的文章欢迎小伙伴们前去点评您的支持就是春人前进的动力
