编程进阶网
杨充
2025-08-14
目录

并发与多线程

# 02.并发与多线程

# 目录介绍

# 2.1 QT线程核心

# 2.1.1 QT线程介绍

Qt 提供了强大的线程管理功能，支持多线程编程。Qt 的线程管理基于 QThread 类，同时提供了高级 API（如 QtConcurrent）来简化多线程编程。

# 2.1.2 线程管理类

  • QThread：平台无关的线程类
  • QThreadPool：线程池管理
  • QRunnable：可在线程池中运行的任务基类
  • QFuture/QFutureWatcher：异步计算模型
  • QPromise：配合 QFuture 使用（Qt 6）

# 2.1.3 同步原语类

  • QMutex/QMutexLocker：互斥锁
  • QReadWriteLock：读写锁
  • QSemaphore：信号量
  • QWaitCondition：条件变量
  • QAtomic*：原子操作类

# 2.2 线程管理基础

# 2.2.1 QThread类

QThread 是 Qt 中用于管理线程的类。每个 QThread 对象代表一个线程，可以通过 start() 启动线程。

QThread 的 run() 方法是线程的入口点，默认实现调用 exec() 进入事件循环。

# 2.2.2 线程与事件循环

Qt 的事件循环（QEventLoop）是线程的核心，用于处理信号与槽、定时器、网络事件等。

默认情况下，QThread 会启动一个事件循环。

# 2.2.3 线程安全

Qt 提供了线程安全的信号与槽机制，支持跨线程通信。 使用 Qt::QueuedConnection 实现跨线程的信号与槽连接。

# 2.3 QThread创建

# 2.3.1 继承QThread

通过继承 QThread 并重写 run() 方法来实现自定义线程逻辑。

#include <QThread>
#include <QDebug>

class MyThread : public QThread {
    Q_OBJECT
protected:
    void run() override {
        qDebug() << "Thread started";
        // 线程逻辑
        QThread::sleep(2);
        qDebug() << "Thread finished";
    }
};

int main(int argc, char *argv[]) {
    QCoreApplication app(argc, argv);
    MyThread thread;
    thread.start(); // 启动线程
    thread.wait();  // 等待线程结束
    return app.exec();
}

# 2.3.2 moveToThread

将 QObject 对象移动到另一个线程，利用信号与槽机制实现线程逻辑。

#include <QThread>
#include <QDebug>

class Worker : public QObject {
    Q_OBJECT
public slots:
    void doWork() {
        qDebug() << "Worker started";
        QThread::sleep(2);
        qDebug() << "Worker finished";
    }
};

int main(int argc, char *argv[]) {
    QCoreApplication app(argc, argv);
    QThread thread;
    Worker worker;
    worker.moveToThread(&thread); // 将 worker 移动到新线程
    QObject::connect(&thread, &QThread::started, &worker, &Worker::doWork);
    thread.start(); // 启动线程
    return app.exec();
}

# 2.4 多线程编程

QtConcurrent 提供了高级 API，用于简化多线程编程，如并行计算和异步任务。

# 2.4.1 并行计算

使用 QtConcurrent::run() 在另一个线程中执行函数。

#include <QtConcurrent>
#include <QDebug>

void myFunction() {
    qDebug() << "Function running in thread";
    QThread::sleep(2);
}

int main(int argc, char *argv[]) {
    QCoreApplication app(argc, argv);
    QFuture<void> future = QtConcurrent::run(myFunction);
    future.waitForFinished(); // 等待任务完成
    return app.exec();
}

# 2.4.2 并行映射

使用 QtConcurrent::map() 对容器中的元素并行处理。

#include <QtConcurrent>
#include <QList>
#include <QDebug>

void square(int &value) {
    value *= value;
}

int main(int argc, char *argv[]) {
    QCoreApplication app(argc, argv);
    QList<int> list = {1, 2, 3, 4, 5};
    QtConcurrent::map(list, square); // 对列表中的每个元素平方
    qDebug() << list; // 输出: (1, 4, 9, 16, 25)
    return app.exec();
}

# 2.5 线程同步

Qt 提供了多种同步原语（Synchronization Primitives），用于管理多线程之间的并发访问和协调。这些同步原语包括互斥锁、读写锁、信号量、条件变量等。

  • QMutex：用于互斥访问共享资源。
  • QReadWriteLock：支持多读单写的锁。
  • QSemaphore：控制对共享资源的访问数量。
  • QWaitCondition：用于线程间的条件等待和唤醒。

# 2.5.1 互斥锁QMutex

QMutex 用于保护共享资源，防止多个线程同时访问。基本用法

#include <QMutex>
#include <QThread>
#include <QDebug>

QMutex mutex;
int counter = 0;

class MyThread : public QThread {
protected:
    void run() override {
        for (int i = 0; i < 1000; ++i) {
            mutex.lock();   // 加锁
            ++counter;
            mutex.unlock(); // 解锁
        }
    }
};

int main(int argc, char *argv[]) {
    QCoreApplication app(argc, argv);
    MyThread thread1, thread2;
    thread1.start();
    thread2.start();
    thread1.wait();
    thread2.wait();
    qDebug() << "Counter:" << counter; // 输出: 2000
    return app.exec();
}

QMutexLocker，QMutexLocker 是一个 RAII（资源获取即初始化）类，用于自动管理 QMutex 的加锁和解锁。

QMutex mutex;
int counter = 0;

class MyThread : public QThread {
protected:
    void run() override {
        for (int i = 0; i < 1000; ++i) {
            QMutexLocker locker(&mutex); // 自动加锁和解锁
            ++counter;
        }
    }
};

# 2.5.2 QReadWriteLock

QReadWriteLock 允许多个线程同时读取共享资源，但写操作是独占的。基本用法：

#include <QReadWriteLock>
#include <QThread>
#include <QDebug>

QReadWriteLock rwLock;
int data = 0;

class ReaderThread : public QThread {
protected:
    void run() override {
        rwLock.lockForRead(); // 加读锁
        qDebug() << "Read data:" << data;
        rwLock.unlock();      // 解读锁
    }
};

class WriterThread : public QThread {
protected:
    void run() override {
        rwLock.lockForWrite(); // 加写锁
        ++data;
        qDebug() << "Write data:" << data;
        rwLock.unlock();       // 解写锁
    }
};

int main(int argc, char *argv[]) {
    QCoreApplication app(argc, argv);

    ReaderThread reader;
    WriterThread writer;
    reader.start();
    writer.start();
    reader.wait();
    writer.wait();

    return app.exec();
}

QReadLocker 和 QWriteLocker，类似于 QMutexLocker，QReadLocker 和 QWriteLocker 是 RAII 类，用于自动管理 QReadWriteLock 的加锁和解锁。

QReadWriteLock rwLock;
int data = 0;

class ReaderThread : public QThread {
protected:
    void run() override {
        QReadLocker locker(&rwLock); // 自动加读锁
        qDebug() << "Read data:" << data;
    }
};

class WriterThread : public QThread {
protected:
    void run() override {
        QWriteLocker locker(&rwLock); // 自动加写锁
        ++data;
        qDebug() << "Write data:" << data;
    }
};

# 2.5.3 QSemaphore

QSemaphore 用于控制对共享资源的访问数量。基本用法

#include <QSemaphore>
#include <QThread>
#include <QDebug>

QSemaphore semaphore(1); // 初始值为 1

class WorkerThread : public QThread {
protected:
    void run() override {
        semaphore.acquire(); // 获取信号量
        qDebug() << "Working...";
        QThread::sleep(2);
        semaphore.release(); // 释放信号量
    }
};

int main(int argc, char *argv[]) {
    QCoreApplication app(argc, argv);

    WorkerThread thread1, thread2;
    thread1.start();
    thread2.start();
    thread1.wait();
    thread2.wait();

    return app.exec();
}

# 2.5.4 QWaitCondition

QWaitCondition 用于线程间的条件等待和唤醒。基本用法

#include <QWaitCondition>
#include <QMutex>
#include <QThread>
#include <QDebug>

QMutex mutex;
QWaitCondition condition;
bool ready = false;

class Producer : public QThread {
protected:
    void run() override {
        QThread::sleep(2);
        mutex.lock();
        ready = true;
        condition.wakeAll(); // 唤醒等待的线程
        mutex.unlock();
    }
};

class Consumer : public QThread {
protected:
    void run() override {
        mutex.lock();
        while (!ready) {
            condition.wait(&mutex); // 等待条件满足
        }
        mutex.unlock();
        qDebug() << "Data is ready!";
    }
};

int main(int argc, char *argv[]) {
    QCoreApplication app(argc, argv);

    Producer producer;
    Consumer consumer;
    producer.start();
    consumer.start();
    producer.wait();
    consumer.wait();

    return app.exec();
}

# 2.5.5 QAtomic

# 2.X 线程安全模式详解

模式 机制 适用
QMutex 互斥锁 保护共享数据
QReadWriteLock 读写锁 读多写少
QSemaphore 信号量 生产者-消费者
QWaitCondition 条件变量 等待复杂条件

# QReadWriteLock

QReadWriteLock rwLock;
QList<SensorData> data;

void readData(int idx) {
    QReadLocker locker(&rwLock);
    return data.at(idx);  // 多个读锁可并发
}

void writeData(const SensorData& d) {
    QWriteLocker locker(&rwLock);
    data.append(d);        // 写锁独占——阻塞所有读写
}

# QWaitCondition 生产者-消费者

QMutex mutex;
QWaitCondition notEmpty;
QList<Data> buffer;

void produce(const Data& d) {
    QMutexLocker lock(&mutex);
    buffer.append(d);
    notEmpty.wakeOne();
}

void consume() {
    QMutexLocker lock(&mutex);
    while (buffer.isEmpty())
        notEmpty.wait(&mutex);    // 释放锁→休眠→被唤醒→重获锁
    Data d = buffer.takeFirst();
}

# 2.Y QtConcurrent 并行计算

对线程池的高级封装——适合数据并行：

#include <QtConcurrent>

// map——并行转换
QList<QImage> images = loadAll();
QFuture<QImage> future = QtConcurrent::mapped(images, processFrame);

// filter——并行过滤
auto result = QtConcurrent::filtered(allFrames, [](const Frame& f) {
    return f.signalStrength > 50;
});

// run——后台执行
QFuture<void> bg = QtConcurrent::run([]() { heavyComputation(); });

选择表：长期后台服务→QThread；一次性大数据→QtConcurrent；短暂异步→QtConcurrent::run

# 2.6 线程池与 QThreadPool

# 2.6.1 QThreadPool 原理

QThreadPool 管理一组可复用的线程——避免频繁创建/销毁线程的开销：

// 全局线程池——Qt 自动创建
QThreadPool* pool = QThreadPool::globalInstance();
pool->setMaxThreadCount(4);  // 最多 4 个并发线程

// 提交任务
class RenderTask : public QRunnable {
public:
    RenderTask(const QImage& img) : m_image(img) {
        setAutoDelete(true);  // 执行完自动 delete
    }
    void run() override {
        QImage processed = processImage(m_image);
        emit finished(processed);
    }
    QImage m_image;
};

pool->start(new RenderTask(image));

QThreadPool 内部结构：

QThreadPool:
  ├── 任务队列 (QQueue<QRunnable*>)  ← 先进先出
  ├── 线程数组 (QList<QThread*>)     ← 最大 maxThreadCount 个
  ├── 活跃线程数: activeThreadCount
  └── 策略: 创建线程直到 maxThreadCount
       → 任务结束后线程休眠 30s
       → 30s 无新任务 → 线程退出（回收）

# 2.6.2 QFuture 与 QPromise（Qt 6）

Qt 6 引入 QPromise——可以直接从工作线程报告进度和结果：

#include <QPromise>

// 使用 QtConcurrent::run 的增强版
void heavyComputation(QPromise<int>& promise) {
    for (int i = 0; i < 100; ++i) {
        if (promise.isCanceled()) return;  // 支持取消
        doWork(i);
        promise.setProgressValue(i + 1);   // 报告进度
    }
    promise.addResult(42);                 // 报告结果
    promise.finish();
}

// 调用侧
QFuture<int> future = QtConcurrent::run(&heavyComputation);

connect(&m_watcher, &QFutureWatcher<int>::progressValueChanged,
        this, [this](int progress) {
    m_progressBar->setValue(progress);    // 实时更新进度条
});
m_watcher.setFuture(future);

// 取消
future.cancel();

# 2.6.3 QFutureSynchronizer

管理多个并发任务的生命周期：

QFutureSynchronizer<void> sync;

sync.addFuture(QtConcurrent::run(task1));
sync.addFuture(QtConcurrent::run(task2));
sync.addFuture(QtConcurrent::run(task3));

// 析构时自动 waitForFinished()——同步器
// sync.waitForFinished();  // 显式等待

# 2.7 原子操作与无锁编程

# 2.7.1 QAtomicInteger / QAtomicInt

QAtomicInteger<int> counter(0);

// 原子加法
counter.fetchAndAddRelaxed(1);   // 返回旧值（无内存屏障）
counter.fetchAndAddAcquire(1);   // 获取语义
counter.fetchAndAddRelease(1);   // 释放语义

// 原子比较交换（CAS）
int expected = 0;
if (counter.testAndSetRelaxed(expected, 1)) {
    // 交换成功——counter 从 0 变成 1
}

// 内存序含义：
// Relaxed: 只保证原子性，不保证对其他变量的可见性顺序
// Acquire: 之后的读/写不会被重排到此操作之前
// Release: 之前的读/写不会被重排到此操作之后

# 2.7.2 无锁栈（lock-free stack）

template<typename T>
class LockFreeStack {
    struct Node {
        T data;
        QAtomicPointer<Node> next;
    };

    QAtomicPointer<Node> m_head;

public:
    void push(const T& value) {
        Node* node = new Node{value};
        node->next.storeRelaxed(m_head.loadRelaxed());

        // CAS 循环——无锁的核心
        while (!m_head.testAndSetRelease(node->next.loadRelaxed(), node)) {
            node->next.storeRelaxed(m_head.loadRelaxed());  // 重试
        }
    }

    T pop() {
        Node* head;
        do {
            head = m_head.loadAcquire();
            if (!head) throw std::runtime_error("Empty");
        } while (!m_head.testAndSetRelease(head, head->next.loadRelaxed()));

        T value = head->data;
        delete head;
        return value;
    }
};

# 2.8 线程安全模式总结

# 2.8.1 同步原语选择指南

场景 推荐同步原语 理由
简单互斥 QMutex + QMutexLocker 最常见
多读少写 QReadWriteLock 多个读锁可并发
限制并发数 QSemaphore 生产者-消费者
等待复杂条件 QWaitCondition 等待+唤醒模式
简单计数器 QAtomicInt 无锁，最快
动态创建线程 QThreadPool 复用，避免创建/销毁开销

# 2.8.2 死锁模式与避免

// ❌ 死锁模式 1：锁序不一致
// 线程 1: lock(A) → lock(B)
// 线程 2: lock(B) → lock(A)  ← 死锁！

// ✅ 解决：统一锁序——总是先锁 A 再锁 B

// ❌ 死锁模式 2：BlockingQueuedConnection 同线程
connect(sender, &Sender::sig, sender, &Receiver::slot,
        Qt::BlockingQueuedConnection);  // 等自己执行完 → 永远等不到！

// ❌ 死锁模式 3：持有锁时调用 wait()
m_mutex.lock();
m_workerThread.wait();  // 等待线程结束 → 但线程可能在等这个锁！
m_mutex.unlock();

# 2.8.3 线程模式选择

长期后台循环服务 → QThread + moveToThread + 信号槽通信
  └── 例: 串口/网络持续监听

一次性并行计算 → QtConcurrent::run / mapped
  └── 例: 批量图片处理、数据转换

定时任务 → QTimer（在主线程或工作线程）
  └── 例: 心跳、传感器轮询

生产者-消费者 → QSemaphore + QWaitCondition
  └── 例: 传感器数据缓冲 → 批量写入磁盘

# 2.9 综合案例

# 案例：嵌入式设备的多线程数据处理管线

// === 管线架构 ===
// 传感器采集(线程1) → 数据滤波(线程2) → 数据分析(线程3) → UI更新(主线程)
//     ↓ 100Hz          ↓ 100Hz           ↓ 10Hz            ↓ 10Hz

class DataPipeline : public QObject {
    Q_OBJECT

public:
    DataPipeline() {
        // 创建处理线程
        m_filterThread = new QThread(this);
        m_analyzeThread = new QThread(this);

        m_filter->moveToThread(m_filterThread);
        m_analyzer->moveToThread(m_analyzeThread);

        m_filterThread->start();
        m_analyzeThread->start();

        // === 管线连接 ===
        // 采集 → 滤波：高频，用 QueuedConnection
        connect(m_sensor, &Sensor::rawData, m_filter, &Filter::onRawData,
                Qt::QueuedConnection);

        // 滤波 → 分析：降频，批处理
        connect(m_filter, &Filter::filteredData, m_analyzer, &Analyzer::onFilteredData,
                Qt::QueuedConnection);

        // 分析 → UI：主线程
        connect(m_analyzer, &Analyzer::resultReady, this, &DataPipeline::updateUI,
                Qt::QueuedConnection);
    }

    ~DataPipeline() {
        m_filterThread->quit();
        m_analyzeThread->quit();
        m_filterThread->wait();
        m_analyzeThread->wait();
    }

public slots:
    void updateUI(const AnalysisResult& result) {
        m_tempLabel->setText(QString::number(result.avgTemp, 'f', 1) + "°C");
        m_statusLabel->setText(result.alert ? "⚠️ Alert" : "OK");
    }

private:
    Sensor* m_sensor;
    Filter* m_filter;
    Analyzer* m_analyzer;
    QThread* m_filterThread;
    QThread* m_analyzeThread;
};

# 2.10 速查表

类 用途 关键方法
QThread 线程管理 start(), quit(), wait(), exec()
QThreadPool 线程池 start(), setMaxThreadCount(), waitForDone()
QRunnable 线程池任务 run(), setAutoDelete()
QFuture 异步结果 result(), waitForFinished(), cancel()
QFutureWatcher 监控异步任务 setFuture(), progressValueChanged()
QPromise (Qt 6) 结果报告 addResult(), setProgressValue(), finish()
QMutex 互斥锁 lock(), unlock(), tryLock()
QMutexLocker RAII 锁 构造加锁，析构解锁
QReadWriteLock 读写锁 lockForRead(), lockForWrite()
QSemaphore 信号量 acquire(), release()
QWaitCondition 条件变量 wait(), wakeOne(), wakeAll()
QAtomicInt 原子操作 fetchAndAdd(), testAndSet()
QtConcurrent::run 异步执行 run(fn), run(fn, promise)
QtConcurrent::mapped 并行映射 mapped(container, fn)

多线程铁律：

  1. GUI 操作必须在主线程——用 QueuedConnection 或 QMetaObject::invokeMethod
  2. moveToThread 优于继承 QThread::run——更清晰的所有权
  3. 使用 QMutexLocker / QReadLocker / QWriteLocker——RAII 防止忘记解锁
  4. BlockingQueuedConnection 两端不能在同一线程——否则死锁
  5. QAtomic 操作注意内存序——默认 Relaxed 不保证可见性
  6. 线程池适合短期任务，长期运行用独立 QThread
  7. QObject::deleteLater() 是线程安全的——跨线程删除对象的推荐方式
上次更新: 2026/07/02, 11:21:43
核心功能基础
文件与IO系统

← 文件与IO系统→

最近更新
01
audit
07-27
02
C++入门教程全章思考题汇编
07-24
03
12.技术团队建设能力
07-21
更多文章>
Theme by Vdoing | Copyright © 2019-2026 杨充 | MIT License | 鄂ICP备2024073355号-1 | 鄂ICP备2024073355号