Linux基于阻塞队列的生产消费者模型详解

2025-04-28 09:35 运维作者： s_little_monster_

一、什么是生产消费者模型

生产消费者模型就是通过一个容器来解决生产者和消费者的强耦合问题，生产者和消费者彼此之间不直接通讯，而是通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接交给阻塞队列，消费者不找生产者索要数据，而是直接从阻塞队列中去取，这样一来，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力

对于生产消费者模型，我们有一个321规则，分别是3种关系，2种角色，1个交易场所

三种关系：生产者和生产者的互斥竞争关系，消费者和消费者的互斥竞争关系，生产者和消费者的互斥、同步关系
两种角色：生产者和消费者
一个交易场所：特定结构的内存空间（如阻塞队列）

二、基于阻塞队列的生产消费者模型

1、理论研究

Linux基于阻塞队列的生产消费者模型详解

在多线程编程中，阻塞队列是一种常用于实现生产者和消费者模型的数据结构，其与普通的队列区别在于，当队列为空时，从队列获取元素的操作将会被阻塞，直到队列中再次被放入元素，当队列满时，往队列中存放元素的操作也会被阻塞，直到有元素从队列中被获取

生产消费者模型最大的好处是，也是生产消费者模型效率高的原因是：在消费者获取数据（一般是网络数据）的时候，生产者可以生产数据，生产者放入数据的时候，消费者可以处理数据，虽然特定内存结构，也就是临界资源区是有锁的，只能由单线程通过，只要将时间合理化，我们就可以实现生产者和消费者的高效率工作，并且将发送数据的线程和处理数据的线程解耦合

Linux基于阻塞队列的生产消费者模型详解

2、多生产多消费模型

（一）BlockQueue.hpp

#pragma once

#include <IOStream>
#include <queue>
#include <pthread.h>
//定义一个模版类，方便我们android使用任何类型进行生产消费
template <class T>
//定义一个阻塞队列
class BlockQueue
{
	//队列默认最大容量
    static const int defalutnum = 20;

public:
    BlockQueue(int maxcap = defalutnum) : maxcap_(maxcap)
    {
    	//初始化互斥锁和生产者和消费者的条件变量
        pthread_mutex_init(&mutex_, nullptr);
        pthread_cond_init(&c_conwww.devze.comd_, nullptr);
        pthread_cond_init(&p_cond_, nullptr);
        //下面注释掉的是设置水位线，设置最低最高水位线
        //在阻塞队列中的数据，在低于最低水位线时是不可被获取的，只能写入
        //在高于最高水位线时是不可被写入的，只能获取
        // low_water_ = maxcap_/3;
        // high_water_ = (maxcap_*2)/3;
    }

	//从队列头取出元素返回
    T pop()
    {
        pthread_mutex_lock(&mutex_);//加锁
        //只能用while不能用if，原因是会出现误唤醒问题，下面说
        while (q_.size() == 0)
        {
            pthread_cond_wait(&c_cond_, &mutex_); 
        }

        T out = q_.front();
        q_.pop();
		//这里是加了水位线的版本，在低于水位线的时候要唤醒生产者
        // if(q_.size()<low_water_) pthread_cond_signal(&p_cond_);
        pthread_cond_signal(&p_cond_); 
        pthread_mutex_unlock(&mutex_);//解锁

        return out;
    }

    void push(const T &in)
    {
        pthread_mutex_lock(&mutex_);//加锁
        //同pop函数
        while (q_.size() == maxcap_)
        {
            pthread_cond_wait(&p_cond_, &mutex_); 
        }
        q_.push(in); 
        
        //这里是加了水位线的版本，在高于水位线的时候要唤醒消费者
        // if(q_.size() > high_water_) pthread_cond_signal(&c_cond_);
        pthread_cond_signal(&c_cond_);
        pthread_mutex_unlock(&mutex_);//解锁
    }
	//析构函数
    ~BlockQueue()
    {
        pthread_mutex_destroy(&mutex_);
        pthread_cond_destroy(&c_cond_);
        pthread_cond_destroy(&p_cond_);
    }

private:
    std::queue<T> q_; 
    
    int maxcap_; // 极大值
    
    pthread_mutex_t mutex_;
    pthread_cond_t c_cond_;
    pthread_cond_t p_cond_;
    
	//最低最高水位线
    // int low_water_;
    // int high_water_;
};

（二）Task.hpp

#pragma once
#include <iostream>
#include <string>

//定义运算方法
std::string opers = "+-*/%";

//枚举错误
enum
{
    DivZero = 1,
    Modzero,
    Unknown
};

class Task
{
public:
    Task(int x, int y, char op) : data1_(x), data2_(y), oper_(op), result_(0), exitcode_(0)
    {}
    void run()
    {
        switch (oper_)
        {
        case '+':
            result_ = data1_ + data2_;
            break;
        case '-':
            result_ = data1_ - data2_;
            break;
        case '*':
            result_ = data1_ * data2_;
            break;
        case '/':
        {
            if (data2_ == 0)
                exitcode_ = DivZero;
            else
                result_ = data1_ / data2_;
        }
        break;
        case '%':
        {
            if (data2_ == 0)
                exitcode_ = ModZero;
            else
                result_ = data1_ % data2_;
        }
        break;
        default:
            exitcode_ = Unknown;
            break;
        }
    }
    //伪函数，通过重载()使run可以像函数一样调用
    void operator()()
    {
        run();
    }
    //返回的运算结果以及错误代码
    std::string GetResult()
    {
        std::string r = std::to_string(data1_);
        r += oper_;
        r += std::to_string(data2_);
        r += "=";
        r += std::to_string(result_);
        r += "[code: ";
        r += std::to_string(exitcode_);
        r += "]";

        return r;
    }
    //返回运算表达式
    std::string GetTask()
    {
        std::string r = std::to_string(data1_);
        r += oper_;
        r += std::to_string(data2_);
        r += "=?";
        return r;
    }
    ~Task()
    {}

private:
    int data1_;
    int data2_;
    char oper_;

    int result_;
    int exitcode_;
};

（三）main.cpp

#include "BlockQueue.hpp"
#include "Task.hpp"
#include <unistd.h>
#include <ctime>

void *Consumer(void *args)
{
    BlockQueue<Task> *bq = static_cast<BlockQueue<Task> *>(args);

    while (true)
    {
        // 消费
        Task t = bq->pop();
        // 计算
        t();
        
		//模拟消费者处理任务
        std::cout << "处理任务: " << t.GetTask() << " 运算结果是： " << t.GetResult() << " thread id: " << pthread_self() << std::endl;
    }
}

void *Productor(void *args)
{
    int len = opers.size();
    BlockQueue<Task> *bq = static_cast<BlockQueue<Task> *>(args);
    int x = 10;
    int y = 20;
    while (true)
    {
        // 用随机数运算模拟生产者生产数据
        int data1 = rand() % 10 + 1; // [1,10]
        usleep(10);
        int data2 = rand() % 10;
        char op = opers[rand() % len];
        Task t(data1, data2, op);

        // 生产
        bq->push(t);
        std::cout << "生产了一个任务: " << t.GetTask() << " thread id: " << pthread_self() << std::endl;
        sleep(1);
    }
}

int main()
{
	//随机数种子
    srand(time(nullptr));

    //给阻塞队列传一个任务
    BlockQueue<Task> *bq = new BlockQueue<Task>();
    //多生产者多消费者
    pthread_t c[3], p[5];
    for (int i = 0; i < 3; i++)
    {
        pthread_create(c + i, nullptr, Consumer, bq);
    }

    for (int i = 0; i < 5; i++)
    {
        pthread_create(p + i, nullptr, Productor, bq);
    }

    for (int i = 0; i < 3; i++)
    {
        pthread_join(c[i], nullptr);
    }
    for (int i = 0; i < 5; i++)
    {
        pthread_join(p[i], nullptr);
    }
    delete bq;
    return 0;
}

Linux基于阻塞队列的生产消费者模型详解

3、误唤醒问题

误唤醒问题就是在调用pop函数或者push函数的时候可能会引起的，下面我们再把代码贴出来，然后把上面有过的注释去掉

//...

	T pop()
    {
        pthread_mutex_lock(&mutex_);
        while (q_.size() == 0) //不能调用if而要用while
        {
            pthread_cond_wait(&c_cond_, &mutex_); 
        }

        T out = q_.front();
        q_.pop();

        pthread_cond_signal(&p_cond_); 
        pthread_mutex_unlock(&mutex_);

    javascript    return out;
    }

    void push(const T &in)
    {
        pthread_mutex_lock(&mutex_);
        whilepython (q_.size() == maxcap_)
        {
            pthread_cond_wait(&p_cond_, &mutex_); 
        }
        q_.push(in); 
        
        pthread_cond_signal(&c_cond_);
        pthread_mutex_unlock(&mutex_);
        
//...

在多生产者 - 多消费者并发编程场景中，误唤醒现象较为常见，假定队列当前处于满状态，当一个消费者线程成功消费一个数据后，队列中会空出一个位置，随后，线程可能多次调用pthread_cond_signal(&p_cond_) 函数，唤醒了一批正在 p_cond_ 条件变量下等待的生产者线程，由于被唤醒的生产者线程需要重新竞争互斥锁，这些线程之间呈现出互斥关系，在先前执行消费操作的线程释放锁之后，仅有一个生产者线程能够成功获取锁，其余虽被唤醒但未能抢到锁的生产者线程只能在锁处等待

当成功获取锁的生产者线程完成数据生产操作后，队列可能再次达到满状态，此时，该线程会调用 pthread_cond_signal(&c_cond_) 函数唤醒一个消费者线程，随后释放锁，在此情形下，被唤醒的线程不仅包括刚刚被唤醒的消费者线程，还涵盖之前被唤醒却未抢到锁的生产者线程，它们会同时参与锁的竞争，若使用 if 语句来判断队列是否已满，当某个生产者线程抢到锁后，可能不会再次对队列状态进行检查，直接尝试向已满的队列中添加数据，从而引发错误

因此，为确保线程安全，应使用 while 循环来包裹 pthread_cond_wait 函数，当一个线程被唤醒并成功获取锁后，不应直接执行队列操作（无论是生产数据还是消费数据），而应再次检查资源是否满足操作条件，若资源就绪，则可继续执行队列操作；若资源未就绪，则应再次调用 pthread_cond_wait 函数，使线程进入休眠状态，等待后续唤醒

总结

以上为个人经验，希望能给大家一个参考，也希望大家多多支持编程客栈(www.devze.com)。

继续阅读：Linux生产消费者 Linux生产消费者模型 Linux阻塞队列

Linux基于阻塞队列的生产消费者模型详解

目录

一、什么是生产消费者模型

二、基于阻塞队列的生产消费者模型

1、理论研究

2、多生产多消费模型

（一）BlockQueue.hpp

（二）Task.hpp

（三）main.cpp

3、误唤醒问题

总结

更多精彩内容

精彩评论

最新运维

Win11迎来Recall等功能! Win 11 AI+ PC 24H2 四月可选更新三大特性介绍

win10设置文件夹怎么使用共享向导?

Windows11启动声音怎么设置? Win11开机声音的设置方法

win10怎么使用触屏? win10平板模式触屏不能用的解决办法

任务栏缩略图怎么设置? Windows系统任务栏缩略图的设置方法

运维排行榜

6款自己搭建开源VPN的工具

UiPath Orchestrator 平台搭建全过程（Windows环境）

常见以太网端口协商不成功或不匹配问题总结

df空间占用已满，但du查看home目录空间占用却很少

Windows Server 2016 Standard Key激活密钥序列号