C++解决方法：多线程同步经典案例之生产者消费者问题

生产者消费者问题（英语：PRoducer-consumer problem），也称有限缓冲问题（英语：Bounded-buffer problem），是一个多线程同步问题的经典案例。该问题描述了共享固定大小缓冲区的两个线程——即所谓的“生产者”和“消费者”——在实际运行时会发生的问题。生产者的主要作用是生成一定量的数据放到缓冲区中，然后重复此过程。与此同时，消费者也在缓冲区消耗这些数据。该问题的关键就是要保证生产者不会在缓冲区满时加入数据，消费者也不会在缓冲区中空时消耗数据。

要解决该问题，就必须让生产者在缓冲区满时休眠（要么干脆就放弃数据），等到下次消费者消耗缓冲区中的数据的时候，生产者才能被唤醒，开始往缓冲区添加数据。同样，也可以让消费者在缓冲区空时进入休眠，等到生产者往缓冲区添加数据之后，再唤醒消费者。

本文用一个ITemRepository类表示产品仓库，其中包含一个数组和两个坐标表示的环形队列、一个std::mutex成员、用来保证每次只被一个线程读写操作 （为了保证打印出来的消息是一行一行的，在它空闲的时候也借用的这个互斥量╮(╯▽╰)╭）、两个std::condition_VARiable表示队列不满和不空的状态，进而保证生产的时候不满，消耗的时候不空。

#pragma once#include chrono>
    //std::chrono#include mutex>
    //std::mutex,std::unique_lock,std::lock_Guard#include thread>
    //std::thread#include condition_variable>
    //std::condition_variable#include iostream>
    //std::cout,std::endl#include map>
//std::mapnamespace MyProducerToConsumer {
        static const int gRepositorySize = 10;
    //total size of the repository    static const int gItemNum = 97;
    //number of products to produce    std::mutex produce_mtx, consume_mtx;
    //mutex for all the producer thread or consumer thread    std::mapstd::thread::id, int>
     threadPErformance;
//records of every thread's producing/consuming number    struct ItemRepository {
    //repository class        int m_ItemBuffer[gRepositorySize];
    //Repository itself (as a circular queue)        int m_ProducePos;
    //rear position of circular queue        int m_ConsumePos;
    //head position of circular queue        std::mutex m_mtx;
    //mutex for operating the repository        std::condition_variable m_RepoUnfull;
    //indicating that this repository is unfull(then producers can produce items)        std::condition_variable m_RepoUnempty;
//indicating that this repository is unempty(then consumers can produce items)    }
    gItemRepo;
    void Produceitem(ItemRepository *ir, int item) {
            std::unique_lock std::mutex>
    ulk(ir->
    m_mtx);
            while ((ir->
    m_ProducePos + 1) % gRepositorySize == ir->
m_ConsumePos) {
    //full(spare one slot for indicating)            std::cout  "Reposity is full. Waiting for consumers..."  std::endl;
                ir->
    m_RepoUnfull.wait(ulk);
//unlocking ulk and waiting for unfull condition        }
            //when unfull        ir->
    m_ItemBuffer[ir->
    m_ProducePos++] = item;
    //procude and shift        std::cout  "Item No."  item  " produced successfully by "            std::this_thread::get_id()"!"  std::endl;
            threadPerformance[std::this_thread::get_id()]++;
            if (ir->
    m_ProducePos == gRepositorySize)//loop            ir->
    m_ProducePos = 0;
            ir->
    m_RepoUnempty.notify_all();
    //item produced, so it's unempty;
 notify all consumers    }
    int ConsumeItem(ItemRepository *ir) {
            std::unique_lockstd::mutex>
    ulk(ir->
    m_mtx);
            while (ir->
    m_ConsumePos == ir->
m_ProducePos) {
    //empty            std::cout  "Repository is empty.Waiting for producing..."  std::endl;
                ir->
    m_RepoUnempty.wait(ulk);
        }
            int item = ir->
    m_ItemBuffer[ir->
    m_ConsumePos++];
            std::cout  "Item No."  item  " consumed successfully by "            std::this_thread::get_id()"!"  std::endl;
            threadPerformance[std::this_thread::get_id()]++;
            if (ir->
    m_ConsumePos == gRepositorySize)            ir->
    m_ConsumePos = 0;
            ir->
    m_RepoUnfull.notify_all();
    //item consumed, so it's unempty;
     notify all consumers        return item;
    }
    void ProducerThread() {
            static int produced = 0;
//static variable to indicate the number of produced items        while (1) {
                std::this_thread::sleep_for(std::chrono::milliseconds(10));
    //sleep long enough in case it runs too fast for other threads to procude            std::lock_guardstd::mutex>
    lck(produce_mtx);
    //auto unlock when break            produced++;
                if (produced >
     gItemNum)break;
                gItemRepo.m_mtx.lock();
                std::cout  "Producing item No."  produced  "..."  std::endl;
                gItemRepo.m_mtx.unlock();
                ProduceItem(&
    gItemRepo, produced);
        }
            gItemRepo.m_mtx.lock();
            std::cout  "Producer thread "  std::this_thread::get_id()             " exited."  std::endl;
            gItemRepo.m_mtx.unlock();
    }
    void ConsumerThread() {
            static int consumed = 0;
        while (1) {
                std::this_thread::sleep_for(std::chrono::milliseconds(10));
                std::lock_guardstd::mutex>
    lck(consume_mtx);
                consumed++;
                if (consumed >
     gItemNum)break;
                gItemRepo.m_mtx.lock();
                std::cout  "Consuming item available..."  std::endl;
                gItemRepo.m_mtx.unlock();
                ConsumeItem(&
    gItemRepo);
        }
            gItemRepo.m_mtx.lock();
            std::cout  "Consumer thread "  std::this_thread::get_id()             " exited."  std::endl;
            gItemRepo.m_mtx.unlock();
    }
    void InitItemRepository(ItemRepository* ir) {
            ir->
    m_ConsumePos = 0;
            ir->
    m_ProducePos = 0;
    }
    void Run() {
            InitItemRepository(&
    gItemRepo);
            std::thread thdConsume[11];
            std::thread thdProduce[11];
            for (auto&
     t : thdConsume)t = std::thread(ConsumerThread);
            for (auto&
     t : thdProduce)t = std::thread(ProducerThread);
            for (auto&
     t : thdConsume)t.join();
            for (auto&
     t : thdProduce)t.join();
            for (auto&
     iter : threadPerformance)cout  iter.First  ":"  iter.second  endl;
    }
}
    

相关文章：

关于java生产者与消费者的实例详解

java多线程之并发协作生产者消费者设计模式

以上就是C++解决方法：多线程同步经典案例之生产者消费者问题的详细内容，更多请关注其它相关文章！

声明：本文内容由网友自发贡献，本站不承担相应法律责任。对本内容有异议或投诉，请联系2913721942#qq.com核实处理，我们将尽快回复您，谢谢合作！