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;};
}#endif

Task.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;
}

结果：

每次运行的结果都不一样：