zthread学习实例十线程间的协助（二）

例一、生产者-消费者

 

 

一个任务制作烤面包（生产者），一个任务给烤面包抹黄油（消费者），还有一个任务是往抹好黄油的烤面包上抹果酱（消费者）。

代码如下：

#include "stdafx.h"#include "zthread/FastMutex.h"#include "zthread/CountedPtr.h"#include "zthread/Runnable.h"#include "zthread/Condition.h"#include <iostream>using namespace ZThread;using namespace std;class Jammer : public Runnable{public:Jammer() : butteredToastReady(Lock), jammed(0), isGotButteredToast(false){}void moreButteredToastReady(){Guard<Mutex> g(Lock);isGotButteredToast = true;butteredToastReady.signal();}void run(){try{while (!Thread::interrupted()){{Guard<Mutex> g(Lock);while (isGotButteredToast == false)butteredToastReady.wait();++jammed;}cout <<"Putting jam on toast : " <<jammed <<endl;{Guard<Mutex> g(Lock);isGotButteredToast = false;}}}catch (Interrupted_Exception& e){cout <<" Jarry : "<<e.what() <<endl;}}private:MutexLock;ConditionbutteredToastReady;boolisGotButteredToast;intjammed;};class Butterer : public Runnable{public:Butterer(CountedPtr<Jammer>& j) : toastReady(Lock), jammer(j), Buttered(0), isGotToast(false){}void moreToastReady(){Guard<Mutex> g(Lock);isGotToast = true;toastReady.signal();}void run(){try{while (!Thread::interrupted()){{Guard<Mutex> g(Lock);while (!isGotToast)toastReady.wait();++Buttered;}cout <<"Buttering toast :" <<Buttered <<endl;jammer->moreButteredToastReady();{Guard<Mutex> g(Lock);isGotToast = false;}}}catch (Interrupted_Exception& e){cout <<" Jarry " <<e.what() <<endl;}}private:MutexLock;ConditiontoastReady;CountedPtr<Jammer>jammer;boolisGotToast;intButtered;};class Toast : public Runnable{public:Toast(CountedPtr<Butterer>& aButterer) : butterer(aButterer), toasted(0) {}void run(){try{while (!Thread::interrupted()){Thread::sleep(1000);cout <<"A new Toast" <<++toasted <<endl;butterer->moreToastReady();}}catch (Interrupted_Exception& e){cerr <<"Jarry " <<e.what() <<endl;}}private:CountedPtr<Butterer> butterer;int toasted;};int _tmain(int argc, _TCHAR* argv[]) {try{CountedPtr<Jammer>jammer(new Jammer);CountedPtr<Butterer> butterer(new Butterer(jammer));CountedPtr<Toast>toast(new Toast(butterer));ThreadedExecutor executor;executor.execute(toast);executor.execute(jammer);executor.execute(butterer);cin.get();executor.interrupt();}catch (Synchronization_Exception& e){cerr <<" Jarry Main " <<e.what() <<endl;}return 0;} 

 

 

 

线程处理问题常常基于需要对任务进行串行化处理。使用队列可以采用同步的方式访问其内部元素，这样可以解决很多线程处理问题，下面基于STL中的deque实现的队列：

#ifndef QUEUE_H#define QUEUE_H#include "zthread/Runnable.h"#include "zthread/PoolExecutor.h" template<class T> class TQueue{//锁子 ZThread::Mutex lock; //线程协作基类 ZThread::Condition cond; //线程队列元素 std::deque<T> data; public: TQueue() : cond(lock) {} void put(T item) { ZThread::Guard<ZThread::Mutex> g(lock); data.push_back(item); cond.signal();} T get() { ZThread::Guard<ZThread::Mutex> g(lock); while(data.empty()) cond.wait(); T returnVal = data.front(); data.pop_front();return returnVal; } }; #endif 

 

1、加入同步以确保在同一时刻不会有两个线程添加对象。

2、加入wait()和signal()在队列空时，挂起线程，并在有多个元素可用时恢复执行。

 

 

 

二、广播

 

signal()函数唤醒了一个正在等待Condition对象的线程。然而，也许会有多个线程在等待某个相同的条件对象，在这种情况下需要使用broadcast()（与调用线程的个数的signal()效果相同）而不是signal()把这些线程唤醒。

 

 

 

使用广播的例子：每辆Car将在几个阶段内装配完成，本例中将看到这样一个阶段：底盘制作好之后的这段时间里可以 同时 安装发动机、驱动传动装置和车轮(广播)。本例中综合中前面中所有关于线程的知识。

#include <iostream> #include <string> #include "zthread/Thread.h" #include "zthread/Mutex.h" #include "zthread/Guard.h" #include "zthread/Condition.h" #include "zthread/ThreadedExecutor.h" #include "TQueue.h" using namespace ZThread; using namespace std; //driveTrain 驱动传动 class Car { intid; boolengine, driveTrain, wheels; public: Car(int idn) : id(idn), engine(false), driveTrain(false), wheels(false) {} // Empty Car object: Car() : id(-1), engine(false), driveTrain(false), wheels(false) {} // Unsynchronized -- assumes atomic bool operations: int getId() { return id; } void addEngine() { engine = true; } bool engineInstalled() { return engine; } void addDriveTrain() { driveTrain = true; } bool driveTrainInstalled() { return driveTrain; } void addWheels() { wheels = true; } bool wheelsInstalled() { return wheels; } friend ostream& operator<<(ostream& os, const Car& c){return os <<"Car " <<c.id <<" ["<<" engine: " <<c.engine<<" driveTrain: " <<c.driveTrain<<" wheels: " <<c.wheels <<" ]";}}; typedef CountedPtr<TQueue<Car> > CarQueue; class ChassisBuilder : public Runnable{CarQueuecarQueue; intcounter;public: ChassisBuilder(CarQueue& cq) : carQueue(cq),counter(0) {} void run(){try {while(!Thread::interrupted()) {Thread::sleep(100); // Make chassis: Car c(counter++);cout <<c <<endl; // Insert into queuecarQueue->put(c);}} catch(Interrupted_Exception&) { /* Exit */ }cout <<"ChassisBuilder off" <<endl; } }; class Cradle{Carc; // Holds current car being worked on booloccupied;MutexworkLock, readyLock;ConditionworkCondition, readyCondition;boolengineBotHired, wheelBotHired, driveTrainBotHired;public: Cradle() : workCondition(workLock), readyCondition(readyLock){ occupied= false;engineBotHired= true;wheelBotHired= true;driveTrainBotHired= true;} //向底盘生产线中加入car对象 //并修改底盘生产线的状态（被占用） void insertCar(Car chassis) {c = chassis;occupied = true;}//从底盘生产线中取出对象 //修正状态为未被占用，并返回car对象Car getCar() {// Can only extract car once if(!occupied){ cerr <<"No Car in Cradle for getCar()" <<endl; return Car(); // "Null" Car object } occupied = false; return c; } // Access car while in cradle: //重载指向成员符号 //可以使用(*Cradle)直接调用Car* operator->() { return &c; }// Allow robots to offer services to this cradle:void offerEngineBotServices(){Guard<Mutex> g(workLock);while(engineBotHired) workCondition.wait();engineBotHired = true; // Accept the job }void offerWheelBotServices() { Guard<Mutex> g(workLock);while(wheelBotHired) workCondition.wait();//这个wait被挂起，直到收到了startWork()的广播 //在startWork()中，雇佣的状态会被修改，因此下一步的操作是修正状态 wheelBotHired = true; // Accept the job } void offerDriveTrainBotServices(){ Guard<Mutex> g(workLock); while(driveTrainBotHired) workCondition.wait(); driveTrainBotHired = true; // Accept the job } // Tell waiting robots that work is ready: void startWork(){ Guard<Mutex> g(workLock); engineBotHired= false; wheelBotHired= false; driveTrainBotHired= false; workCondition.broadcast(); //等价于3个signal()//workCondition.signal(); //workCondition.signal(); //workCondition.signal(); } // Each robot reports when their job is done: void taskFinished() { Guard<Mutex> g(readyLock); //Guard<Mutex> g(workLock); readyCondition.signal();} // Director waits until all jobs are done: void waitUntilWorkFinished(){ Guard<Mutex> g(readyLock);while(!(c.engineInstalled() && c.driveTrainInstalled() && c.wheelsInstalled())) readyCondition.wait();}};typedef CountedPtr<Cradle> CradlePtr; class Director : public Runnable{ CarQueuechassisQueue, finishingQueue;CradlePtrcradle;public:Director(CarQueue& cq, CarQueue& fq, CradlePtr cr) : chassisQueue(cq), finishingQueue(fq), cradle(cr) {} void run() {try{while(!Thread::interrupted()){// Blocks until chassis is available: cradle->insertCar(chassisQueue->get());// Notify robots car is ready for workcradle->startWork();// Wait until work completes cradle->waitUntilWorkFinished();// Put car into queue for further work finishingQueue->put(cradle->getCar()); } }catch(Interrupted_Exception&) { /* Exit */ }cout <<"Director off" <<endl; } };class EngineRobot : public Runnable{ CradlePtr cradle;public: EngineRobot(CradlePtr cr) : cradle(cr) {} void run() { try{ while(!Thread::interrupted()){// Blocks until job is offered/accepted: cradle->offerEngineBotServices();cout <<"Installing engine" <<endl; (*cradle)->addEngine();//隐式的调用了operator->()还是隐式的类型转换？ cradle->taskFinished(); } } catch(Interrupted_Exception&) { /* Exit */ } cout <<"EngineRobot off" <<endl;}};class DriveTrainRobot : public Runnable{ CradlePtr cradle; public: DriveTrainRobot(CradlePtr cr) : cradle(cr) {}void run() { try { while(!Thread::interrupted()){ // Blocks until job is offered/accepted: cradle->offerDriveTrainBotServices();cout <<"Installing DriveTrain" <<endl;(*cradle.operator ->())->addDriveTrain(); cradle->taskFinished(); } } catch(Interrupted_Exception&) { /* Exit */ } cout <<"DriveTrainRobot off" <<endl; } }; class WheelRobot : public Runnable{ CradlePtr cradle; public: WheelRobot(CradlePtr cr) : cradle(cr) {} void run() { try{ while(!Thread::interrupted()) { // Blocks until job is offered/accepted: cradle->offerWheelBotServices(); cout <<"Installing Wheels" <<endl;(*cradle)->addWheels(); cradle->taskFinished(); } } catch(Interrupted_Exception&) { /* Exit */ } cout <<"WheelRobot off" <<endl; }};class Reporter : public Runnable{ CarQueue carQueue; public: Reporter(CarQueue& cq) : carQueue(cq) {} void run(){try { while(!Thread::interrupted()) { cout <<carQueue->get() <<endl; } } catch(Interrupted_Exception&) { /* Exit */ } cout <<"Reporter off" <<endl; } };int main() { cout <<"Press <Enter> to quit" <<endl; try { CarQueue chassisQueue(new TQueue<Car>), finishingQueue(new TQueue<Car>); CradlePtr cradle(new Cradle); ThreadedExecutor assemblyLine; assemblyLine.execute(new EngineRobot(cradle)); assemblyLine.execute(new DriveTrainRobot(cradle)); assemblyLine.execute(new WheelRobot(cradle)); assemblyLine.execute(new Director(chassisQueue, finishingQueue, cradle));assemblyLine.execute(new Reporter(finishingQueue)); // Start everything running by producing chassis: assemblyLine.execute(new ChassisBuilder(chassisQueue)); cin.get(); assemblyLine.interrupt(); } catch(Synchronization_Exception& e) { cerr <<e.what() <<endl; } } ///:~