Qt多线程

本文最后更新于：2024年2月14日 晚上

Qt多线程

  Qt多线程的开发目的是为了保证在进行耗时操作时，可以处理用户的其他输入输出。比如，如果在UI线程里面进行耗时操作，界面会不响应用户操作。以及提升程序性能。现在的电脑一般都是多核CPU，多线程并行处理事务，可以大大提升程序的性能。

Qt信号槽的一些事 | 渡世白玉 (dushibaiyu.com)

QtConcurrent 线程使用说明 - 知乎 (zhihu.com)

Qt之QtConcurrent-CSDN博客

一、多线程

(1)信号与槽的连接方式

五种连接方式

1
2
3
4
5
6
7
enum ConnectionType {
	AutoConnection,				//默认
	DirectConnection,			//立即调用
	QueuedConnection,			//放在接收者队列
	BlockingQueuedConnection,	//与放在接收者队列，同时阻塞发送者
	UniqueConnection = 0x80		//标志位，避免重复链接
};

1.连接方式一：AutoConnection

  一般信号槽不会写第五个参数，其实使用的默认值，使用这个值则连接类型会在信号发送时决定。

  • 如果接收者和发送者在同一个线程，则自动使用Qt::DirectConnection类型。
  • 如果接收者和发送者不在一个线程，则自动使用Qt::QueuedConnection类型。

2.连接方式二：DirectConnection

  函数会在信号发送的时候直接被调用，槽函数运行于信号发送者所在线程，有点类似于回调函数。效果看上去就像是直接在信号发送位置调用了槽函数。这个在多线程环境下比较危险，可能会造成崩溃。

3.连接方式三：QueuedConnection

  槽函数在控制回到接收者所在线程的事件循环时被调用，槽函数运行于信号接收者所在线程。发送信号之后，槽函数不会立刻被调用，等到接收者的当前函数执行完毕，并进入信号接收者的事件循环后，槽函数才会被调用。多线程环境下一般用这个。

4.连接方式四：BlockingQueuedConnection

  槽函数的调用时机与Qt::QueuedConnection一致，不过发送完信号后发送者所在线程会阻塞，直到槽函数运行完。使用该连接方式，接收者和发送者绝对不能在一个线程，否则程序会死锁！！！在多线程间需要同步的场合可能需要这个。

4.连接方式五：UniqueConnection

  严格上来说相比上面四种连接方式，它并不算新的连接方式，而是用于修饰上面的四种连接方式。他实现的效果就是避免重复连接，因为Qt的信号槽是可以同一个信号和槽函数重复多次连接。这种通常都会是只执行一次就好，那就通过这个标志位进行修饰，达到多次连接（实际上也只是连接了一次）也只调用一次槽函数的效果。用(Qt::ConnectionType|Qt::UniqueConnection)来修饰。

(2)创建多线程的两种方式

1. 继承QThread

  继承 QThread 是创建线程的一个普通方法，其中创建的线程只有 run()方法在线程里的，其他类内定义的方法都在主线程内。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
class WorkerThread : public QThread {
    Q_OBJECT
    void run() override {
        QString result;
        /* ... 耗时或阻塞操作 ... */
        emit resultReady( result );
    }
signals:
    void resultReady( const QString& s );
};

void MyObject::startWorkInAThread() {
    WorkerThread* workerThread = new WorkerThread( this );
    connect( workerThread, &WorkerThread::resultReady, this, &MyObject::handleResults );
    connect( workerThread, &WorkerThread::finished, workerThread, &QObject::deleteLater );
    workerThread->start();
}
  • run函数在新线程中执行，run函数执行结束，线程结束。
  • WorkerThread实例化的对象属于创建他的线程，而不是run函数所在线程。
  • WorkerThread没有事件循环，除非在run()函数中调用exec();
  • 队列连接到WorkerThread的slot函数，slot函数在创建WorkerThread对象的线程中执行。
  • 直接调用WorkerThread的方法，该方法的执行线程为调用处的线程。

2. 继承QObject

  继承 QObject 类的方法更加灵活，它通过QObject::moveToThread()函数，将一个 QObeject的类转移到一个线程里执行。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
class Worker : public QObject {
    Q_OBJECT

public slots:
    void doWork( const QString& parameter ) {  // 工作槽函数
        QString result;
        /* ... 耗时或阻塞操作 ... */
        emit resultReady( result );  // 结果完成信号
    }
signals:
    void resultReady( const QString& result );  // 结果完成信号
};

class Controller : public QObject {
    Q_OBJECT
    QThread workerThread;

public:
    Controller() {
        Worker* worker = new Worker;
        worker->moveToThread( &workerThread );                                        // 移入线程
        connect( &workerThread, &QThread::finished, worker, &QObject::deleteLater );  // 线程的结束信号 绑定 工作对象的销毁槽函数
        connect( this, &Controller::operate, worker, &Worker::doWork );               // 控制对象的操作信号 绑定 工作对象的工作槽函数
        connect( worker, &Worker::resultReady, this, &Controller::handleResults );    // 工作对象的结果完成信号 绑定 控制对象的结果处理槽函数
        workerThread.start();                                                         // 线程开始
    }
    ~Controller() {
        workerThread.quit();
        workerThread.wait();
    }
public slots:
    void handleResults( const QString& );  // 槽函数 处理结果函数
signals:
    void operate( const QString& );  // 信号 操作信号
};
  • 调用moveToThread函数的对象不能设置父对象。
  • Worker类中的槽函数可以跟任意线程的任意信号建立连接，队列连接时，在新线程中执行。
  • 直接调用Worker类中的函数，在调用线程内执行。
  • 同时发送多个与Worker类中槽函数连接的信号，槽函数依次执行。

二、线程池QThreadPool

  线程池是一种线程使用模式，它管理着一组可重用的线程，可以处理分配过来的可并发执行的任务。线程池设有最大线程数，可以避免线程数过多会导致额外的线程切换开销。线程池管理的线程具有可重用性，可以减少创建和销毁线程的次数。它的主要目的是减少程序员编写的重复代码，提高程序的效率和性能，在高并发的项目中会用到。

(1)单线程代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
#include <QDebug>
#include <QElapsedTimer>

int main( int argc, char* argv[] ) {
    QString string;
    string    = "1000101010";                 // 创建字符串
    string    = string.repeated( 10000000 );  // 重复10000000次
    int count = 0;                            // 计数'1'

    QElapsedTimer timer;
    timer.start();  // 开始计时

    for ( int i = 0, len = string.length(); i < len; ++i ) {
        if ( string[ i ] == '1' ) {
            count++;
        }
    }

    qDebug() << "spend time: " << timer.elapsed() << "ms, count: " << count;  // 统计耗时
}

(2)线程池代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
#include <QDebug>
#include <QElapsedTimer>
#include <QMutex>
#include <QThreadPool>

class Counter : public QRunnable {
public:
    Counter( int index, int len, QString* string, QMutex* mutex, int* count ) {  // 初始化函数
        m_index  = index;
        m_length = len;
        m_string = string;
        m_mutex  = mutex;
        m_count  = count;
    }

private:
    void run() {
        int count = 0;
        for ( int i = m_index, end = m_index + m_length; i < end; ++i ) {
            if ( ( *m_string )[ i ] == '1' ) {
                count++;
            }
        }

        /* 多线程操作共享变量需要加锁 */
        m_mutex->lock();
        *m_count += count;
        m_mutex->unlock();
    }

    // 成员变量
    int       m_index;
    int       m_length;
    QString*  m_string;
    QMutex*   m_mutex;
    int*      m_count;
};

int main( int argc, char* argv[] ) {
    QString string;
    string    = "1000101010";
    string    = string.repeated( 10000000 );
    int count = 0;

    QElapsedTimer timer;
    timer.start();  // 开始计时

    auto pool = QThreadPool::globalInstance();  // 创建全局线程池对象
    pool->setMaxThreadCount( 8 );               // 设置最多8个线程

    QMutex mutex;
    /* 分割成100块，分别计算 */
    for ( int i = 0, span = string.length() / 100; i < 100; i++ ) {
        Counter* a = new Counter( i * span, span, &string, &mutex, &count );  // 每个计数类对象进行计数
        a->setAutoDelete( true );                                             // 执行完后自动释放
        pool->start( a );
    }
    pool->waitForDone();  // 等待线程全部完成

    qDebug() << "spend time: " << timer.elapsed() << "ms, count: " << count;  // 耗时统计
}
  • 默认情况下，run函数执行完，线程对象会被线程池自动删除。可以使用setAutoDelete函数设置。
  • QThreadPool::start()多次启动设置为autoDelete的QRunnable对象，可能导致崩溃。

三、并发编程QtConcurrent

  QtConcurrent提供了高级api，使编写多线程程序时，不需要使用诸如互斥锁、读写锁、等待条件或信号量等低级线程安全类。每调用一次QtConcurrent::run()函数，就新建立一个线程。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
#include <QApplication>
#include <QDebug>
#include <QString>
#include <QThread>
#include "qtconcurrentrun.h"
class A {
public:
    A() {
        std::cout << "我是 A 的构造函数" << std::endl;
    };

    QString m_func( QString name ) {
        m_name = name;
        std::cout << "我是 A 的成员函数" << std::endl;
    }
    QString m_name;
};

int main( int argc, char** argv ) {
    QApplication          app( argc, argv );
    QByteArray            bytearray = "hello world";
    A*                    a         = new A();
    QFuture< QString >    fut1      = QtConcurrent::run( bytearray, &QByteArray::split, ',' );    // 1.调用QByteArray的常量成员函数split(),传递常量引用，bytearray
    QFuture< QByteArray > fut2      = QtConcurrent::run( a, &A::m_func, QString( "Thread 2" ) );  // 2.非常量成员函数运行在一个新的线程，传递指针

    QString result1 = fut1.result();
    QString result2 = fut2.result();

    fut1.waitForFinished();
    fut2.waitForFinished();
}
  • Concurrent run，在线程池内起一个线程来执行一个函数。

(1)并行处理一批数据map

1. 更改原数据QtConcurrent::map

1
2
3
4
5
6
7
8
9
10
11
12
void toUpper( QString& str ) {
    str = str.toUpper();
}
QStringList strWords;
strWords << "Apple"
         << "Banana"
         << "cow"
         << "dog"
         << "Egg";
auto future = QtConcurrent::map( strWords, toUpper );
future.waitForFinished();
// strWords = {"APPLE", "BANANA", "COW", "DOG", "EGG"}

2. 不改变原数据，返回处理结果QtConcurrent::map

1
2
3
4
5
6
7
8
9
10
11
12
13
QString toUpper( const QString& str ) {
    return str.toUpper();
}
QStringList strWords;
strWords << "Apple"
         << "Banana"
         << "cow"
         << "dog"
         << "Egg";
auto future = QtConcurrent::mapped( strWords, toUpper );
future.waitForFinished();
qDebug() << future.results();
// 输出：("APPLE", "BANANA", "COW", "DOG", "EGG")

3. 处理后的结果还需要进行处理QtConcurrent::mappedReduced

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
QString toUpper( const QString& str ) {
    return str.toUpper();
}
void reduceFun( QList< QString >& dictionary, const QString& string ) {
    dictionary.push_back( QString( "result: " ) + string );
}

QStringList strWords;
strWords << "Apple"
         << "Banana"
         << "cow"
         << "dog"
         << "Egg";
auto future = QtConcurrent::mappedReduced( strWords, toUpper, reduceFun );
future.waitForFinished();
qDebug() << future.result();
// 输出：("result: BANANA", "result: APPLE", "result: COW", "result: DOG", "result: EGG")
  • 注意，上述代码输出结果的顺序与原数据的顺序已经不一样了。mappedReduced 函数的逻辑是：启动多个线程执行toUper 对链表中的每个元素进行处理，处理后的结果再逐一交给reduceFun函数处理。reduceFun 并不会等toUper的所有线程执行完毕后才开始执行，并且同一时刻，只有一个 reduceFun 线程在执行。如果需要使最后的输出结果顺序与输入相一致，就要用mappedReduced 函数的第四个参数，赋值为QtConcurrent::OrderedReduce， 即可保证顺序一致。

(2)过滤操作一批数据filter

  一般用于对一批数据的过滤操作。同样也包含filter, filtered, filteredReduce 三种用法。与Concurrent::map 类似，filter 函数会改变原始数据，filtered 函数将处理结果保存在filtered 函数的返回值中，filteredReduce 会将过滤后的数据再调用reduceFun 函数处理。这里就不再进行详细赘述，可以完全参考map函数的用法使用。