抄自維基百科 :
生產者消費者問題(英語: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::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'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_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'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; } }
相關文章:
以上是C++解決方法:多執行緒同步經典案例之生產者消費者問題的詳細內容。更多資訊請關注PHP中文網其他相關文章!