Реализация очереди задач с приоритетами на Qt

Введение

Когда фоновые задачи в приложении неравнозначны по важности — например, обработка пользовательского ввода важнее фоновой синхронизации с сервером — простая FIFO-очередь не подходит: важная задача может застрять за длинной серией менее значимых. В статье построим очередь задач с приоритетами на основе std::priority_queue, интегрированную с потоковой моделью Qt через QThread и QWaitCondition.

Концепция

Очередь с приоритетами хранит задачи, упорядоченные не по времени добавления, а по заданному приоритету — при извлечении всегда возвращается задача с наивысшим приоритетом среди ожидающих. Реализация на основе std::priority_queue внутри потокобезопасной обёртки, аналогичной классическому producer-consumer паттерну (статья про QWaitCondition), даёт рабочий поток, который всегда обрабатывает наиболее приоритетную из доступных задач.

Пример кода

#pragma once
#include <QObject>
#include <QThread>
#include <QMutex>
#include <QWaitCondition>
#include <functional>
#include <queue>
#include <vector>

struct PrioritizedTask
{
    int priority; // больше значение — выше приоритет
    std::function<void()> task;

    bool operator<(const PrioritizedTask &other) const
    {
        return priority < other.priority; // priority_queue — max-heap по умолчанию
    }
};

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

    ~PriorityTaskQueue() override
    {
        stop();
        wait();
    }

    void enqueue(int priority, std::function<void()> task)
    {
        QMutexLocker locker(&m_mutex);
        m_queue.push({priority, std::move(task)});
        m_condition.wakeOne();
    }

    void stop()
    {
        QMutexLocker locker(&m_mutex);
        m_stopping = true;
        m_condition.wakeAll();
    }

protected:
    void run() override
    {
        while (true) {
            std::function<void()> taskToRun;
            {
                QMutexLocker locker(&m_mutex);
                while (m_queue.empty() && !m_stopping) {
                    m_condition.wait(&m_mutex);
                }
                if (m_stopping && m_queue.empty()) {
                    return;
                }
                taskToRun = std::move(const_cast<PrioritizedTask &>(m_queue.top()).task);
                m_queue.pop();
            }
            taskToRun(); // выполняется за пределами блокировки мьютекса
        }
    }

private:
    QMutex m_mutex;
    QWaitCondition m_condition;
    std::priority_queue<PrioritizedTask> m_queue;
    bool m_stopping = false;
};

// Пример использования
#include "prioritytaskqueue.h"
#include <QCoreApplication>
#include <QDebug>

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

    PriorityTaskQueue queue;
    queue.start();

    queue.enqueue(1, []() { qDebug() << "Низкий приоритет: фоновая синхронизация"; });
    queue.enqueue(10, []() { qDebug() << "Высокий приоритет: обработка пользовательского ввода"; });
    queue.enqueue(5, []() { qDebug() << "Средний приоритет: обновление кеша"; });

    QThread::msleep(500);
    queue.stop();
    queue.wait();

    return 0;
}

Пояснения к коду

PrioritizedTask оборачивает задачу (std::function<void()>) вместе с числовым приоритетом и определяет operator<, используемый std::priority_queue для упорядочивания — поскольку priority_queue по умолчанию реализует max-heap, задача с наибольшим значением priority всегда оказывается на вершине и извлекается первой через top().

PriorityTaskQueue наследует QThread и переопределяет run(): рабочий поток в цикле ждёт (через QWaitCondition, не расходуя CPU впустую) появления задач в очереди, извлекает задачу с наивысшим приоритетом и выполняет её уже за пределами блокировки мьютекса — это важно, чтобы выполнение самой задачи не блокировало добавление новых задач из других потоков через enqueue().

Подводные камни

Голодание задач с низким приоритетом (starvation). Если задачи с высоким приоритетом поступают непрерывным потоком, задачи с низким приоритетом могут никогда не быть выполнены — для систем, где это недопустимо, нужен механизм возрастания приоритета «застрявших» задач со временем (aging) или гарантированная квота на низкоприоритетные задачи.
const_cast для извлечения задачи из top() — std::priority_queue::top() возвращает константную ссылку, и для перемещения std::function из неё (вместо копирования) применён const_cast, что технически безопасно только потому, что элемент сразу удаляется через pop() и больше не используется в исходном виде — без этой гарантии const_cast для модификации был бы недопустим.
Исключение внутри выполняемой задачи, не перехваченное самой задачей, приведёт к нераспространённому исключению из run() потока, что в Qt обычно означает завершение приложения — задачи, переданные в очередь, должны сами обрабатывать ожидаемые исключения, если они в принципе могут возникать.
Отсутствие отмены уже поставленной в очередь задачи. Реализация в примере не поддерживает удаление конкретной задачи из очереди после её добавления — если такая функциональность нужна (например, отмена устаревшего запроса), задачу нужно дополнительно обернуть в объект с флагом «отменено», проверяемым перед фактическим выполнением.