Solution C++ : cas classique de synchronisation multi-thread : problème producteur-consommateur

Copié de Wikipédia :

Problème de producteur-consommateur (anglais : problème de producteur-consommateur), également connu sous le nom de Problème de tampon limité (anglais : Problème de tampon limité), est un cas classique de problème de synchronisation multi-thread. Cette question décrit un problème qui se produit lorsque deux threads partageant un tampon de taille fixe – les soi-disant « producteur » et « consommateur » – s'exécutent réellement. Le rôle principal du producteur est de générer une certaine quantité de données et de les mettre dans la mémoire tampon, puis de répéter le processus. Dans le même temps, le consommateur consomme également les données stockées dans la mémoire tampon. La clé de ce problème est de garantir que le producteur n'ajoute pas de données lorsque le tampon est plein et que le consommateur ne consomme pas de données lorsque le tampon est vide.

Pour résoudre ce problème, vous devez laisser le producteur dormir lorsque le tampon est plein (ou simplement abandonner les données. Le producteur ne peut pas être réveillé jusqu'à la prochaine fois que le consommateur consomme les données dans le tampon Start). ajouter des données au tampon. De même, vous pouvez également laisser le consommateur s'endormir lorsque le tampon est vide, attendre que le producteur ajoute des données au tampon, puis réveiller le consommateur.

Cet article utilise une classe ItemRepository pour représenter l'entrepôt de produits, qui contient un tableau et deux files d'attente circulaires représentées par des coordonnées, et un membre std::mutex pour garantir qu'il est lu et écrit uniquement par un thread à la fois. (Afin de garantir que les messages imprimés sont ligne par ligne, le mutex ╮(╯▽╰)╭ est également emprunté lorsqu'il est inactif. Les deux std::condition_variables représentent l'état de la file d'attente). être plein et non vide, garantissant ainsi qu'il n'est pas plein lorsqu'il est produit, mais pas vide lorsqu'il est consommé.

#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::map
namespace 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::map<std::thread::id, int> threadPerformance;//records of every thread&#39;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&#39;s unempty; notify all consumers
    }

    int ConsumeItem(ItemRepository *ir) {
        std::unique_lock<std::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&#39;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_guard<std::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_guard<std::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;
    }
}

Articles associés :

Explication détaillée d'exemples de producteurs et de consommateurs Java

Production collaborative simultanée de multi-threads Java Modèle de conception du consommateur

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!