Linux -- 生产消费模型之环形队列、信号量
目录
前言
环形队列
如何用环形队列实现生产消费模型？
信号量
sem_t
sem_init（初始化信号量）
sem_destroy（销毁信号量）
什么是PV操作？
sem_wait（P操作，减少信号量）
sem_post（V操作，增加信号量）
用信号量实现环形队列
代码
Thread.hpp 代码：
Task.hpp 代码：
RingQueue.hpp 代码：
test.cc 代码：
结果：
前言
Linux -- 生产消费模型-CSDN博客
https://blog.csdn.net/2301_76973016/article/details/144733376?spm=1001.2014.3001.5501在之前实现生产消费模型的代码中，我们把整个阻塞队列（缓冲区）视为一个整体进行访问，整个缓冲区只能由生产者或者消费者访问，不能生产者和消费者同时访问，不能边读取数据边放入数据，为了达到这一目标，我们可以用环形队列来实现。
环形队列
环形队列是一种线性数据结构，其操作基于先进先出（FIFO, First In First Out）原则，但它的最后一个元素连接到第一个元素以形成一个环。这种结构可以更有效地利用存储空间，因为它允许在队列的末尾达到数组的末端时“环绕”回到数组的开始。
如何用环形队列实现生产消费模型？
我们用环形队列来实现缓冲区！
当缓冲区中没有数据，即环形队列为空，即消费者和生产者指向同一个位置时时，消费者没办法消费，所以只能生产者先走，让生产者先去输出数据，此时消费者休眠。生产者每生产一个数据，数据就会占一个空间，所以生产者往前走了一步，_productor_step++，继续生产，当环形队列中有数据时，就可以唤醒消费者：
当缓冲区中的数据满了，即环形队列为满，如果生产者继续往前走，即回到了环形队列的头，生产者继续生产则会覆盖之前的数据，那么这时只能让消费者先走，让消费者取出数据，腾出空间，此时生产者休眠，当有空间时，就可以唤醒生产者：
当环形队列既不为空，也不为满时，此时生产者和消费者谁先走都没关系，两者互不干扰，也就是生产者和消费者可以实现并发地访问缓冲区！直到环形队列再次为空或者为满时，再来调整让消费者先走还是生产者先走。
从上面的三种情况可以看出：
- 生产者关注的是环形队列的空间资源，而消费者关注的是环形队列中的数据资源，；
- 生产者每生产一个数据，空间资源 -1，数据资源 +1；
- 消费者每取出一个数据，空间资源 +1，数据资源 -1；
- 两者都申请了自己需要的资源，释放了对方需要的资源。
为了实现这一过程，我们采用信号量来实现。
信号量
信号量主要用于多线程或多进程编程中的同步机制，以确保多个线程或进程能够安全地共享资源或协调它们的行为。通过使用信号量，可以控制同时访问特定资源的线程或进程数量，从而避免竞争条件和其他并发问题。
信号量其实是一个计数器，是对资源的预定机制：
- 如果信号量申请成功，相当于资源申请成功了，我们就可以使用这个资源；
- 如果信号量申请失败，就会在信号量下的队列阻塞等待。
sem_t
sem_t是 POSIX 标准定义的一种用于实现进程间同步的互斥锁和信号量的数据类型。
sem_init（初始化信号量）
#include <semaphore.h>int sem_init(sem_t *sem, int pshared, unsigned int value);
sem_t *sem: 指向要初始化的信号量对象的指针。
int pshared: 决定信号量是否可以在多个进程间共享。
- 如果设置为
0，则信号量仅限于同一进程内的线程之间共享。- 如果设置为非零值，则信号量可以在不同进程之间共享（具体支持取决于操作系统实现）。
unsigned int value: 信号量的初始值。这个值代表了可以同时进行的并发操作的数量。
sem_destroy（销毁信号量）
#include <semaphore.h>int sem_destroy(sem_t *sem);
sem_t *sem: 指向要销毁的信号量对象的指针。
什么是PV操作？
PV操作是操作系统中用于进程同步的一种机制，通常与信号量相关联。PV操作的名字来源于荷兰语“Probeer te Verlagen”（尝试减少）和“Verhoog”（增加），在中文文献中也常被称为“P操作”和“V操作”。这两个操作一起构成了信号量的基本操作，用来控制对共享资源的访问，以确保多个进程或线程之间能够正确地协调工作。
- P操作（Proberen te Verlagen, 尝试减少）是指尝试将信号量的值减一；
- V操作（Verhoog, 增加）是指将信号量的值加一。
PV操作主要用于解决临界区问题、互斥访问以及进程间的同步。它们确保了多个进程或线程不会同时访问相同的共享资源，从而避免了数据竞争和不一致的状态。
sem_wait（P操作，减少信号量）
#include <semaphore.h>int sem_wait(sem_t *sem);
sem_t *sem: 指向要操作的信号量对象的指针。
如果信号量的当前值为零，那么
sem_wait调用将会阻塞，直到信号量的值变为正数，然后将信号量的值减一。这通常用来控制对共享资源的访问，确保同一时间只有一个或指定数量的线程可以访问该资源。
sem_post（V操作，增加信号量）
#include <semaphore.h>int sem_post(sem_t *sem);
sem_t *sem: 指向要操作的信号量对象的指针。
它会增加信号量的值，并唤醒一个或多个等待该信号量的线程。如果当前没有线程在等待这个信号量，那么这次增加操作只是简单地增加了信号量的值。
用信号量实现环形队列
有了信号量之后，就可以把 环形队列 的使用 用信号量来表示，下面用伪代码简单了解一下，其中取模是为了当走到环形队列的尾时，再走一步可以回到环形队列的头：
- 对于生产者：
P(room);//申请空间资源//信号量申请成功，可以使用该空间，且空间资源-1 ring_queue[_productor_step] = in;//生产 _productor_step++; _productor_step %= n;V(data);//数据资源+1
- 对于消费者：
P(data);//申请数据资源//信号量申请成功，可以使用该数据，且数据资源-1 out = ring_queue[_consumer_step];//消费 _consumer_step++; _consumer_step %= n;V(room);//空间资源+1
代码
Thread.hpp 代码：
#ifndef __THREAD_HPP__
#define __THREAD_HPP__#include <iostream>
#include <string>
#include <unistd.h>
#include <functional>
#include <pthread.h>namespace ThreadModule
{template <typename T>using func_t = std::function<void(T &, std::string name)>;// typedef std::function<void(const T&)> func_t;template <typename T>class Thread{public:void Excute(){_func(_data, _threadname);}public:Thread(func_t<T> func, T &data, const std::string name = "none-name"): _func(func), _data(data), _threadname(name), _stop(true){}static void *threadroutine(void *args) // 类成员函数，形参是有this指针的！！{Thread<T> *self = static_cast<Thread<T> *>(args);self->Excute();return nullptr;}bool Start(){int n = pthread_create(&_tid, nullptr, threadroutine, this);if (!n){_stop = false;return true;}else{return false;}}void Detach(){if (!_stop){pthread_detach(_tid);}}void Join(){if (!_stop){pthread_join(_tid, nullptr);}}std::string name(){return _threadname;}void Stop(){_stop = true;}~Thread() {}private:pthread_t _tid;std::string _threadname;T &_data; // 为了让所有的线程访问同一个全局变量func_t<T> _func;bool _stop;};
}#endifTask.hpp 代码：
#pragma once#include<iostream>
#include<functional>using Task=std::function<void()>;void Download()
{std::cout<<"this is a download task"<<std::endl;
}RingQueue.hpp 代码：
#pragma once#include <iostream>
#include <string>
#include <semaphore.h>
#include <pthread.h>
#include <vector>template <typename T>
class RingQueue
{
private: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): _cap(cap), _consumer_step(0), _productor_step(0), _ring_queue(cap){sem_init(&_room_sem, 0, _cap); // 空间的信号量初始值为 _capsem_init(&_data_sem, 0, 0);pthread_mutex_init(&_consumer_mutex, nullptr);pthread_mutex_init(&_productor_mutex, nullptr);}void Enqueue(const T &in){P(_room_sem); // 入环形队列时，先申请空间信号量，占空间Lock(_productor_mutex);//因为 _productor_step 是临界资源，需要加锁保护临界资源// 既拿到锁了，也有空间，就可以生产_ring_queue[_productor_step++] = in; // 生产_productor_step %= _cap;Unlock(_productor_mutex);V(_data_sem); // 放入一个数据，数据信号量增加}void Pop(T *out){P(_data_sem);Lock(_consumer_mutex);//因为 _consumer_step 是临界资源，需要加锁保护临界资源*out = _ring_queue[_consumer_step++];_consumer_step %= _cap;Unlock(_consumer_mutex);V(_room_sem); // 空出一个空间}~RingQueue(){sem_destroy(&_data_sem);sem_destroy(&_room_sem);pthread_mutex_destroy(&_consumer_mutex);pthread_mutex_destroy(&_productor_mutex);}private:std::vector<T> _ring_queue; // 用数组模拟环形队列int _cap; // 环形队列的上限// 访问队列的下标int _consumer_step;int _productor_step;// 信号量sem_t _room_sem;sem_t _data_sem;// 互斥锁pthread_mutex_t _consumer_mutex;pthread_mutex_t _productor_mutex;
};test.cc 代码：
#include "Task.hpp"
#include "Thread.hpp"
#include "RingQueue.hpp"#include <string>
#include <vector>
#include <unistd.h>using namespace ThreadModule;
// 给类填好模板，并重命名
using ringqueue_t = RingQueue<Task>;// 创建线程
void InitComm(std::vector<Thread<ringqueue_t>> *threads, int num, ringqueue_t &rq, func_t<ringqueue_t> func, std::string who)
{for (int i = 0; i < num; i++){std::string name = "thread-" + std::to_string(i + 1) + "-" + who;threads->emplace_back(func, rq, name);}
}void Consumer(ringqueue_t &rq, std::string name)
{while (true){sleep(2);Task t;rq.Pop(&t); // 取出任务t(); // 处理任务std::cout << "Consumer: [ " << name << " ]" << std::endl;}
}void Productor(ringqueue_t &rq, std::string name)
{// 开始生产while (true){rq.Enqueue(Download); // 放入任务std::cout << "Productor: [ " << name << " ]" << std::endl;}
}
void StartAll(std::vector<Thread<ringqueue_t>> &threads)
{for (auto &thread : threads){std::cout << "start: " << thread.name() << std::endl;thread.Start();}
}
void InitProductor(std::vector<Thread<ringqueue_t>> *threads, int num, ringqueue_t &rq)
{InitComm(threads, num, rq, Productor, "Productor");
}void InitConsumer(std::vector<Thread<ringqueue_t>> *threads, int num, ringqueue_t &rq)
{InitComm(threads, num, rq, Consumer, "Consumer");
}void WaitAll(std::vector<Thread<ringqueue_t>> &threads)
{for (auto &thread : threads){thread.Join();}
}
int main()
{// new 调用构造函数，创建一个上限为 10 的环形队列ringqueue_t *rq = new ringqueue_t(10);// 给线程传任务所需要的参数，并把线程放入数组管理起来std::vector<Thread<ringqueue_t>> threads;InitProductor(&threads, 3, *rq);InitConsumer(&threads, 5, *rq);StartAll(threads);WaitAll(threads);return 0;
}结果：
每次运行的结果都不一样：
