【Linux线程】Linux多线程实践：深入生产者消费者模型-linux运维-PHP中文网

【Linux线程】Linux多线程实践：深入生产者消费者模型

爱谁谁

发布： 2025-06-20 08:50:16

原创

283人浏览过

?前言：在当今的软件开发领域，多线程编程已经成为了一种不可或缺的技术。特别是在linux 操作系统下，多线程编程的应用更是广泛而深入。而在多线程编程中，生产者消费者模型无疑是一个经典且重要的并发编程模式

本文旨在为读者提供一个全面而深入的Linux多线程中生产者消费者模型的学习指南。我们将从模型的基本概念出发，逐步深入到Linux多线程编程的实战技巧。通过详细的代码示例和深入的解析，我们将帮助读者掌握如何在Linux多线程环境下实现高效且稳定的生产者消费者模型

?1. 生产者消费者模型

作用：

生产者和消费者之间的关系：

生产者消费者模型遵守 “321原则”：

生产者消费者模型优点：

解耦支持并发支持忙闲不均

?2. 基于BlockingQueue的生产者消费者模型

代码示例：(LockGuard.hpp)

代码语言：javascript代码运行次数：0运行复制

#pragma once#include <iostream>#include <pthread.h>#include <unistd.h>#include <string>#include <vector>#include <functional>using namespace std;class Mutex // //封装pthread库的锁{public:    Mutex(pthread_mutex_t *lock)        :_lock(lock)    {}    void lock()    {        pthread_mutex_lock(_lock);    }    void unlock()    {        pthread_mutex_unlock(_lock);    }    ~Mutex()    {}private:    pthread_mutex_t *_lock;};class LockGuard // 封装了锁后,出了作用域会自动调用析构函数，用来自动解锁{public:    LockGuard(pthread_mutex_t *lock)        :_mutex(lock)    {        _mutex.lock();    }    ~LockGuard()    {        _mutex.unlock();    }private:    Mutex _mutex;};

登录后复制

代码示例：(BlockQueue.hpp)

代码语言：javascript代码运行次数：0运行复制

#include <iostream>#include <pthread.h>#include <queue>#include <ctime>#include <unistd.h>#include <sys/types.h>#include "LockGuard.hpp"using namespace std;const int defaultcap = 5;  // for test 最大容量template <class T>class BlockQueue{public:    BlockQueue(int cap = defaultcap)        :_capacity(cap)    {        pthread_mutex_init(&_mutex, nullptr);        pthread_cond_init(&_p_cond, nullptr);        pthread_cond_init(&_c_cond, nullptr);    }    bool IsFull()    {        return _q.size() == _capacity;    }    bool IsEmpty()    {        return _q.size() == 0;    }    bool Push(const T &in) // 生产者    {        LockGuard lockguard(&_mutex);        // pthread_mutex_lock(&_mutex);        // if(IsFull())        while(IsFull())        {            // 阻塞等待            pthread_cond_wait(&_p_cond, &_mutex);        }        _q.push(in);        // if(_q.size() > _productor_water_line) pthread_cond_signal(&_c_cond);        pthread_cond_signal(&_c_cond);        // pthread_mutex_unlock(&_mutex);        return true;    }    bool Pop(T *out) // 消费者    {        LockGuard lockguard(&_mutex);        // pthread_mutex_lock(&_mutex);        // if(IsEmpty())        while(IsEmpty())        {            // 阻塞等待            pthread_cond_wait(&_c_cond, &_mutex);        }        *out = _q.front();        _q.pop();        pthread_cond_signal(&_p_cond);        // pthread_mutex_unlock(&_mutex);        return true;    }    ~BlockQueue()    {        pthread_mutex_destroy(&_mutex);        pthread_cond_destroy(&_p_cond);        pthread_cond_destroy(&_c_cond);    }private:    queue<T> _q;    int _capacity; // 列表的容量，满了就不能生存，空了就不能消费    pthread_mutex_t _mutex; // 生产者消费者共用的一把锁    pthread_cond_t _p_cond; // 生产者    pthread_cond_t _c_cond; // 消费者    int _productor_water_line; // _productor_water_line == _capacity / 3 * 2;    int _consumer_water_line; // _consumer_water_line == _capacity / 3;};

登录后复制

关于BlockQueue.hpp的代码，还是比较好理解的，我们唯一要注意的是我们生产者和消费者在进行阻塞等待时，在检测临界资源时，我们要尽可能使用while，而不是if，当我们唤醒资源时，可能会被唤醒多个等待的资源，而对应条件的判断，但是当锁被申请时，其他资源又将继续申请锁，此时的条件很都可能是不满足的，所以这可能会导致代码出错，我们也把这种情况称为：伪唤醒

阻塞式的任务队列

?3. POSIX信号量信号量本质是一把计数器申请信号本质就是预定资源POSIX信号量的相关函数

初始化信号量：

代码语言：javascript代码运行次数：0运行复制

int sem_init(sem_t *sem, int pshared, unsigned int value);

登录后复制

参数：

sem：指向要初始化的信号量对象的指针pshared：指示信号量是否可以在进程间共享。如果 pshared 为 0，信号量将仅在当前进程内的线程间共享。如果 pshared 非零，信号量可以在进程间共享（这通常需要特定的权限和配置）value：信号量的初始值。这个值必须大于或等于 0

销毁信号量：

代码语言：javascript代码运行次数：0运行复制

int sem_destroy(sem_t *sem);

登录后复制

等待信号量：

代码语言：javascript代码运行次数：0运行复制

int sem_wait(sem_t *sem);

登录后复制

sem_wait 是 POSIX 信号量（semaphore）API 中的一个函数，用于对信号量进行“等待”操作，也就是尝试对信号量进行“减1”操作。如果信号量的当前值大于0，sem_wait 会成功地将信号量的值减1，并立即返回。然而，如果信号量的当前值为0，sem_wait 会阻塞调用线程，直到信号量的值变为大于0（通常是通过另一个线程调用 sem_post 来实现的）

发布信号量：

代码语言：javascript代码运行次数：0运行复制

int sem_post(sem_t *sem);

登录后复制

sem_post 是 POSIX 信号量（semaphore）API 中的一个函数，用于对信号量进行“发布”或“增加”操作。具体来说，sem_post 会将信号量的值加1，并可能唤醒一个或多个正在等待该信号量的线程（如果它们因为调用 sem_wait 而被阻塞）

?4. 基于环形队列的生产消费模型环形队列

环形结构起始状态和结束状态都是一样的，不好判断为空或者为满，所以可以通过加计数器或者标记位来判断满或者空。另外也可以预留一个空的位置，作为满的状态

实现原理与条件生产者不能将消费者套圈：即生产者不能超越消费者太多，否则会导致数据被覆盖。这可以通过信号量或条件变量来控制生产者的生产速度消费者不能超过生产者：消费者消费的数据量必须小于等于生产者生产的数据量。这可以通过信号量或条件变量来控制消费者的消费速度为空时生产者先运行：当队列为空时，消费者无法从队列中获取数据，因此生产者应该先运行并生产数据为满时消费者先运行：当队列为满时，生产者无法向队列中添加数据，因此消费者应该先运行并消费数据以腾出空间

代码示例：(RingQueue.hpp)

代码语言：javascript代码运行次数：0运行复制

#pragma once#include <iostream>#include <pthread.h>#include <unistd.h>#include <vector>#include <semaphore.h>#include <ctime>#include <sys/types.h>using namespace std;const int defaultsize = 5;template <class T>class RingQueue{private:    void P(sem_t &sem)    {        sem_wait(&sem);    }    void V(sem_t &sem)    {        sem_post(&sem);    }public:    RingQueue(int size = defaultsize)        :_ringqueue(size)        ,_size(size)        ,_p_step(0)        ,_c_step(0)    {        sem_init(&_space_sem, 0, size);        sem_init(&_data_sem, 0, 0);    }    void Push(const T &in)    {        P(_space_sem);        // pthread_mutex_lock(&_p_mutex);        _ringqueue[_p_step] = in;        _p_step++;        _p_step %= _size;        // pthread_mutex_unlock(&_p_mutex);        V(_data_sem);    }    void Pop(T *out)    {        P(_data_sem);        // pthread_mutex_lock(&_c_mutex);        *out = _ringqueue[_c_step];        _c_step++;        _c_step %= _size;        // pthread_mutex_unlock(&_c_mutex);        V(_space_sem);    }    ~RingQueue()    {        sem_destroy(&_space_sem);        sem_destroy(&_data_sem);    }private:    vector<T> _ringqueue;    int _size;    int _p_step;    int _c_step;pthread_mutex_t _p_mutex;pthread_mutex_t _c_mutex;    sem_t _space_sem; // 空间信号    sem_t _data_sem; // 数据信号};

登录后复制

环形式的任务队列