信号量实现基于环形队列的生产者消费者模型

信号量实现基于环形队列的生产者消费者模型

引用

CSDN

1.

https://blog.csdn.net/D5486789_/article/details/145992525

本文详细介绍了如何使用信号量实现基于环形队列的生产者消费者模型。文章首先讲解了信号量的基本概念和接口使用方法，然后通过具体的代码实现，展示了如何在环形队列中使用信号量来实现线程同步。

1. 前言

多个线程在互斥的情况下并发访问同一个共享资源时，由于竞争锁能力强弱的原因，可能造成竞争锁能力弱的线程饥饿，为了解决这个问题，我们需要让这些线程按照一定的顺序访问同一个共享资源，也就是实现线程同步。实现线程同步可以使用条件变量，对条件变量感兴趣的读者可以看看这篇文章 —— 线程同步。除了使用条件变量实现线程同步，我们还可以使用信号量来实现线程同步。同时在博主之前的文章中有讲过 基于Blockqueue的生产者消费者模型，因此，我还想在这篇文章中讲解如何使用信号量实现 基于环形队列的生产者消费者模型。

2. 信号量

信号量的概念

在生活中，我们肯定见过不少预定资源的机制，预定资源大概的意思就是说，资源不一定要被我持有才是我的，只要我预定了，在未来的一段时间内就是我的。而资源是有限的，当资源数量小于0时，预定资源就会失败，相当于有一把计数器，记录着资源的数量。而信号量的本质就是一把计数器，描述临界资源数量的计数器。

执行流可以通过申请和释放信号量来对信号量这把计数器做减1加1操作，当执行流申请信号量成功，计数器就减1，表示预定资源成功，该执行流就可以继续向后运行；当这把计数器减为0之后，表示没有资源了，申请信号量就会失败，执行流就会阻塞在当前位置，直到申请信号量成功才能继续向后运行；当释放信号量，计数器就加1，表示资源数量增加一个，其他执行流就可以继续申请信号量了。

• 执行流可以是进程 或者 线程，我们以线程为例。
• 其中，申请信号量操作叫做信号量的P操作，释放信号量操作叫做信号量的V操作。

也就是说，我们可以通过PV操作来对临界资源进行保护，而所有的线程想要访问临界资源都必须先申请信号量，也就是说所有的线程都得先看到同一个信号量，因此，信号量本身就是一种临界资源。信号量是用来保护临界资源的，不应该让别人来保护它，这也就要求信号量的操作（++和--）必须是原子的。如果信号量的值是1，这不就是一把锁吗？

看到这里，相信你已经对信号量有一个初步的认识了，下面我们来看看信号量的接口有哪些。

信号量的接口

信号量有很多版本，常见的有 POSIX信号量、system V信号量，我们以POSIX信号量为例，因为它操作上比较简单，使用POSIX信号量的接口 需要包含头文件**<semaphore.h>。**

初始化信号量

初始化信号量的函数为sem_init。

功能：用于初始化一个信号量。

函数原型：int sem_init(sem_t *sem, int pshared, unsigned int value)

参数：

• sem：指向要初始化的信号量对象的指针，信号量对象通常是一个sem_t类型的变量。
• pshared：指定信号量的共享范围：
• 如果pshared != 0，信号量是进程间共享的，此时信号量必须位于共享内存区域。
• 如果pshared == 0，信号量是线程间共享的（默认行为）。
• value：指定信号量的初始值，通常是一个非负整数，表示初始可用资源的数量。

返回值：

• 成功时返回0。
• 失败时返回-1，并设置errno以指示错误类型。

等待信号量

等待信号量其实是申请信号量，使用sem_wait函数。

功能：用于对信号量执行 P 操作（也称为 wait 操作）。它的主要作用是尝试获取信号量，如果信号量的值大于 0，则将其减 1；如果信号量的值为 0，则调用线程或进程会被阻塞，直到信号量的值变为大于 0。

函数原型：int sem_wait(sem_t *sem)

参数：

• sem：指向信号量对象的指针。

返回值：

• 成功时返回0。
• 失败时返回-1，并设置errno以指示错误类型

发布信号量

发布信号量其实是释放自己所持有的信号量，使用sem_post函数。

功能：用于对信号量执行 V 操作（也称为 signal 操作）。它的主要作用是释放信号量，将信号量的值加 1。如果有线程或进程正在等待该信号量（即调用了sem_wait并被阻塞），sem_post会唤醒其中一个等待的线程或进程。

函数原型：int sem_post(sem_t *sem)

参数：

• sem：指向信号量对象的指针。

返回值：

• 成功时返回0。
• 失败时返回-1，并设置errno以指示错误类型。

销毁信号量

销毁信号量使用sem_destroy函数。

功能：用于销毁一个信号量。

函数原型：int sem_destroy(sem_t *sem)

参数：

• sem：指向要销毁的信号量对象的指针。

返回值：

• 成功时返回0。
• 失败时返回-1，并设置errno以指示错误类型。

我们已经了解了信号量的相关接口，下面我们使用信号量来实现一个基于环形队列的生产者消费者模型。

3. 实现基于环形队列的生产者消费者模型

在阅读这部分内容时，我建议你阅读一下这篇博客 —— 基于阻塞队列的生产者消费者模型，这里面讲解了如何实现一个基于阻塞队列的生产者消费者模型；因为实现环形队列的生产者消费者模型 和 实现 阻塞队列的生产者消费者模型 大致思想是一样的的，只是数据交易的产所不同，线程之间同步的方式不同，并且，博主我会引用这里面的一些代码。

环形队列

对于环形队列，我们可以使用数组来模拟，每次对下标进行取余操作；当下标到达最后一个位置的下一个位置，取余能够让其回到开头位置，从而模拟环形队列。

在环形队列中，有两种资源，一种是数据资源，一种是空间资源；对于生产者来说，最关心的就是空间资源，只要申请空间资源成功，就一定有空间，就一定能够生产；对于消费者来说，最关心的就是数据资源，只要申请数据资源成功，就一定有数据，就一定能够获取数据。因此，我们可以使用两个信号量来表示这两种资源。

在生产者生产数据的时候，需要申请的是空间资源，也就是自己关系的资源，当生产完成的时候，意味着空间资源减少了一个，数据资源增加了一个，所以，生产者P的是自己关心的资源，V的是对方关心的资源；反过来对于消费者也是一样的。

需要注意的是，生产者和消费者的超时下标是相同的，并且都是按照从左到右的顺序访问 数组模拟的环形队列，不会说 随机访问环形队列的任意一个位置。

基于环形队列的生产者消费者模型的实现

环形队列代码

#pragma once
#include <iostream>
#include <string>
#include <vector>
#include <semaphore.h>
#include <pthread.h>
#define QUEUESIZE 5
template<typename T>
class RingQueue
{
private:
    // 1. 环形队列
    std::vector<T> _ring_queue;
    int _cap; // 环形队列的容量上限
    // 2. 生产和消费的下标
    int _productor_step;
    int _consumer_step;
    // 3. 定义信号量
    sem_t _room_sem; // 生产者关心
    sem_t _data_sem; // 消费者关心
    // 4. 定义锁，维护多生产多消费之间的互斥关系
    pthread_mutex_t _productor_mutex;
    pthread_mutex_t _consumer_mutex;
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 = QUEUESIZE)
        : _ring_queue(cap), _cap(cap),  _productor_step(0), _consumer_step(0)
    {
        // 构造的时候完成信号量和互斥量的初始化
        sem_init(&_room_sem, 0, _cap);
        sem_init(&_data_sem, 0, 0);
        pthread_mutex_init(&_productor_mutex, nullptr);
        pthread_mutex_init(&_consumer_mutex, nullptr);
    }
    void PushData(const T &data)
    {
        // 申请空间资源
        P(_room_sem);
        // 申请锁，互斥的访问环形队列
        Lock(_productor_mutex);
        // 代码走到这里，意味着申请空间资源成功，一定有空间可以用
        // 往队列中放数据
        _ring_queue[_productor_step++] = data;
        _productor_step %= _cap; // 不要忘了取余来模拟环形队列
        // 释放锁
        Unlock(_productor_mutex);
        // 将对方关心的资源进行V操作
        V(_data_sem);
    }
    void TakeData(T& out)
    {
        // 申请数据资源
        P(_data_sem);
        // 申请锁，互斥的访问环形队列
        Lock(_consumer_mutex);
        // 代码走到这里，意味着申请数据资源成功，一定有数据可以用
        // 从队列中获取数据
        out = _ring_queue[_consumer_step++];
        _consumer_step %= _cap;
        // 解锁
        Unlock(_consumer_mutex);
        // 将对方关心的资源进行V操作
        V(_room_sem);
    }
    ~RingQueue()
    {
        // 析构的时候销毁信号量和互斥锁
        sem_destroy(&_room_sem);
        sem_destroy(&_data_sem);
        pthread_mutex_destroy(&_productor_mutex);
        pthread_mutex_destroy(&_consumer_mutex);
    }
};

任务类型代码

未来，我们在队列中放的是一个个的任务类型的数据，消费者取出任务后可以执行任务，我们编写的任务只是简单的计算两个数字的和，读者可以自行编写一些自己喜欢的任务。

class Task
{
public:
    Task() {}
    Task(int num1, int num2) : _num1(num1), _num2(num2), _result(0)
    {
    }
    void Excute()
    {
        _result = _num1 + _num2;
    }
    std::string ResultToString() // 打印计算结果
    {
        return std::to_string(_num1) + "+" + std::to_string(_num2) + "=" + std::to_string(_result);
    }
    std::string DebugToString()  // 打印生产的任务
    {
        return std::to_string(_num1) + "+" + std::to_string(_num2) + "=?";
    }
    ~Task()
    {}
private:
    int _num1;
    int _num2;
    int _result;
};

主程序代码

#include <iostream>
#include <vector>
#include <unistd.h>
#include "RingQueue.hpp"
#include "task.hpp"
#define PRODUCTOR_NUM  1  // 定义生产者线程的默认线程数
#define CONSUMER_NUM  1   // 定义消费者线程的默认线程数
// 消费者线程执行的代码
void *consumer(void *arg)
{
    RingQueue<Task> *bqp = (RingQueue<Task>*)arg;
    Task task; // 输出型参数
    while(true){
        bqp->TakeData(task); // 获取数据
        task.Excute();       // 消费数据
        std::cout << "consumer task: " << task.ResultToString() << std::endl;
        sleep(1);
    }
}
// 生产者线程执行的代码
void *producter(void *arg)
{
    RingQueue<Task> *bqp = (RingQueue<Task>*)arg;
    srand((unsigned long)time(NULL));
    while(true){
        // 生产数据
        int num1 = rand() % 1000; 
        int num2 = rand() % 1000; 
        Task task(num1,num2);
        // 放入数据
        bqp->PushData(task);
        std::cout << "product task: " << task.DebugToString() << std::endl;
        sleep(1);
    }
}
int main()
{
    RingQueue<Task> bq;          // 创建一个阻塞队列
    std::vector<pthread_t> tids; // 存储所有线程的id，用于等待所有线程
    // 创建消费者线程
    for(int i = 0; i < CONSUMER_NUM; ++i)
    {
        pthread_t id = 0;
        // 创建消费者线程的时候，把阻塞队列传进去，让生产者和消费者能够看到同一个阻塞队列
        pthread_create(&id, NULL, consumer, (void*)&bq);
        tids.push_back(id);
    }
    
    // 创建生产者线程
    for(int i = 0; i < PRODUCTOR_NUM; ++i)
    {
        pthread_t id = 0;
        // 创建生产者线程的时候，把阻塞队列传进去，让生产者和消费者能够看到同一个阻塞队列
        pthread_create(&id, NULL, producter, (void*)&bq);
        tids.push_back(id);
    }
    // 等待所有线程
    for(int i = 0; i < tids.size(); ++i)
    {
        pthread_join(tids[i], NULL);
    }
    return 0;
}

运行结果：

• 我们可以看到生产者生产的任务被消费者拿到并执行了。

本文原文来自CSDN