Qt之Concurrent框架

簡述

QtConcurrent 命名空間提供了高級 API，使得能夠在不使用低級線程原語（例如：互斥、讀寫鎖、等待條件或信號量）的狀況下編寫多線程程序，使用 QtConcurrent 編寫的程序根據可用的處理器核心數自動調整所使用的線程數。這意味着，當在將來部署多核系統時，如今編寫的應用程序將繼續適應。

• 簡述
• 用法
• Qt Concurrent
• 單詞統計
• 更多參考

用法

在 C++ API changes 有關於 Qt Concurrent 的更改說明：

Qt Concurrent has been moved from Qt Core to its own module

意思是說，Qt Concurrent 已經被從 Qt Core 中移到本身的模塊中了。因此，要連接到 Qt Concurrent 模塊，須要在 qmake 項目文件中添加：

QT += concurrent

注意： QtConcurrent::Exception 類被重命名爲 QException，而且 QtConcurrent::UnhandledException 類被重命名爲 QUnhandledException，他們仍然位於 Qt Core 中。

Qt Concurrent

QtConcurrent 包含了函數式編程風格 APIs 用於並行列表處理，包括用於共享內存（非分佈式）系統的 MapReduce 和 FilterReduce 實現，以及用於管理 GUI 應用程序異步計算的類：

• Concurrent Map 和 Map-Reduce

  • QtConcurrent::map()：將一個函數應用於一個容器中的每一項，就地修改 items。
  • QtConcurrent::mapped()：和 map() 相似，只是它返回一個包含修改內容的新容器。
  • QtConcurrent::mappedReduced()：和 mapped() 相似，只是修改後的結果減小或組合成一個單一的結果。

• Concurrent Filter 和 Filter-Reduce

  • QtConcurrent::filter()：從一個容器中刪除全部 items，基於一個 filter 函數的結果。
  • QtConcurrent::filtered()：和 filter() 相似，只是它返回一個包含過濾內容的新容器。
  • QtConcurrent::filteredReduced()：和 filtered() 相似，只是過濾後的結果減小或組合成一個單一的結果。

• Concurrent Run

  • QtConcurrent::run()：在另外一個線程中運行一個函數。

• QFuture：表示異步計算的結果

• QFutureIterator：容許經過 QFuture 遍歷可用的結果
• QFutureWatcher：容許使用信號槽來監控一個 QFuture
• QFutureSynchronizer：是一個方便的類，用於一些 QFutures 的自動同步

Qt Concurrent 支持多種兼容 STL 的容器和迭代器類型，可是最好使用具備隨機訪問迭代器的 Qt 容器，例如：QList 或 QVector。map 和 filter 函數都接受容器和 begin/end 迭代器。

STL 迭代器支持概述：

迭代器類型	示例類	支持狀態
Input Iterator		不支持
Output Iterator		不支持
Forward Iterator	std::slist	支持
Bidirectional Iterator	QLinkedList, std::list	支持
Random Access Iterator	QList, QVector, std::vector	支持和推薦

在 Qt Concurrent 迭代大量輕量級 items 的狀況下，隨機訪問迭代器能夠更快，由於它們容許跳過容器中的任何點。此外，使用隨機訪問迭代器容許 Qt Concurrent 經過 QFuture::progressValue() 和 QFutureWatcher::progressValueChanged() 來提供進度信息。

非就地修改的函數（例如：mapped() 和 filtered()），在調用時會建立容器的副本。若是正在使用的是 STL 容器，此複製操做可能須要一段時間，在這種狀況下，建議爲容器指定 begin 和 end 迭代器。

單詞統計

厲害了 Concurrent，來看一個單詞統計的示例：

#include <QList>
#include <QMap>
#include <QTextStream>
#include <QString>
#include <QStringList>
#include <QDir>
#include <QTime>
#include <QApplication>
#include <QDebug>

#include <qtconcurrentmap.h>

using namespace QtConcurrent;

// 遞歸搜索文件
QStringList findFiles(const QString &startDir, QStringList filters)
{
    QStringList names;
    QDir dir(startDir);

    foreach (QString file, dir.entryList(filters, QDir::Files))
        names += startDir + '/' + file;

    foreach (QString subdir, dir.entryList(QDir::AllDirs | QDir::NoDotAndDotDot))
        names += findFiles(startDir + '/' + subdir, filters);
    return names;
}

typedef QMap<QString, int> WordCount;

// 單線程單詞計數器函數
WordCount singleThreadedWordCount(QStringList files)
{
    WordCount wordCount;
    foreach (QString file, files) {
        QFile f(file);
        f.open(QIODevice::ReadOnly);
        QTextStream textStream(&f);
        while (textStream.atEnd() == false)
            foreach (const QString &word, textStream.readLine().split(' '))
                wordCount[word] += 1;

    }
    return wordCount;
}

// countWords 計算單個文件的單詞數，該函數由多個線程並行調用，而且必須是線程安全的。
WordCount countWords(const QString &file)
{
    QFile f(file);
    f.open(QIODevice::ReadOnly);
    QTextStream textStream(&f);
    WordCount wordCount;

    while (textStream.atEnd() == false)
        foreach (const QString &word, textStream.readLine().split(' '))
            wordCount[word] += 1;

    return wordCount;
}

// reduce 將 map 的結果添加到最終結果，該函數只能由一個線程一次調用。
void reduce(WordCount &result, const WordCount &w)
{
    QMapIterator<QString, int> i(w);
    while (i.hasNext()) {
        i.next();
        result[i.key()] += i.value();
    }
}

int main(int argc, char** argv)
{
    QApplication app(argc, argv);
    qDebug() << "finding files...";
    QStringList files = findFiles("../../", QStringList() << "*.cpp" << "*.h");
    qDebug() << files.count() << "files";

    int singleThreadTime = 0;
    {
        QTime time;
        time.start();
        // 單線程統計，與 mapreduce 機制實現的做對比
        WordCount total = singleThreadedWordCount(files);
        singleThreadTime = time.elapsed();
        // 打印出所耗費的時間
        qDebug() << "single thread" << singleThreadTime;
    }

    int mapReduceTime = 0;
    {
        QTime time;
        time.start();
        // mappedReduced 方式進行統計
        WordCount total = mappedReduced(files, countWords, reduce);
        mapReduceTime = time.elapsed();
        qDebug() << "MapReduce" << mapReduceTime;
    }
    // 輸出 mappedReduced 方式比單線程處理方式要快的倍數
    qDebug() << "MapReduce speedup x" << ((double)singleThreadTime - (double)mapReduceTime) / (double)mapReduceTime + 1;
}

輸出以下：

finding files…
2185 files
single thread 5221
MapReduce 2256
MapReduce speedup x 2.31427

能夠看出，共查找了 2185 個文件，單線程使用了 5221 毫秒，MapReduce 方式使用了 2256 毫秒，比單線程要快 2.31427 倍。通過反覆嘗試，基本都在 2 倍以上。

更多參考

• Qt之Concurrent Map和Map-Reduce
• Qt之Concurrent Filter和Filter-Reduce
• Qt之Concurrent Run
• Qt之QFuture
• Qt之QFutureWatcher