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程序后台任务：**保持界面响应性，耗时操作放入任务队列。

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