1. 信號量的原理

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

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

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

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

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

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

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

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

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

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

• WC.tryAcquire(1),有一個人嘗試進(jìn)去，但是因為沒有位置了，他不等待，走了,tryAcquire()函數(shù)返回 false:

• WC.acquire()，有一個人必須進(jìn)去，但是因為沒有位置了，他就一直在外面等著，直到有其他人出來，空余出位置來。

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

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

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

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

在實例 samp13_5中，創(chuàng)建類似于 Producer/Consumer 模型的兩個線程類 QThreadDAQ 和QThreadShow。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

QThreadDAQ 是數(shù)據(jù)采集線程，例如在使用數(shù)據(jù)采集卡進(jìn)行連續(xù)數(shù)據(jù)采集時，需要一個單獨的線程將采集卡采集的數(shù)據(jù)讀取到緩沖區(qū)內(nèi)。
QThreadShow 是數(shù)據(jù)讀取線程,用于讀取已存滿數(shù)據(jù)的緩沖區(qū)中的數(shù)據(jù)并傳遞給主線程顯示,采用信號與槽機制與主線程交互。
QThreadDAQ/QThreadShow 類的定義與使用 QWaitCondition 的實例 samp13_4中的QThreadProducer/QThreadConsumer 類的定義類似，只是QThreadShow 的信號 newValue()采用了指針作為傳遞參數(shù)，用于一次傳遞出一個緩沖區(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ū)序號quint8   counter=0;//數(shù)據(jù)生成器QSemaphore emptyBufs(2);//信號量：空的緩沖區(qū)個數(shù)，初始資源個數(shù)為2
QSemaphore fullBufs; //滿的緩沖區(qū)個數(shù)，初始資源為0QThreadDAQ::QThreadDAQ()
{}void QThreadDAQ::stopThread()
{m_stop=true;
}void QThreadDAQ::run()
{m_stop=false;//啟動線程時令m_stop=falsebufNo=0;//緩沖區(qū)序號curBuf=1; //當(dāng)前寫入使用的緩沖區(qū)counter=0;//數(shù)據(jù)生成器int n=emptyBufs.available();if (n<2)  //保證 線程啟動時emptyBufs.available==2emptyBufs.release(2-n);while(!m_stop)//循環(huán)主體{emptyBufs.acquire();//獲取一個空的緩沖區(qū)for(int i=0;i<BufferSize;i++) //產(chǎn)生一個緩沖區(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)生一個數(shù)}bufNo++;//緩沖區(qū)序號if (curBuf==1) // 切換當(dāng)前寫入緩沖區(qū)curBuf=2;elsecurBuf=1;fullBufs.release(); //有了一個滿的緩沖區(qū),available==1}quit();
}void QThreadShow::run()
{m_stop=false;//啟動線程時令m_stop=falseint n=fullBufs.available();if (n>0)fullBufs.acquire(n); //將fullBufs可用資源個數(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();//釋放一個空緩沖區(qū)emit    newValue(bufferData,BufferSize,seq);//給主線程傳遞數(shù)據(jù)}quit();
}QThreadShow::QThreadShow()
{}void QThreadShow::stopThread()
{m_stop=true;
}

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

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

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

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

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

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

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

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

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

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

3. QThreadDAQ和QThreadShow 的使用

設(shè)計窗口基于 QDialog 應(yīng)用程序 samp13_5，對話框的類定義如下(省略了一些不重要的或與前面實例重復(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類定義了兩個線程的實例，threadProducer 和 threadConsumer。

自定義了一個槽函數(shù) onthreadB_newValue()，用于與 threadConsumer 的信號關(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()的功能就是讀取一個緩沖區(qū)里的數(shù)據(jù)并顯示，其實現(xiàn)代碼如下

void Dialog::onthreadB_newValue(int *data, int count, int bufNo)
{ //讀取threadConsumer 傳遞的緩沖區(qū)的數(shù)據(jù)QString  str=QString::asprintf("第 %d 個緩沖區(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 是一個數(shù)組指針，count 是緩沖區(qū)長度。（此處注意主線程和子線程利用信號槽傳遞數(shù)組值的方法）

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

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

啟動線程時，先啟動 threadConsumer，再啟動 threadProducer，否則可能丟失第1個緩沖區(qū)的數(shù)據(jù)。
結(jié)束線程時，都采用 terminate()函數(shù)強制結(jié)束線程，因為兩個線程之間有互鎖的關(guān)系，若不使用terminate()強制結(jié)束會出現(xiàn)線程無法結(jié)束的問題。

程序運行時的界面如圖 13-3 所示

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

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

4. 源碼

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

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ū)序號quint8   counter=0;//數(shù)據(jù)生成器QSemaphore emptyBufs(2);//信號量：空的緩沖區(qū)個數(shù)，初始資源個數(shù)為2
QSemaphore fullBufs; //滿的緩沖區(qū)個數(shù)，初始資源為0QThreadDAQ::QThreadDAQ()
{}void QThreadDAQ::stopThread()
{m_stop=true;
}void QThreadDAQ::run()
{m_stop=false;//啟動線程時令m_stop=falsebufNo=0;//緩沖區(qū)序號curBuf=1; //當(dāng)前寫入使用的緩沖區(qū)counter=0;//數(shù)據(jù)生成器int n=emptyBufs.available();if (n<2)  //保證 線程啟動時emptyBufs.available==2emptyBufs.release(2-n);while(!m_stop)//循環(huán)主體{emptyBufs.acquire();//獲取一個空的緩沖區(qū)for(int i=0;i<BufferSize;i++) //產(chǎn)生一個緩沖區(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)生一個數(shù)}bufNo++;//緩沖區(qū)序號if (curBuf==1) // 切換當(dāng)前寫入緩沖區(qū)curBuf=2;elsecurBuf=1;fullBufs.release(); //有了一個滿的緩沖區(qū),available==1}quit();
}void QThreadShow::run()
{m_stop=false;//啟動線程時令m_stop=falseint n=fullBufs.available();if (n>0)fullBufs.acquire(n); //將fullBufs可用資源個數(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();//釋放一個空緩沖區(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(); //因為threadB可能處于等待狀態(tài)，所以用terminate強制結(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 個緩沖區(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(); //因為threadB可能處于等待狀態(tài)，所以用terminate強制結(jié)束threadConsumer.wait();//threadProducer.terminate();//結(jié)束線程的run()函數(shù)執(zhí)行threadProducer.wait();//ui->btnStartThread->setEnabled(true);ui->btnStopThread->setEnabled(false);
}void Dialog::on_btnStartThread_clicked()
{//啟動線程threadConsumer.start();threadProducer.start();ui->btnStartThread->setEnabled(false);ui->btnStopThread->setEnabled(true);
}