Очереди задач для C++: реализация и использование

Очереди задач для C++: от теории к практике

Очередь задач (task queue) — один из ключевых паттернов многопоточного программирования на C++. Он позволяет эффективно распределять работу между потоками, избегая гонок данных и излишнего потребления ресурсов. В этой статье мы разберём, как создавать и использовать очереди задач в C++ с акцентом на производительность и безопасность.

Зачем нужна очередь задач?

В современных приложениях часто возникает необходимость выполнять множество небольших задач параллельно. Прямое создание потоков для каждой задачи приводит к накладным расходам: создание потока занимает время и память. Очередь задач решает эту проблему через пул потоков (thread pool):

• Повторное использование потоков — потоки не уничтожаются после выполнения задачи, а возвращаются в пул.
• Балансировка нагрузки — задачи распределяются равномерно.
• Управление ресурсами — ограничение количества одновременно работающих потоков.

Базовая реализация очереди задач на C++

Рассмотрим простую, но потокобезопасную очередь задач с использованием std::queue и мьютексов. Для этого мы используем стандартную библиотеку C++11 и выше.

#include <queue>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <thread>
#include <vector>

class TaskQueue {
public:
    using Task = std::function<void()>;

    explicit TaskQueue(size_t numThreads) : stop(false) {
        for (size_t i = 0; i < numThreads; ++i) {
            workers.emplace_back([this] { workerLoop(); });
        }
    }

    ~TaskQueue() {
        {
            std::unique_lock<std::mutex> lock(mutex);
            stop = true;
        }
        condition.notify_all();
        for (auto& worker : workers) {
            if (worker.joinable()) {
                worker.join();
            }
        }
    }

    void enqueue(Task task) {
        {
            std::unique_lock<std::mutex> lock(mutex);
            tasks.push(std::move(task));
        }
        condition.notify_one();
    }

private:
    void workerLoop() {
        while (true) {
            Task task;
            {
                std::unique_lock<std::mutex> lock(mutex);
                condition.wait(lock, [this] { return stop || !tasks.empty(); });
                if (stop && tasks.empty()) {
                    return;
                }
                task = std::move(tasks.front());
                tasks.pop();
            }
            task(); // Выполнение задачи
        }
    }

    std::queue<Task> tasks;
    std::mutex mutex;
    std::condition_variable condition;
    bool stop;
    std::vector<std::thread> workers;
};

Пример использования

Теперь покажем, как использовать наш класс TaskQueue для параллельного выполнения простых вычислений:

#include <iostream>

int main() {
    TaskQueue tq(4); // Пул из 4 потоков

    for (int i = 0; i < 10; ++i) {
        tq.enqueue([i] {
            std::cout << "Задача " << i << " выполнена потоком "
                      << std::this_thread::get_id() << std::endl;
        });
    }

    // Деструктор tq дождётся завершения всех задач
    return 0;
}

Продвинутые возможности: очередь с приоритетами

Иногда требуется обрабатывать задачи в порядке их важности. Для этого можно использовать std::priority_queue вместо обычной очереди. Добавим поддержку приоритетов:

#include <queue>
#include <functional>

class PriorityTaskQueue {
public:
    using Task = std::function<void()>;

    struct PrioritizedTask {
        int priority;
        Task task;

        bool operator<(const PrioritizedTask& other) const {
            return priority < other.priority; // Высший приоритет — первым
        }
    };

    void enqueue(Task task, int priority = 0) {
        std::unique_lock<std::mutex> lock(mutex);
        tasks.push({priority, std::move(task)});
        condition.notify_one();
    }

    // ... остальная логика аналогична TaskQueue, но с std::priority_queue
private:
    std::priority_queue<PrioritizedTask> tasks;
    // ...
};

Советы по оптимизации

• Избегайте блокировок — используйте lock-free структуры (например, moodycamel::ConcurrentQueue) для высоконагруженных систем.
• Настройте размер пула — оптимальное число потоков равно std::thread::hardware_concurrency().
• Используйте move-семантику — передавайте задачи через std::move, чтобы избежать копирования.
• Разделяйте задачи по ти