Qt基于等待条件QWaitCondition实现的任务队列模型示例

核心概念

Qt中的QWaitCondition是一个用于多线程同步的类，允许线程在某些条件满足时唤醒其他等待的线程。它通常与QMutex配合使用，协调线程之间的执行顺序，适用于生产者-消费者模型、任务队列调度等场景。

​wait()：使当前线程进入等待状态，并释放关联的互斥锁。当条件满足时，线程被唤醒并重新获取锁。
​wakeOne()：唤醒一个正在等待的线程（按操作系统调度策略选择）。
​wakeAll()：唤醒所有正在等待的线程。

常见用法

**​线程等待条件：**线程检查条件，若不满足则调用wait()进入等待。
**​条件通知：**另一线程修改条件后，调用wakeOne()或wakeAll()通知等待线程。
**​配合QMutex：**所有条件检查和修改必须在QMutex保护下进行，避免竞态条件。

任务队列描述

**​任务队列：**主线程将任务添加到队列，多个工作线程从队列中获取并执行任务。
**​同步机制：**当队列为空时，工作线程等待；当新任务到达时，唤醒一个线程处理。
**​优雅退出：**支持安全终止所有工作线程。

代码实现

#include <QCoreApplication>
#include <QThread>
#include <QMutex>
#include <QWaitCondition>
#include <QQueue>
#include <QDebug>
#include <functional>// 定义任务类型（使用std::function包装可调用对象）
using Task = std::function<void()>;// 线程安全的任务队列
class TaskQueue {
public:void addTask(const Task& task) {QMutexLocker locker(&m_mutex);m_queue.enqueue(task);m_cond.wakeOne(); // 唤醒一个等待线程}Task getTask() {QMutexLocker locker(&m_mutex);//离开作用域（即析构）时自动对m_mutex解锁// 使用while防止虚假唤醒while (m_queue.isEmpty() && !m_stop) {m_cond.wait(&m_mutex); // 自动释放锁并等待，被唤醒（即wait返回）之后自动对m_mutex重新上锁}if (m_stop) return nullptr; // 终止信号return m_queue.dequeue();}void stop() {QMutexLocker locker(&m_mutex);m_stop = true;m_cond.wakeAll(); // 唤醒所有线程退出}private:QMutex m_mutex;QWaitCondition m_cond;QQueue<Task> m_queue;bool m_stop = false;
};// 工作线程：不断从队列中取任务执行
class Worker : public QThread {
public:Worker(TaskQueue* queue) : m_queue(queue) {}void run() override {while (true) {Task task = m_queue->getTask();if (!task) break; // 收到终止信号task(); // 执行任务}qDebug() << "Worker thread" << QThread::currentThread() << "stopped.";}private:TaskQueue* m_queue;
};int main(int argc, char* argv[]) {QCoreApplication a(argc, argv);// 创建任务队列和4个工作线程TaskQueue queue;QList<Worker*> workers;for (int i = 0; i < 4; ++i) {Worker* worker = new Worker(&queue);worker->start();workers.append(worker);}// 添加10个任务到队列for (int i = 0; i < 10; ++i) {queue.addTask([i]() {qDebug() << "Task" << i << "processed by" << QThread::currentThread();QThread::msleep(100); // 模拟任务耗时});}// 等待所有任务执行完毕QThread::sleep(2);// 停止所有工作线程queue.stop();for (Worker* worker : workers) {worker->wait();delete worker;}qDebug() << "All workers stopped.";return 0;
}

运行代码输出

Task 0 processed by QThread(0x1087708)
Task 3 processed by QThread(0x10879d0)
Task 1 processed by QThread(0x10878d0)
Task 2 processed by QThread(0x1087800)
Task 4 processed by QThread(0x1087800)
Task 5 processed by QThread(0x10878d0)
Task 7 processed by QThread(0x10879d0)
Task 6 processed by QThread(0x1087708)
Task 8 processed by QThread(0x10878d0)
Task 9 processed by QThread(0x1087800)
Worker thread QThread(0x1087708) stopped.
Worker thread QThread(0x10879d0) stopped.
Worker thread QThread(0x1087800) stopped.
Worker thread QThread(0x10878d0) stopped.
All workers stopped.

代码解析

1. ​任务队列（TaskQueue类）​

​线程安全操作：
通过QMutex保护任务队列（m_queue）和停止标志（m_stop）。

条件等待：
getTask()中使用while (m_queue.isEmpty() && !m_stop)防止虚假唤醒。
当队列为空且未收到停止信号时，调用m_cond.wait(&m_mutex)释放锁并等待。

终止机制：
stop()方法设置m_stop = true并通过wakeAll()唤醒所有线程。
工作线程收到nullptr任务时退出循环。

2. ​工作线程（Worker类）​

**​任务循环：**持续调用getTask()获取任务，直到收到终止信号。
**​任务执行：**直接调用task()执行实际逻辑（如I/O操作、计算等）。

3. ​主线程逻辑

**​创建线程池：**启动4个工作线程监听任务队列。
**​动态添加任务：**通过Lambda表达式生成任务，支持任意类型的操作。
​优雅退出：
调用queue.stop()通知所有线程停止。
使用worker->wait()确保线程安全退出后清理资源。

关键特性

**​动态任务分配：**支持任意数量、类型的任务。
**​高效唤醒策略：**wakeOne()确保每次新任务只唤醒一个线程，减少竞争。
**​资源安全：**通过RAII（QMutexLocker）自动管理锁，避免死锁。
**​跨平台：**代码在Windows/Linux/macOS上行为一致。

应用场景扩展

​**Web服务器：**将HTTP请求作为任务分配给线程池处理。
**​批量数据处理：**多个线程并行处理文件、数据库操作。
**​GUI程序后台任务：**保持界面响应性，耗时操作放入任务队列。

通过这种模式，可以轻松实现高并发任务处理，同时避免手动管理线程的复杂性。