1. 信號(hào)量的原理

信號(hào)量(Semaphore)是另一種限制對(duì)共享資源進(jìn)行訪問(wèn)的線程同步機(jī)制，它與互斥量(Mutex)相似，但是有區(qū)別。一個(gè)互斥量只能被鎖定一次，而信號(hào)量可以多次使用。信號(hào)量通常用來(lái)保護(hù)一定數(shù)量的相同的資源，如數(shù)據(jù)采集時(shí)的雙緩沖區(qū)。

QSemaphore 是實(shí)現(xiàn)信號(hào)量功能的類，它提供以下幾個(gè)基本的函數(shù):

• acquire(int n)嘗試獲得 n 個(gè)資源。如果沒(méi)有這么多資源，線程將阻塞直到有 n 個(gè)資源可用
• release(int n)釋放 n 個(gè)資源，如果信號(hào)量的資源已全部可用之后再 release()，就可以創(chuàng)建更多的資源，增加可用資源的個(gè)數(shù):
• int available()返回當(dāng)前信號(hào)量可用的資源個(gè)數(shù)，這個(gè)數(shù)永遠(yuǎn)不可能為負(fù)數(shù)，如果為 0，就說(shuō)明當(dāng)前沒(méi)有資源可用;
• bool tryAcquire(int n = 1)，嘗試獲取 n 個(gè)資源，不成功時(shí)不阻塞線程。

定義QSemaphore 的實(shí)例時(shí)，可以傳遞一個(gè)數(shù)值作為初始可用的資源個(gè)數(shù)。

下面的一段示意代碼，說(shuō)明 QSemaphore 的幾個(gè)函數(shù)的作用。

QSemaphore WC(5);  // WC.available() == 5，初始資源個(gè)數(shù)為 5個(gè)
WC.acquire(4):  // WC.available() == 1，用了4 個(gè)資源，還剩余1個(gè)可用
WC.release(2);  // WC.available() == 3，釋放了2個(gè)資源，剩余3個(gè)可用
WC.acquire(3);  // WC.available() == 0，又用了3 個(gè)資源，剩余0個(gè)可用
WC.tryAcquire(1);  //因?yàn)閃C.available() == 0，返回 false，
WC.acquire();  //因?yàn)?wc.available() == 0，沒(méi)有資源可用，阻塞

為了理解信號(hào)量及上面這段代碼的意義，可以假想變量 WC 是一個(gè)公共衛(wèi)生間，初始化時(shí)定義WC有5個(gè)位置可用。

• WC.acquire(4)，成功進(jìn)去 4 個(gè)人，占用了4 個(gè)位置，還剩余1個(gè)位置

• WC.release(2)，出來(lái)了 2個(gè)人，剩余3 個(gè)位置可用:

• WC.acquire(3)，又進(jìn)去 3 個(gè)人，剩余0個(gè)位置可用;

• WC.tryAcquire(1),有一個(gè)人嘗試進(jìn)去，但是因?yàn)闆](méi)有位置了，他不等待，走了,tryAcquire()函數(shù)返回 false:

• WC.acquire()，有一個(gè)人必須進(jìn)去，但是因?yàn)闆](méi)有位置了，他就一直在外面等著，直到有其他人出來(lái)，空余出位置來(lái)。

互斥量相當(dāng)于列車上的衛(wèi)生間，一次只允許一個(gè)人進(jìn)出，信號(hào)量則是多人公共衛(wèi)生間，允許多人進(jìn)出。n 個(gè)資源就是信號(hào)量需要保護(hù)的共享資源，至于資源如何分配，就是內(nèi)部處理的問(wèn)題了。

2. 雙緩沖區(qū)數(shù)據(jù)采集和讀取線程類設(shè)計(jì)

理解：可以用于實(shí)現(xiàn)自行定義的緩沖區(qū)大小，利用2個(gè)子線程對(duì)不斷產(chǎn)生的數(shù)據(jù)不間斷進(jìn)行寫入及處理，主線程主要進(jìn)行顯示

信號(hào)量通常用來(lái)保護(hù)一定數(shù)量的相同的資源，如數(shù)據(jù)采集時(shí)的雙緩沖區(qū)，適用于Producer/Consumer 模型。

在實(shí)例 samp13_5中，創(chuàng)建類似于 Producer/Consumer 模型的兩個(gè)線程類 QThreadDAQ 和QThreadShow。qmythread.h 文件中這兩個(gè)類的定義如下:

#ifndef QMYTHREAD_H
#define QMYTHREAD_H//#include    <QObject>
#include    <QThread>class QThreadDAQ : public QThread
{Q_OBJECTprivate:bool    m_stop=false; //停止線程
protected:void    run() Q_DECL_OVERRIDE;
public:QThreadDAQ();void    stopThread();
};class QThreadShow : public QThread
{Q_OBJECT
private:bool    m_stop=false; //停止線程
protected:void    run() Q_DECL_OVERRIDE;
public:QThreadShow();void    stopThread();
signals:void    newValue(int *data,int count, int seq);
};
#endif // QMYTHREAD_H

QThreadDAQ 是數(shù)據(jù)采集線程，例如在使用數(shù)據(jù)采集卡進(jìn)行連續(xù)數(shù)據(jù)采集時(shí)，需要一個(gè)單獨(dú)的線程將采集卡采集的數(shù)據(jù)讀取到緩沖區(qū)內(nèi)。
QThreadShow 是數(shù)據(jù)讀取線程,用于讀取已存滿數(shù)據(jù)的緩沖區(qū)中的數(shù)據(jù)并傳遞給主線程顯示,采用信號(hào)與槽機(jī)制與主線程交互。
QThreadDAQ/QThreadShow 類的定義與使用 QWaitCondition 的實(shí)例 samp13_4中的QThreadProducer/QThreadConsumer 類的定義類似，只是QThreadShow 的信號(hào) newValue()采用了指針作為傳遞參數(shù)，用于一次傳遞出一個(gè)緩沖區(qū)的數(shù)據(jù)。

qmythread.cpp 文件中QThreadDAQ和QThreadShow 的主要功能代碼如下:

#include    "qmythread.h"
#include    <QSemaphore>const int BufferSize = 8;
int buffer1[BufferSize];
int buffer2[BufferSize];
int curBuf=1; //當(dāng)前正在寫入的Bufferint bufNo=0; //采集的緩沖區(qū)序號(hào)quint8   counter=0;//數(shù)據(jù)生成器QSemaphore emptyBufs(2);//信號(hào)量：空的緩沖區(qū)個(gè)數(shù)，初始資源個(gè)數(shù)為2
QSemaphore fullBufs; //滿的緩沖區(qū)個(gè)數(shù)，初始資源為0QThreadDAQ::QThreadDAQ()
{}void QThreadDAQ::stopThread()
{m_stop=true;
}void QThreadDAQ::run()
{m_stop=false;//啟動(dòng)線程時(shí)令m_stop=falsebufNo=0;//緩沖區(qū)序號(hào)curBuf=1; //當(dāng)前寫入使用的緩沖區(qū)counter=0;//數(shù)據(jù)生成器int n=emptyBufs.available();if (n<2)  //保證 線程啟動(dòng)時(shí)emptyBufs.available==2emptyBufs.release(2-n);while(!m_stop)//循環(huán)主體{emptyBufs.acquire();//獲取一個(gè)空的緩沖區(qū)for(int i=0;i<BufferSize;i++) //產(chǎn)生一個(gè)緩沖區(qū)的數(shù)據(jù){if (curBuf==1)buffer1[i]=counter; //向緩沖區(qū)寫入數(shù)據(jù)elsebuffer2[i]=counter;counter++; //模擬數(shù)據(jù)采集卡產(chǎn)生數(shù)據(jù)msleep(50); //每50ms產(chǎn)生一個(gè)數(shù)}bufNo++;//緩沖區(qū)序號(hào)if (curBuf==1) // 切換當(dāng)前寫入緩沖區(qū)curBuf=2;elsecurBuf=1;fullBufs.release(); //有了一個(gè)滿的緩沖區(qū),available==1}quit();
}void QThreadShow::run()
{m_stop=false;//啟動(dòng)線程時(shí)令m_stop=falseint n=fullBufs.available();if (n>0)fullBufs.acquire(n); //將fullBufs可用資源個(gè)數(shù)初始化為0while(!m_stop)//循環(huán)主體{fullBufs.acquire(); //等待有緩沖區(qū)滿,當(dāng)fullBufs.available==0阻塞int bufferData[BufferSize];int seq=bufNo;if(curBuf==1) //當(dāng)前在寫入的緩沖區(qū)是1，那么滿的緩沖區(qū)是2for (int i=0;i<BufferSize;i++)bufferData[i]=buffer2[i]; //快速拷貝緩沖區(qū)數(shù)據(jù)elsefor (int i=0;i<BufferSize;i++)bufferData[i]=buffer1[i];emptyBufs.release();//釋放一個(gè)空緩沖區(qū)emit    newValue(bufferData,BufferSize,seq);//給主線程傳遞數(shù)據(jù)}quit();
}QThreadShow::QThreadShow()
{}void QThreadShow::stopThread()
{m_stop=true;
}

在共享變量區(qū)定義了兩個(gè)緩沖區(qū) buffer1和 buffer2，都是長(zhǎng)度為 BufferSize 的數(shù)組。

變量 curBuf 記錄當(dāng)前寫入操作的緩沖區(qū)編號(hào)，其值只能是 1或2，表示 bufferl 或 buffer2，bufNo是累積的緩沖區(qū)個(gè)數(shù)編號(hào)，counter 是模擬采集數(shù)據(jù)的變量。
信號(hào)量emptyBufs 初始資源個(gè)數(shù)為2，表示有2個(gè)空的緩沖區(qū)可用。

信號(hào)量 fullBufs初始化資源個(gè)數(shù)為0，表示寫滿數(shù)據(jù)的緩沖區(qū)個(gè)數(shù)為零。

QThreadDAQ::run()采用雙緩沖方式進(jìn)行模擬數(shù)據(jù)采集，線程啟動(dòng)時(shí)初始化共享變量，特別的是使emptyBufs 的可用資源個(gè)數(shù)初始化為2。
在while 循環(huán)體里，第一行語(yǔ)句 emptyBufs.acquire()使信號(hào)量emptyBufs 獲取一個(gè)資源，即獲取一個(gè)空的緩沖區(qū)。用于數(shù)據(jù)緩存的有兩個(gè)緩沖區(qū)，只要有一個(gè)空的緩沖區(qū)，就可以向這個(gè)緩沖區(qū)寫入數(shù)據(jù)。

while 循環(huán)體里的 for 循環(huán)每隔 50 毫秒使 counter 值加 1，然后寫入當(dāng)前正在寫入的緩沖區(qū)，當(dāng)前寫入哪個(gè)緩沖區(qū)由 curBuf 決定。counter 是模擬采集的數(shù)據(jù)，連續(xù)增加可以判斷采集的數(shù)據(jù)是否連續(xù)。

完成 for 循環(huán)后正好寫滿一個(gè)緩沖區(qū)，這時(shí)改變 curBuf 的值，切換用于寫入的緩沖區(qū)。

寫滿一個(gè)緩沖區(qū)之后，使用 fullBufs.release()為信號(hào)量 fullBufs 釋放一個(gè)資源，這時(shí) fullBufs.available==l，表示有一個(gè)緩沖區(qū)被寫滿了。這樣，QThreadShow 線程里使用 fullBufs.acquire()就可以獲得一個(gè)資源，可以讀取已寫滿的緩沖區(qū)里的數(shù)據(jù)。

QThreadShow::run()用于監(jiān)測(cè)是否有已經(jīng)寫滿數(shù)據(jù)的緩沖區(qū)，只要有緩沖區(qū)寫滿了數(shù)據(jù)，就立刻讀取出數(shù)據(jù)，然后釋放這個(gè)緩沖區(qū)給 OThreadDAQ 線程用于寫入。

QThreadShow::run()函數(shù)的初始化部分使 fullBufs.available==0，即線程剛啟動(dòng)時(shí)是沒(méi)有資源的。

在 while循環(huán)體里第一行語(yǔ)句就是通過(guò) fullBufs.acquire()以阻塞方式獲取一個(gè)資源，只有當(dāng)QThreadDAQ 線程里寫滿一個(gè)緩沖區(qū)，執(zhí)行一次fullBufs.release()后，fullBufs.acquire()才獲得資源并執(zhí)行后面的代碼。后面的代碼就立即用臨時(shí)變量將緩沖區(qū)里的數(shù)據(jù)讀取出來(lái)，再調(diào)用emptyBufs.release()給信號(hào)量emptyBufs 釋放一個(gè)資源，然后發(fā)射信號(hào) newValue，由主線程讀取數(shù)據(jù)并顯示。
所以，這里使用了雙緩沖區(qū)、兩個(gè)信號(hào)量實(shí)現(xiàn)采集和讀取兩個(gè)線程的協(xié)調(diào)操作。采集線程里使用emptyBufs.acquire()獲取可以寫入的緩沖區(qū)。
實(shí)際使用數(shù)據(jù)采集卡進(jìn)行連續(xù)數(shù)據(jù)采集時(shí)，采集線程是不能停頓下來(lái)的，也就是說(shuō)萬(wàn)一讀取線程執(zhí)行較慢，采集線程是不會(huì)等待的。所以實(shí)際情況下，讀取線程的操作應(yīng)該比采集線程快。

3. QThreadDAQ和QThreadShow 的使用

設(shè)計(jì)窗口基于 QDialog 應(yīng)用程序 samp13_5，對(duì)話框的類定義如下(省略了一些不重要的或與前面實(shí)例重復(fù)的部分內(nèi)容):

class Dialog : public QDialog
{Q_OBJECTprivate:QThreadDAQ   threadProducer;QThreadShow   threadConsumer;
private slots:void    onthreadB_newValue(int *data, int count, int bufNo);};

Dialog類定義了兩個(gè)線程的實(shí)例，threadProducer 和 threadConsumer。

自定義了一個(gè)槽函數(shù) onthreadB_newValue()，用于與 threadConsumer 的信號(hào)關(guān)聯(lián)，在 Dialog的構(gòu)造函數(shù)里進(jìn)行了關(guān)聯(lián)。

connect(&threadConsumer,SIGNAL(newValue(int*,int,int)),this,SLOT(onthreadB_newValue(int*,int,int)));

槽函數(shù)onthreadB_newValue()的功能就是讀取一個(gè)緩沖區(qū)里的數(shù)據(jù)并顯示，其實(shí)現(xiàn)代碼如下

void Dialog::onthreadB_newValue(int *data, int count, int bufNo)
{ //讀取threadConsumer 傳遞的緩沖區(qū)的數(shù)據(jù)QString  str=QString::asprintf("第 %d 個(gè)緩沖區(qū)：",bufNo);for (int i=0;i<count;i++){str=str+QString::asprintf("%d, ",*data);data++;}str=str+'\n';ui->plainTextEdit->appendPlainText(str);
}

傳遞的指針型參數(shù)int*data 是一個(gè)數(shù)組指針，count 是緩沖區(qū)長(zhǎng)度。（此處注意主線程和子線程利用信號(hào)槽傳遞數(shù)組值的方法）

“啟動(dòng)線程”和“結(jié)束線程”兩個(gè)按鈕的代碼如下(省略了按鍵使能控制的代碼):

void Dialog::on_btnStopThread_clicked()
{//結(jié)束線程
//    threadConsumer.stopThread();//結(jié)束線程的run()函數(shù)執(zhí)行threadConsumer.terminate(); //因?yàn)閠hreadB可能處于等待狀態(tài)，所以用terminate強(qiáng)制結(jié)束threadConsumer.wait();//threadProducer.terminate();//結(jié)束線程的run()函數(shù)執(zhí)行threadProducer.wait();//ui->btnStartThread->setEnabled(true);ui->btnStopThread->setEnabled(false);
}void Dialog::on_btnStartThread_clicked()
{//啟動(dòng)線程threadConsumer.start();threadProducer.start();ui->btnStartThread->setEnabled(false);ui->btnStopThread->setEnabled(true);
}

啟動(dòng)線程時(shí)，先啟動(dòng) threadConsumer，再啟動(dòng) threadProducer，否則可能丟失第1個(gè)緩沖區(qū)的數(shù)據(jù)。
結(jié)束線程時(shí)，都采用 terminate()函數(shù)強(qiáng)制結(jié)束線程，因?yàn)閮蓚€(gè)線程之間有互鎖的關(guān)系，若不使用terminate()強(qiáng)制結(jié)束會(huì)出現(xiàn)線程無(wú)法結(jié)束的問(wèn)題。

程序運(yùn)行時(shí)的界面如圖 13-3 所示

從圖 13-3 可以看出，沒(méi)有出現(xiàn)丟失緩沖區(qū)或數(shù)據(jù)點(diǎn)的情況，兩個(gè)線程之間協(xié)調(diào)的很好，將QThreadDAQ:run()函數(shù)中模擬采樣率的延時(shí)時(shí)間調(diào)整為2秒也沒(méi)問(wèn)題(正常設(shè)置為50毫秒)。

在實(shí)際的數(shù)據(jù)采集中，要保證不丟失緩沖區(qū)或數(shù)據(jù)點(diǎn)，數(shù)據(jù)讀取線程的速度必須快過(guò)數(shù)據(jù)寫入緩沖區(qū)的線程的速度。

4. 源碼

4.1 可視化UI設(shè)計(jì)框架

4.2 qmythread.h

#ifndef QMYTHREAD_H
#define QMYTHREAD_H//#include    <QObject>
#include    <QThread>class QThreadDAQ : public QThread
{Q_OBJECTprivate:bool    m_stop=false; //停止線程
protected:void    run() Q_DECL_OVERRIDE;
public:QThreadDAQ();void    stopThread();
};class QThreadShow : public QThread
{Q_OBJECT
private:bool    m_stop=false; //停止線程
protected:void    run() Q_DECL_OVERRIDE;
public:QThreadShow();void    stopThread();
signals:void    newValue(int *data,int count, int seq);
};
#endif // QMYTHREAD_H

4.3 qmythread.cpp

#include    "qmythread.h"
#include    <QSemaphore>const int BufferSize = 8;
int buffer1[BufferSize];
int buffer2[BufferSize];
int curBuf=1; //當(dāng)前正在寫入的Bufferint bufNo=0; //采集的緩沖區(qū)序號(hào)quint8   counter=0;//數(shù)據(jù)生成器QSemaphore emptyBufs(2);//信號(hào)量：空的緩沖區(qū)個(gè)數(shù)，初始資源個(gè)數(shù)為2
QSemaphore fullBufs; //滿的緩沖區(qū)個(gè)數(shù)，初始資源為0QThreadDAQ::QThreadDAQ()
{}void QThreadDAQ::stopThread()
{m_stop=true;
}void QThreadDAQ::run()
{m_stop=false;//啟動(dòng)線程時(shí)令m_stop=falsebufNo=0;//緩沖區(qū)序號(hào)curBuf=1; //當(dāng)前寫入使用的緩沖區(qū)counter=0;//數(shù)據(jù)生成器int n=emptyBufs.available();if (n<2)  //保證 線程啟動(dòng)時(shí)emptyBufs.available==2emptyBufs.release(2-n);while(!m_stop)//循環(huán)主體{emptyBufs.acquire();//獲取一個(gè)空的緩沖區(qū)for(int i=0;i<BufferSize;i++) //產(chǎn)生一個(gè)緩沖區(qū)的數(shù)據(jù){if (curBuf==1)buffer1[i]=counter; //向緩沖區(qū)寫入數(shù)據(jù)elsebuffer2[i]=counter;counter++; //模擬數(shù)據(jù)采集卡產(chǎn)生數(shù)據(jù)msleep(50); //每50ms產(chǎn)生一個(gè)數(shù)}bufNo++;//緩沖區(qū)序號(hào)if (curBuf==1) // 切換當(dāng)前寫入緩沖區(qū)curBuf=2;elsecurBuf=1;fullBufs.release(); //有了一個(gè)滿的緩沖區(qū),available==1}quit();
}void QThreadShow::run()
{m_stop=false;//啟動(dòng)線程時(shí)令m_stop=falseint n=fullBufs.available();if (n>0)fullBufs.acquire(n); //將fullBufs可用資源個(gè)數(shù)初始化為0while(!m_stop)//循環(huán)主體{fullBufs.acquire(); //等待有緩沖區(qū)滿,當(dāng)fullBufs.available==0阻塞int bufferData[BufferSize];int seq=bufNo;if(curBuf==1) //當(dāng)前在寫入的緩沖區(qū)是1，那么滿的緩沖區(qū)是2for (int i=0;i<BufferSize;i++)bufferData[i]=buffer2[i]; //快速拷貝緩沖區(qū)數(shù)據(jù)elsefor (int i=0;i<BufferSize;i++)bufferData[i]=buffer1[i];emptyBufs.release();//釋放一個(gè)空緩沖區(qū)emit    newValue(bufferData,BufferSize,seq);//給主線程傳遞數(shù)據(jù)}quit();
}QThreadShow::QThreadShow()
{}void QThreadShow::stopThread()
{m_stop=true;
}

4.4 dialog.h

#ifndef DIALOG_H
#define DIALOG_H#include    <QDialog>
#include    <QTimer>#include    "qmythread.h"namespace Ui {
class Dialog;
}class Dialog : public QDialog
{Q_OBJECTprivate:QThreadDAQ   threadProducer;QThreadShow   threadConsumer;
protected:void    closeEvent(QCloseEvent *event);
public:explicit Dialog(QWidget *parent = 0);~Dialog();private slots:void    onthreadA_started();void    onthreadA_finished();void    onthreadB_started();void    onthreadB_finished();void    onthreadB_newValue(int *data, int count, int bufNo);void on_btnClear_clicked();void on_btnStopThread_clicked();void on_btnStartThread_clicked();private:Ui::Dialog *ui;
};#endif // DIALOG_H

4.5 dialog.cpp

#include "dialog.h"
#include "ui_dialog.h"void Dialog::closeEvent(QCloseEvent *event)
{//窗口關(guān)閉if (threadProducer.isRunning()){threadProducer.terminate();//結(jié)束線程的run()函數(shù)執(zhí)行threadProducer.wait();//}if (threadConsumer.isRunning()){threadConsumer.terminate(); //因?yàn)閠hreadB可能處于等待狀態(tài)，所以用terminate強(qiáng)制結(jié)束threadConsumer.wait();//}event->accept();
}Dialog::Dialog(QWidget *parent) :QDialog(parent),ui(new Ui::Dialog)
{ui->setupUi(this);connect(&threadProducer,SIGNAL(started()),this,SLOT(onthreadA_started()));connect(&threadProducer,SIGNAL(finished()),this,SLOT(onthreadA_finished()));connect(&threadConsumer,SIGNAL(started()),this,SLOT(onthreadB_started()));connect(&threadConsumer,SIGNAL(finished()),this,SLOT(onthreadB_finished()));connect(&threadConsumer,SIGNAL(newValue(int*,int,int)),this,SLOT(onthreadB_newValue(int*,int,int)));
}Dialog::~Dialog()
{delete ui;
}void Dialog::onthreadA_started()
{ui->LabA->setText("Thread Producer狀態(tài): started");
}void Dialog::onthreadA_finished()
{ui->LabA->setText("Thread Producer狀態(tài): finished");
}void Dialog::onthreadB_started()
{ui->LabB->setText("Thread Consumer狀態(tài): started");
}void Dialog::onthreadB_finished()
{ui->LabB->setText("Thread Consumer狀態(tài): finished");
}void Dialog::onthreadB_newValue(int *data, int count, int bufNo)
{ //讀取threadConsumer 傳遞的緩沖區(qū)的數(shù)據(jù)QString  str=QString::asprintf("第 %d 個(gè)緩沖區(qū)：",bufNo);for (int i=0;i<count;i++){str=str+QString::asprintf("%d, ",*data);data++;}str=str+'\n';ui->plainTextEdit->appendPlainText(str);
}void Dialog::on_btnClear_clicked()
{ui->plainTextEdit->clear();
}void Dialog::on_btnStopThread_clicked()
{//結(jié)束線程
//    threadConsumer.stopThread();//結(jié)束線程的run()函數(shù)執(zhí)行threadConsumer.terminate(); //因?yàn)閠hreadB可能處于等待狀態(tài)，所以用terminate強(qiáng)制結(jié)束threadConsumer.wait();//threadProducer.terminate();//結(jié)束線程的run()函數(shù)執(zhí)行threadProducer.wait();//ui->btnStartThread->setEnabled(true);ui->btnStopThread->setEnabled(false);
}void Dialog::on_btnStartThread_clicked()
{//啟動(dòng)線程threadConsumer.start();threadProducer.start();ui->btnStartThread->setEnabled(false);ui->btnStopThread->setEnabled(true);
}