1.生产消费模型补充

1.1对Cond进行封装

#include <iostream>
#include "Mutex.hpp"
#include <pthread.h>

using namespace LockModule;

namespace CondModule
{
   class Cond
{
public:
    Cond()
    {
      int n=::pthread_cond_init(&_cond,nullptr);
      (void)n;
    }
    void wait(Mutex &mutex)
    {
       int n =::pthread_cond_wait(&_cond,mutex.LockPtr());
       (void)n;
    }
    void Notify()
    {
        int n =::pthread_cond_signal(&_cond);
        (void)n;
   
    }
    void NotifyAll()
    {
        int n =::pthread_cond_broadcast(&_cond);
        (void)n;
    }
    ~Cond()
    {
       pthread_cond_destroy(&_cond);
    }
private:
    pthread_cond_t _cond;
};
}

而生产者消费者模型交易场所不仅仅可以交易数据，还可以交易任务；

1.2 将生产者消费者模型改为基于任务队列的模型

Blockqueue.hpp

#pragma once

#include <iostream>
#include <queue>
#include <unistd.h>
#include <pthread.h>
#include "Cond.hpp"
#include "Mutex.hpp"

namespace BlockQueueModule
{
    static const int gcap=10;
    using namespace LockModule;
    using namespace CondModule;
    template <typename T>
    class BlockQueue
    {
    public:
        bool IsFull() { return _cap == _q.size(); }
        bool IsEmpty() { return _q.empty(); }

        BlockQueue(int cap = gcap)
            : _cap(cap), _cwait_num(0), _pwait_num(0)
        {
           
        }
        void Equeue(const T &in)
        {
        
           Lockguard lockguard(_mutex);
            if (IsFull())
            {
                 _pwait_num++;
                std::cout << "生产者进入等待..." << std::endl;
                _productor_cond.wait(_mutex);
                _pwait_num--;
                std::cout << "生产者被唤醒..." << std::endl;
            }
                _q.push(in);
                _pwait_num++;
                if (_pwait_num)
                {
                    //std::cout << "消费者被唤醒..." << std::endl;
                    _consumer_cond.Notify();
                    _pwait_num--;
                }
          
        
        }
        void Pop(T *out)
        {
            Lockguard lockguard(_mutex);
            if (IsEmpty())
            {
                _cwait_num++;
                std::cout << "消费者进入等待..." << std::endl;
                _consumer_cond.wait(_mutex);
                _cwait_num--;
                std::cout << "消费者被唤醒..." << std::endl;
                
            }
                *out = _q.front();
                std::cout << "消费者被唤醒..." << std::endl;
                _q.pop();
                _cwait_num++;
                if (_cwait_num)
                {
                    _productor_cond.Notify();
                    _cwait_num--;
                }
            
            
        }
        ~BlockQueue()
        {
           
        }

    private:
        std::queue<T> _q;
        int _cap;
        Mutex _mutex;
        Cond _productor_cond;
        Cond _consumer_cond;
        int _cwait_num;
        int _pwait_num;
    };
}

Task.hpp

#include <iostream>
#include <unistd.h>
namespace TaskModule
{
 class Task
{
  public:
   Task()
   {
      
   }
   Task(int a,int b)
   :x(a)
   ,y(b)
   {

   }
   void Excute()
   {
      sleep(1);
      result= x+y;
   }
   int Result()
   {
       return result;
   }
   int X()
   {
        return x;
   }
   int Y()
   {
        return y;
   }
   ~Task()
   {

   }
   private:
     int x;
     int y;
     int result;
};

}

Main.cpp

#include "BlockQueue.hpp"
#include "Task.hpp"
#include <ctime>
using namespace BlockQueueModule;
using namespace TaskModule;


void *Consumer(void * args)
{
    BlockQueue<Task> * bq = static_cast<BlockQueue<Task>*> (args);
    while(true)
    {
        Task t;
        //1.从bq中拿数据
        bq->Pop(&t);
        //2.做处理
        t.Excute();
        printf("consumer ,处理完成一个任务,result:%d+%d=%d\n",t.X(),t.Y(),t.Result());
         
    }

}

void * Productor(void * args)
{
    BlockQueue<Task>* bq = static_cast<BlockQueue<Task>*>(args);
    int data=10;
    while(true)
    {
        //1.从外部获取数据
        int x=rand()%10+1; //[1.10)
        int y=rand()%100+1;  //[1,100)
        //2.构建任务
        Task t(x,y);
        bq->Equeue(t);
        printf("productor ,创建了一个任务:%d+%d = ?\n",t.X(),t.Y());
        data++;
    }
}

int main()
{
    srand(time(nullptr)^getpid());
   BlockQueue<Task> * bq = new BlockQueue<Task>(5);
   pthread_t c,p;
   pthread_create(&c,nullptr,Consumer,bq);
   pthread_create(&p,nullptr,Productor,bq);

   pthread_join(c,nullptr);
   pthread_join(p,nullptr);

   delete bq;
   return 0;
}

当然我们这个类也可以直接使用仿函数；

1.3 基于生产者消费者模型阻塞队列的应用场景

应用场景生产者消费者队列 / 缓冲区实际作用
线程池任务执行提交任务的业务线程线程池里的工作线程任务队列削峰、异步、控制并发量
日志系统业务代码（打日志）后台日志写入线程日志队列不阻塞主线程，提高性能
消息队列 MQ消息发送方消息消费方MQ 队列系统解耦、异步、流量削峰
网络 IO 数据处理网络接收线程业务解析线程数据缓冲队列防止粘包、提升并发处理能力
UI 界面刷新后台数据线程UI 主线程事件 / 消息队列避免 UI 卡顿，线程安全更新
游戏帧更新逻辑帧生成渲染 / 物理线程指令队列稳定帧率，解耦逻辑与渲染
爬虫 / 数据采集下载 / 抓取线程解析 / 存储线程URL / 数据队列控制爬取速度，防止被封
音视频播放解码线程播放渲染线程音帧队列平滑播放，抗抖动
订单系统下单请求库存 / 支付 / 物流线程订单队列高并发下保证订单不丢失
大数据处理数据生成 / 采集计算 / 分析线程数据队列流式处理，异步计算

1.4生产消费模型的补充知识

①生产者的数据是哪里来的？消费者只是把数据拿走了吗？不做其他处理了吗？

从网络和用户键盘中获取，而在生产者等待数据的时候，在阻塞队列中有残留的任务，而消费者这个时候可以从队列中拿任务，并且处理任务，这样生产者和消费者这两个线程不就并行起来了吗？这就是为什么这个模型效率高的原因；也就是说生产者获取数据生产任务的时候，消费者也可以在队列中拿数据并且处理任务；所以这个模型的高效不仅是只有在交易场所中并行的效率高；

1.5基于环形队列的生产消费模型

让tail代表生产者，head代表消费者，tail先走，生产一个数据，消费一个空间；head后走，消费一个数据，产生一个空间，为空的时候，保证tail(生产者)先走，原子性生产，为满的时候保证head(消费者)先走，原子性的先消费；这样为空为满的时候就表现出了线程的同步互斥，而不为空不为满的时候线程间就是并发的；

我们这个环形队列的生产消费模型的信号量应该有几个呢？首先信号量的数目表示资源个数，而资源又分为数据和空间，所以我们今天使用两个信号量，一个是sem_t data ,一个是sem_t space；

2.posix 信号量

2.1什么是信号量

信号量就是一个带计数的锁，用来控制最多同时有多少个线程能进入某段代码(共享资源)。互斥锁（mutex）：只能 1 个线程进，同一时刻只允许 一个线程 进入临界区，这样的信号量就叫做互斥锁，也叫做也叫做二元信号量；信号量（semaphore）：可以设置 N 个线程 同时进入临界区；
申请信号量的本质：申请一份 “许可”，让线程可以继续往下执行；没有许可就阻塞等待。

“资源是否准备成功 == 申请信号量是否成功”  -> 非常重要！！！！

2.2 posix信号量是怎么使用

①初始化信号量

② 销毁信号量

③等待信号量

④发布信号量

2.3posix信号量和system V信号量的区别是什么

特性System V 信号量POSIX 信号量
标准来源老 Unix System V IPC现代 IEEE POSIX 标准
基本单位信号量集合（数组）一个 semid 包含多个信号量单个信号量（sem_t）一次只操作一个
标识方式key_t 键值 + semid（整数 ID）有名：文件名（如 /sem_name）无名：内存地址（sem_t*）
API 风格复杂、步骤多、无下划线semget / semop / semctl简单、直观、带下划线sem_init/wait/post/destroy
P/V 操作用 semop + struct sembuf 结构体支持一次操作多个信号量、任意加减值sem_wait()（-1）、sem_post()（+1）只能单个、固定 ±1
类型只有一种（进程间）分两种：• 无名：线程间（内存）• 有名：进程间（文件）
生命周期内核持久进程退出还在，必须 semctl(IPC_RMID) 删无名：随进程销毁有名：手动 sem_unlink
性能较重、全是系统调用轻量、无名几乎无内核开销
超时不支持（要自己封装）支持 sem_timedwait
非阻塞无直接函数（要自己判断）有 sem_trywait
适用场景复杂多资源同步、老项目、跨进程复杂协作线程同步、简单进程同步、新项目、高性能场景

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

所以对于生产者和消费者竞争进程而言，他们都需要一把琐；

Main.cc  -> 定义进程实现测试

#include "Mutex.hpp"
#include "RingBuffer.hpp"
#include <ctime>


using namespace RingBufferModule;

void *Consumer(void * args)
{
    RingBuffer<int> * rb= static_cast<RingBuffer<int> *> (args);
    int data=0;
    while(true)
    {
        sleep(1);
        int data;
        rb->pop(&data);
        std::cout << "consumer 消费了一个数据："<< data<<std::endl;
        data++;
    }

}


void * Productor(void * args)
{
    RingBuffer<int> * rb = static_cast<RingBuffer<int> * >(args);
    int data=0;
    while(true)
    {  
        rb->Equeue(data);
        std::cout << "productor 生产了一个数据："<< data <<std::endl;
        data++;
    }
}

int main()
{

   RingBuffer<int> * rb = new RingBuffer<int> (5);
   pthread_t c1,c2,c3;
   pthread_t p1,p2;
   pthread_create(&c1,nullptr,Consumer,rb);
   pthread_create(&c2,nullptr,Productor,rb);
   pthread_create(&c3,nullptr,Productor,rb);
   pthread_create(&p1,nullptr,Productor,rb);
   pthread_create(&p2,nullptr,Productor,rb);

   pthread_join(c1,nullptr);
   pthread_join(c2,nullptr);
   pthread_join(c3,nullptr);
   pthread_join(p1,nullptr);
   pthread_join(p2,nullptr);

   delete rb;
   return 0;
}

RingBuffer.hpp -> 实现环形队列

#include <iostream>
#include <unistd.h>
#include <vector>
#include "Sem.hpp"
#include "Mutex.hpp"

namespace RingBufferModule
{
    using namespace SemModule;
    using namespace LockModule;
    static const int gcap = 10;
    template <typename T>
    class RingBuffer
    {
    public:
        RingBuffer(int cap)
            :_ring(cap)
            ,_p_step(0)
            ,_c_step(0)
            ,_cap(cap)
            ,_datasem(0)
            ,_spacesem(cap)
             
        {}

        void Equeue(const T & in)
        {
            _spacesem.P();
            {
                Lockguard lockguard(_p_lock);
                _ring[_p_step] = in;
                _p_step++;
                _p_step %= _cap;
            }
            _datasem.V();
        }
        void pop(T*out)
        {
            _datasem.P();
            {
                Lockguard lockguard(_c_lock);
                *out = _ring[_c_step];
                _c_step++;
                _c_step %= _cap;  //维持环形特性 
            }
            _spacesem.V();
        }

        ~RingBuffer()
        {
        }

    private:
        std::vector<T> _ring;
        int _cap;
        int _p_step;
        int _c_step;
        Sem _datasem;      //数据信号量
        Sem _spacesem;     //空间信号量
        Mutex _p_lock;
        Mutex _c_lock;
    };

}

Sem.hpp   -> 对底层的sem进行分装

#pragma once 
#include <iostream>
#include <pthread.h>
#include <semaphore.h>

namespace SemModule
{
 class Sem
{
  static const int defaultvalue =1;
  public:
      Sem(int value=defaultvalue):_init_value(value)
      {
        int n =sem_init(&_sem,0,value);
        (void)n;
      }
      void P()   //自动对信号量做--
      {
         int n=sem_wait(&_sem);
         (void)n;
      }
      void V ()   //自动对信号量做++，对一个信号量做解锁之后，我们的信号量数目就会增多
      {
          int n = sem_post(&_sem);
          (void)n;
      }
      ~Sem()
      {
           int n=sem_destroy(&_sem);
           (void)n;
      }
    private:
       sem_t _sem;
       int _init_value;
};
}

    现象：

                            

                                                                            

# linux # c++
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

玩转 AtomCode！AtomGit「码动四季・开源同行」夏季征稿获奖名单出炉！

avatar AtomGit开源社区
cover

开源鸿蒙跨平台直播｜napigen 开源鸿蒙跨端实践：打通 Flutter 与 ArkTS

avatar AtomGit开源社区
cover

榜单出炉！AtomGit「源启盛夏」暑期开源成长计划正式收官

avatar AtomGit开源社区