Qt 多线程编程 #

┌─────────────────────────────────────────────────────────────┐
│                    Qt 多线程方式                             │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  QThread                                                    │
│  ├── 底层线程类                                             │
│  ├── 继承或 moveToThread                                    │
│  └── 最灵活但需要更多代码                                   │
│                                                             │
│  QtConcurrent                                               │
│  ├── 高级并发 API                                           │
│  ├── 自动管理线程池                                         │
│  └── 适合简单并行任务                                       │
│                                                             │
│  QThreadPool / QRunnable                                   │
│  ├── 线程池和可运行对象                                     │
│  ├── 复用线程资源                                           │
│  └── 适合大量短任务                                         │
│                                                             │
└─────────────────────────────────────────────────────────────┘

class WorkerThread : public QThread
{
    Q_OBJECT
public:
    WorkerThread(QObject *parent = nullptr) : QThread(parent) {}

protected:
    void run() override
    {
        // 在新线程中执行的代码
        for (int i = 0; i < 100; ++i) {
            if (isInterruptionRequested()) {
                break;  // 响应中断请求
            }
            
            emit progressChanged(i);
            QThread::msleep(100);
        }
        
        emit workFinished();
    }

signals:
    void progressChanged(int percent);
    void workFinished();
};

// 使用
WorkerThread *thread = new WorkerThread(this);
connect(thread, &WorkerThread::progressChanged, [](int percent) {
    qDebug() << "Progress:" << percent;
});
connect(thread, &WorkerThread::workFinished, []() {
    qDebug() << "Finished";
});
connect(thread, &WorkerThread::finished, thread, &QObject::deleteLater);

thread->start();  // 启动线程

// 停止线程
thread->requestInterruption();
thread->wait();  // 等待线程结束

class Worker : public QObject
{
    Q_OBJECT
public:
    explicit Worker(QObject *parent = nullptr) : QObject(parent) {}

public slots:
    void doWork(int parameter)
    {
        // 在工作线程中执行
        for (int i = 0; i < 100; ++i) {
            if (QThread::currentThread()->isInterruptionRequested()) {
                break;
            }
            
            emit progressChanged(i);
            QThread::msleep(100);
        }
        
        emit workFinished();
    }

signals:
    void progressChanged(int percent);
    void workFinished();
};

// 使用
class Controller : public QObject
{
    Q_OBJECT
public:
    Controller() {
        Worker *worker = new Worker;
        worker->moveToThread(&workerThread);
        
        connect(&workerThread, &QThread::finished, worker, &QObject::deleteLater);
        connect(this, &Controller::startWork, worker, &Worker::doWork);
        connect(worker, &Worker::progressChanged, this, &Controller::onProgress);
        connect(worker, &Worker::workFinished, this, &Controller::onFinished);
        
        workerThread.start();
    }
    
    ~Controller() {
        workerThread.requestInterruption();
        workerThread.quit();
        workerThread.wait();
    }
    
    void start() {
        emit startWork(42);
    }

signals:
    void startWork(int parameter);

private slots:
    void onProgress(int percent) {
        qDebug() << "Progress:" << percent;
    }
    
    void onFinished() {
        qDebug() << "Finished";
    }

private:
    QThread workerThread;
};

class Counter
{
public:
    void increment() {
        QMutexLocker locker(&m_mutex);  // RAII 方式
        ++m_value;
    }
    
    void decrement() {
        QMutexLocker locker(&m_mutex);
        --m_value;
    }
    
    int value() const {
        QMutexLocker locker(&m_mutex);
        return m_value;
    }

private:
    mutable QMutex m_mutex;
    int m_value = 0;
};

// 手动加锁
QMutex mutex;
mutex.lock();
// 临界区代码
mutex.unlock();

// 尝试加锁
if (mutex.tryLock()) {
    // 获取锁成功
    mutex.unlock();
}

class SharedData
{
public:
    QString read() const {
        QReadLocker locker(&m_lock);
        return m_data;
    }
    
    void write(const QString &data) {
        QWriteLocker locker(&m_lock);
        m_data = data;
    }

private:
    mutable QReadWriteLock m_lock;
    QString m_data;
};

// 生产者-消费者示例
const int BufferSize = 10;
QSemaphore freeSpace(BufferSize);  // 空闲空间
QSemaphore usedSpace(0);           // 已用空间

void producer()
{
    for (int i = 0; i < 100; ++i) {
        freeSpace.acquire();  // 等待空闲空间
        
        // 生产数据
        produceData(i);
        
        usedSpace.release();  // 增加已用空间
    }
}

void consumer()
{
    for (int i = 0; i < 100; ++i) {
        usedSpace.acquire();  // 等待数据
        
        // 消费数据
        consumeData();
        
        freeSpace.release();  // 增加空闲空间
    }
}

QMutex mutex;
QWaitCondition condition;
bool ready = false;

// 等待线程
void waitForCondition()
{
    QMutexLocker locker(&mutex);
    while (!ready) {
        condition.wait(&mutex);  // 释放锁并等待
    }
    // 条件满足，继续执行
}

// 唤醒线程
void setCondition()
{
    QMutexLocker locker(&mutex);
    ready = true;
    condition.wakeAll();  // 唤醒所有等待线程
    // 或 condition.wakeOne();  // 唤醒一个线程
}

#include <QtConcurrent>

// 在线程池中运行函数
QFuture<int> future = QtConcurrent::run([]() {
    // 耗时操作
    QThread::sleep(2);
    return 42;
});

// 等待结果
int result = future.result();
qDebug() << "Result:" << result;

// 检查是否完成
if (future.isFinished()) {
    qDebug() << "Finished";
}

// 使用 QFutureWatcher 监控
QFutureWatcher<int> *watcher = new QFutureWatcher<int>(this);
connect(watcher, &QFutureWatcher<int>::finished, []() {
    qDebug() << "Task finished";
});
connect(watcher, &QFutureWatcher<int>::resultReadyAt, [](int index) {
    qDebug() << "Result ready at" << index;
});

watcher->setFuture(future);

// 并行映射
QList<int> list = {1, 2, 3, 4, 5};

// 对每个元素应用函数
QFuture<void> future = QtConcurrent::map(list, [](int &value) {
    value *= 2;  // 修改原列表
});

future.waitForFinished();
qDebug() << list;  // {2, 4, 6, 8, 10}

// mapped - 返回新列表
QList<int> doubled = QtConcurrent::mapped(list, [](int value) {
    return value * 2;
});

// mappedReduced - 映射并归约
int sum = QtConcurrent::mappedReduced(list,
    [](int value) { return value * 2; },  // 映射函数
    [](int &result, int value) { result += value; }  // 归约函数
);

QList<int> list = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10};

// 过滤（修改原列表）
QFuture<void> future = QtConcurrent::filter(list, [](int value) {
    return value % 2 == 0;  // 保留偶数
});

future.waitForFinished();
qDebug() << list;  // {2, 4, 6, 8, 10}

// filtered - 返回新列表
QList<int> evens = QtConcurrent::filtered(list, [](int value) {
    return value % 2 == 0;
});

class MyRunnable : public QRunnable
{
public:
    MyRunnable(int id) : m_id(id) {
        setAutoDelete(true);  // 自动删除
    }
    
    void run() override {
        qDebug() << "Task" << m_id << "running in thread" 
                 << QThread::currentThread();
        QThread::sleep(1);
        qDebug() << "Task" << m_id << "finished";
    }

private:
    int m_id;
};

// 使用全局线程池
for (int i = 0; i < 10; ++i) {
    QThreadPool::globalInstance()->start(new MyRunnable(i));
}

// 等待所有任务完成
QThreadPool::globalInstance()->waitForDone();

// 自定义线程池
QThreadPool pool;
pool.setMaxThreadCount(4);  // 最大线程数

for (int i = 0; i < 10; ++i) {
    pool.start(new MyRunnable(i));
}

pool.waitForDone();

template<typename T>
class ThreadSafeQueue
{
public:
    void enqueue(const T &value) {
        QMutexLocker locker(&m_mutex);
        m_queue.enqueue(value);
        m_condition.wakeOne();
    }
    
    T dequeue() {
        QMutexLocker locker(&m_mutex);
        while (m_queue.isEmpty()) {
            m_condition.wait(&m_mutex);
        }
        return m_queue.dequeue();
    }
    
    bool isEmpty() const {
        QMutexLocker locker(&m_mutex);
        return m_queue.isEmpty();
    }

private:
    mutable QMutex m_mutex;
    QWaitCondition m_condition;
    QQueue<T> m_queue;
};

// 错误：在主线程执行耗时操作
void onButtonClicked() {
    heavyOperation();  // 阻塞 UI
}

// 正确：在工作线程执行
void onButtonClicked() {
    QtConcurrent::run([this]() {
        heavyOperation();
        QMetaObject::invokeMethod(this, "onOperationFinished",
                                  Qt::QueuedConnection);
    });
}

class WorkerThread : public QThread
{
public:
    ~WorkerThread() {
        requestInterruption();
        quit();
        wait();  // 等待线程结束
    }
    
protected:
    void run() override {
        while (!isInterruptionRequested()) {
            // 执行任务
        }
    }
};

// 跨线程信号槽自动使用队列连接
connect(worker, &Worker::dataReady, 
        receiver, &Receiver::processData,
        Qt::QueuedConnection);  // 数据会被复制