Qt多线程问题3：基于线程同步实现生产者消费者模型-编程知识

Qt多线程问题3：基于线程同步实现生产者消费者模型

之前自己写了一个收数据的功能，但是没有数据的时候就要不停监听，那监听多久合适呢？1ms ？10 ms ? 100ms？但是你也不确定数据什么时候来，那么就可以使用经典的生产者消费者模型。这次的实验中我参考了人民邮电出版社<<Qt5.9 C++开发指南>>中的实现。

原书中的实现，将锁和同步量声明为全局变量，并不是很优雅。我们可以实现一个管理类，将锁和同步量写在管理类里边，管理类管理生产者和消费者，生产者和消费者都可以看到管理类内的互斥量和同步量，代码结构如下：

管理类ThreadManager

头文件的声明如下：

#ifndef THREADMANAGER_H
#define THREADMANAGER_H#include <QMutex>
#include <QObject>
#include <QThread>
#include <QSharedPointer>
#include <QScopedPointer>
#include <QMutexLocker>
#include <QWaitCondition>
#include <ProducerThread.h>
#include <ConsumerThread.h>class ThreadManager: public QObject
{Q_OBJECT
public:ThreadManager();void initConnections();void start();private:QSharedPointer<QMutex>              m_mutex;QSharedPointer<QWaitCondition>      m_waitCondition;QScopedPointer<ProducerThread>      m_producer;QScopedPointer<ConsumerThread>      m_consumer;
};#endif // THREADMANAGER_H

因为互斥量mutex是不支持赋值的，所以可以用共享指针给管理起来，然后在生产者和消费者线程中都声明同样的共享指针，指向同一个互斥量和同步量。.cpp实现如下：

#include "ThreadManager.h"ThreadManager::ThreadManager(): m_consumer(new ConsumerThread(this)), m_producer(new ProducerThread(this)), m_mutex(new QMutex), m_waitCondition(new QWaitCondition)
{m_producer->setSharedMutex(m_mutex);m_producer->setSharedCondition(m_waitCondition);m_consumer->setSharedMutex(m_mutex);m_consumer->setSharedCondition(m_waitCondition);initConnections();
}void ThreadManager::initConnections()
{connect(m_producer.data(), &ProducerThread::sendData,m_consumer.data(), &ConsumerThread::comsumeData);
}void ThreadManager::start()
{m_consumer->start();m_producer->updateStopState(true);m_producer->start();
}

函数的其他说明：构造函数， m_producer与m_consumer，都设置好互斥量和同步量的共享指针；initConnections()：连接生产者与消费者；start()：模拟启动。

生产者ProducerThread

声明如下：

#ifndef PRODUCERTHREAD_H
#define PRODUCERTHREAD_H#include <QSharedPointer>
#include <QWaitCondition>
#include <QThread>
#include <QDebug>
#include <QMutexLocker>class ProducerThread: public QThread
{Q_OBJECT
public:explicit ProducerThread(QObject *parent);void updateStopState(bool state) {m_state = state;}void setSharedMutex(QSharedPointer<QMutex> point);void setSharedCondition(QSharedPointer<QWaitCondition> point);private:bool                                m_state;int                                 m_startVal;QSharedPointer<QMutex>              m_mutex;QSharedPointer<QWaitCondition>      m_waitCondition;protected:void run() override;signals:void sendData(int val);
};#endif // PRODUCERTHREAD_H

函数说明：updateStopState()：决定是否继续发数据；setSharedMutex()和setSharedContition()，给成员变量m_mutex以及m_waitCondition赋值，指向管理者的互斥量与同步量；定义好signal，sendData()用来发送数据。.cpp实现如下：

#include "ProducerThread.h"
#include "ThreadManager.h"ProducerThread::ProducerThread(QObject *parent): m_state(false), m_startVal(0)
{
}void ProducerThread::setSharedMutex(QSharedPointer<QMutex> point)
{m_mutex = point;
}void ProducerThread::setSharedCondition(QSharedPointer<QWaitCondition> point)
{m_waitCondition = point;
}void ProducerThread::run()
{while(m_state) {QMutexLocker lock(m_mutex.data());qDebug() << "Send Data: " << QString::number(m_startVal);emit sendData(m_startVal++);lock.unlock();m_waitCondition->wakeOne();QThread::sleep(2);}
}

在run()函数中，不停加锁再发数据，发数据后用wakeOne()唤醒正在阻塞的消费者，然后中间间隔2s。

消费者ConsumerThread

#ifndef CONSUMERTHREAD_H
#define CONSUMERTHREAD_H#include <QDebug>
#include <QThread>
#include <QMutexLocker>
#include <QSharedPointer>
#include <QWaitCondition>class ConsumerThread: public QThread
{Q_OBJECT
public:explicit ConsumerThread(QObject *parent);void updateState(bool state) {m_state = state;}void setSharedMutex(QSharedPointer<QMutex> point);void setSharedCondition(QSharedPointer<QWaitCondition> point);private:bool                                m_state;int                                 m_data;QSharedPointer<QMutex>              m_mutex;QSharedPointer<QWaitCondition>      m_waitCondition;protected:void run() override;public slots:void comsumeData(int val);
};#endif // CONSUMERTHREAD_H

逻辑基本和生产者一样，定义好槽函数，用来处理数据即可。

#include "ConsumerThread.h"
#include "ThreadManager.h"ConsumerThread::ConsumerThread(QObject *parent): m_state(true)
{
}void ConsumerThread::setSharedMutex(QSharedPointer<QMutex> point)
{m_mutex = point;
}void ConsumerThread::setSharedCondition(QSharedPointer<QWaitCondition> point)
{m_waitCondition = point;
}void ConsumerThread::comsumeData(int val)
{m_data = val;
}void ConsumerThread::run()
{while(m_state) {QMutexLocker lock(m_mutex.data());m_waitCondition->wait(m_mutex.data());qDebug() << "Consume Data: " << m_data;lock.unlock();}
}

run()中的部分比较重要，现加锁，再通过wait解锁，等待唤醒，得到数据就打印。

主函数

#include "mainwindow.h"
#include <QDebug>
#include <QApplication>
#include "ThreadManager.h"int main(int argc, char *argv[])
{QApplication a(argc, argv);MainWindow w;// w.show();ThreadManager manager;manager.start();qDebug() << "Finished!";return a.exec();
}

通过manager.start()开始模拟，执行结果如下：