Qt - 生產者和消費者模型示例

簡介

使用條件變量，信號量，兩種示例方式去實現生產者和消費者模型

 

1、條件變量   QWaitCondition

#ifndef MUTEXWAITCONDITION
#define MUTEXWAITCONDITION

#include <QCoreApplication>
#include <iostream>
#include <QThread>
#include <QMutex>
#include <QWaitCondition>

using namespace std;

class MutexWaitCondition {
public:
    //預計生產(或消費)數量
    int loopCount;
    //當前產品數量
    int product;
    //倉庫能容納的最大產品數量
    int capacity;

    QMutex mutex;
    /*
     The QWaitCondition class provides a condition variable for synchronizing threads
     QWaitCondition類為線程同步提供了一個條件變量
     */
    QWaitCondition productIsNotFull;
    QWaitCondition productIsNotEmpty;

    //生產者
    class Producer : public QThread {
    public:
        Producer(MutexWaitCondition *manager) : QThread() {
            this->manager = manager;
        }
    protected:
        void run() {
            for(int i=0; i<manager->loopCount; i++) {
                manager->mutex.lock();
                while(manager->product == manager->capacity) {
                    /*
                     bool QWaitCondition::wait(QReadWriteLock * lockedReadWriteLock, unsigned long time = ULONG_MAX)
                     Releases the lockedReadWriteLock and waits on the wait condition
                     釋放該鎖，並且阻塞當前線程，直到條件滿足（即調用wake方法被喚醒）
                     */
                    manager->productIsNotFull.wait(&manager->mutex);
                }
                manager->product++;   //當前產品數量++之前產品就要入隊
                //cout<<"P";
                cout<<i<<".P="<<manager->product<<", ";
                manager->productIsNotEmpty.wakeAll();
                manager->mutex.unlock();
            }
        }
    private:
        MutexWaitCondition *manager;
    };

    //消費者
    class Consumer : public QThread {
    public:
        Consumer(MutexWaitCondition *manager) : QThread() {
            this->manager = manager;
        }
    protected:
        void run() {
            for(int i=0; i<manager->loopCount; i++) {
                manager->mutex.lock();
                while(manager->product == 0) {
                    manager->productIsNotEmpty.wait(&manager->mutex);
                }
                manager->product--;   //當前產品數量--之前產品就要出隊
                //cout<<"C";
                cout<<i<<".C="<<manager->product<<", ";
                manager->productIsNotFull.wakeAll();
                manager->mutex.unlock();
            }
        }
    private:
        MutexWaitCondition *manager;
    };

//無修飾的方法，默認是private的
public:
    void test(int loopCount, int capacity)
    {
        this->loopCount = loopCount;
        this->capacity = capacity;
        this->product = 0;

        Producer producer(this);
        Consumer consumer(this);
        //thread.start會調用thread內部的run方法
        producer.start();
        consumer.start();
        /*
         Blocks the thread until either of these conditions is met:
         阻塞該線程直到所有條件都滿足
         */
        producer.wait();
        consumer.wait();
        cout<<endl;
    }
};
#endif // MUTEXWAITCONDITION

 

2、信號量   QSemaphore

#ifndef SEMAPHOREMUTEX
#define SEMAPHOREMUTEX

#include <QCoreApplication>
#include <iostream>
#include <QThread>
#include <QSemaphore>
//注意:網上很多Semaphore不用mutex的做法都是錯誤的
#include <QMutex>

using namespace std;

class SemaphoreMutex {
public:
    //預計生產(或消費)數量
    int loopCount;
    //當前產品數量: productSemphore.avaliable()
    //int product;
    //倉庫能容納的最大產品數量
    int capacity;

    QMutex mutex;
    /*
     The QSemaphore class provides a general counting semaphore
     QSemaphore類提供了一個通用的計數信號量
     */
    QSemaphore *productSemphore;
    QSemaphore *leftSpaceSemaphore;

    //生產者
    class Producer : public QThread {
    public:
        Producer(SemaphoreMutex *manager) : QThread() {
            this->manager = manager;
        }
    protected:
        void run() {
            for(int i=0; i<manager->loopCount; i++) {
                /*
                 Tries to acquire n resources guarded by the semaphore. If n > available(), this call will block until enough resources are available.
                 嘗試去獲取(減去)n個被信號量控制的資源。如果n>可用資源數量，它就會阻塞直到有足夠的資源為止。
                 */
                manager->leftSpaceSemaphore->acquire();
                //之所以lock要在acquire后面是因為: 如果消費者拿到了鎖，那么又沒有商品，那么久會導致死鎖
                manager->mutex.lock();   //加鎖之后產品就要入隊，然后再release
                manager->productSemphore->release();
                //cout<<"P";
                cout<<i<<".P="<<manager->productSemphore->available()<<", ";
                manager->mutex.unlock();
            }
        }
    private:
        SemaphoreMutex *manager;
    };

    //消費者
    class Consumer : public QThread {
    public:
        Consumer(SemaphoreMutex *manager) : QThread() {
            this->manager = manager;
        }
    protected:
        void run() {
            for(int i=0; i<manager->loopCount; i++) {
                manager->productSemphore->acquire();
                manager->mutex.lock();   //加鎖之后產品就要出隊，然后再release
                manager->leftSpaceSemaphore->release();
                //cout<<"C";
                cout<<i<<".C="<<manager->productSemphore->available()<<", ";
                manager->mutex.unlock();
            }
        }
    private:
        SemaphoreMutex *manager;
    };

//無修飾的方法，默認是private的
public:
    void test(int loopCount, int capacity)
    {
        this->loopCount = loopCount;
        this->capacity = capacity;

        //參數為: 信號量的當前值
        productSemphore = new QSemaphore(0);
        leftSpaceSemaphore = new QSemaphore(capacity);

        Producer producer(this);
        Consumer consumer(this);
        //thread.start會調用thread內部的run方法
        producer.start();
        consumer.start();
        /*
         Blocks the thread until either of these conditions is met:
         阻塞該線程直到所有條件都滿足
         */
        producer.wait();
        consumer.wait();
        cout<<endl;
    }
};

#endif // SEMAPHOREMUTEX