C++-Lösung: klassischer Fall der Multithread-Synchronisation: Producer-Consumer-Problem

Aus Wikipedia kopiert:

Produzenten-Konsumenten-Problem (Englisch: Produzenten-Konsumenten-Problem), auch bekannt als Limited Buffering Problem (Englisch: Das Problem des begrenzten Puffers ist ein klassischer Fall eines Multithread-Synchronisationsproblems. Diese Frage beschreibt ein Problem, das auftritt, wenn zwei Threads, die sich einen Puffer fester Größe teilen – der sogenannte „Produzent“ und „Konsumer“ – tatsächlich ausgeführt werden. Die Hauptaufgabe des Produzenten besteht darin, eine bestimmte Datenmenge zu generieren, sie in den Puffer zu legen und dann den Vorgang zu wiederholen. Gleichzeitig verbraucht der Verbraucher auch die Daten im Puffer. Der Schlüssel zu diesem Problem besteht darin, sicherzustellen, dass der Produzent keine Daten hinzufügt, wenn der Puffer voll ist, und der Verbraucher keine Daten verbraucht, wenn der Puffer leer ist.

Um dieses Problem zu lösen, müssen Sie den Produzenten in den Ruhezustand versetzen, wenn der Puffer voll ist (oder die Daten einfach aufgeben). Der Produzent kann erst dann geweckt werden, wenn der Konsument die Daten im Puffer das nächste Mal verbraucht Hinzufügen von Daten zum Puffer. Ebenso können Sie den Verbraucher in den Ruhezustand versetzen, wenn der Puffer leer ist, warten, bis der Produzent Daten zum Puffer hinzufügt, und den Verbraucher dann aufwecken.

In diesem Artikel wird eine ItemRepository-Klasse zur Darstellung des Produktlagers verwendet, die ein Array und zwei kreisförmige Warteschlangen enthält, die durch Koordinaten dargestellt werden, sowie ein std::mutex-Mitglied, um sicherzustellen, dass es nur von Benutzern gelesen und geschrieben wird jeweils ein Thread. (Um sicherzustellen, dass die gedruckten Nachrichten Zeile für Zeile vorliegen, wird der Mutex ╮(╯▽╰)╭ auch ausgeliehen, wenn er inaktiv ist. Die beiden std::condition_variables repräsentieren den Status der Warteschlange Es muss voll und nicht leer sein, wodurch sichergestellt wird, dass es bei der Produktion nicht voll und beim Verbrauch nicht leer ist.

#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;
    }
}

Verwandte Artikel:

Detaillierte Beispiele für Java-Produzenten und -Konsumenten

Java Multi-Threaded Concurrent Collaborative Producer Consumption Design Pattern

Das obige ist der detaillierte Inhalt vonC++-Lösung: klassischer Fall der Multithread-Synchronisation: Producer-Consumer-Problem. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!