Очереди задач в C++: практическое руководство с примерами

Введение: зачем нужны очереди задач в C++?

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

В C++ стандартная библиотека предоставляет мощный контейнер std::queue, который идеально подходит для реализации очередей задач. Однако на практике требуется не просто хранить элементы, а безопасно передавать их между потоками, обрабатывать ошибки и управлять жизненным циклом задач. В этой статье мы разберем, как создавать очереди задач на C++ с нуля, используя современные возможности языка (C++11 и новее).

Мы рассмотрим:

• Базовую работу с std::queue
• Потокобезопасную очередь задач
• Реализацию пула потоков (thread pool)
• Приоритетные очереди

1. Базовая очередь задач с std::queue

Начнем с простого примера. std::queue — это контейнер-адаптер, который работает по принципу FIFO (First In, First Out). Он предоставляет методы push(), pop(), front() и empty().

Вот как можно организовать очередь задач, где каждая задача — это функция (или вызываемый объект):

#include <iostream>
#include <queue>
#include <functional>

int main() {
    // Очередь задач, где каждая задача — это std::function<void()>
    std::queue<std::function<void()>> taskQueue;

    // Добавляем задачи
    taskQueue.push([]() { std::cout << "Задача 1 выполнена\n"; });
    taskQueue.push([]() { std::cout << "Задача 2 выполнена\n"; });
    taskQueue.push([]() { std::cout << "Задача 3 выполнена\n"; });

    // Обрабатываем очередь
    while (!taskQueue.empty()) {
        auto task = taskQueue.front();
        task();  // Выполняем задачу
        taskQueue.pop();  // Удаляем выполненную задачу
    }

    return 0;
}

Этот код работает, но только в однопоточном режиме. Если несколько потоков будут одновременно добавлять или извлекать задачи, возникнет состояние гонки. Для многопоточности нужна синхронизация.

2. Потокобезопасная очередь задач

Чтобы сделать очередь безопасной для многопоточного доступа, мы используем мьютекс (std::mutex) и условную переменную (std::condition_variable). Это классический паттерн producer-consumer (производитель-потребитель).

Реализация потокобезопасной очереди:

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

class ThreadSafeQueue {
private:
    std::queue<std::function<void()>> queue_;
    std::mutex mutex_;
    std::condition_variable cond_;
    bool stop_ = false;

public:
    void push(std::function<void()> task) {
        std::lock_guard<std::mutex> lock(mutex_);
        queue_.push(task);
        cond_.notify_one();  // Уведомляем один ожидающий поток
    }

    std::function<void()> pop() {
        std::unique_lock<std::mutex> lock(mutex_);
        cond_.wait(lock, [this]() { return !queue_.empty() || stop_; });
        if (stop_ && queue_.empty()) {
            return nullptr;  // Сигнал остановки
        }
        auto task = queue_.front();
        queue_.pop();
        return task;
    }

    void stop() {
        std::lock_guard<std::mutex> lock(mutex_);
        stop_ = true;
        cond_.notify_all();  // Будим все потоки для завершения
    }
};

Ключевые моменты:

• std::lock_guard используется для простой блокировки при добавлении.
• std::unique_lock необходим для condition_variable::wait(), так как он может временно отпускать блокировку.
• Условная переменная позволяет потоку-потребителю ждать появления новой задачи, не тратя процессорное время на пустые проверки.
• Метод stop() корректно завершает работу всех потоков.

3. Пул потоков (Thread Pool) на основе очереди задач

Пул потоков — это набор заранее созданных рабочих потоков, которые ждут задачи из очереди. Это гораздо эффективнее, чем создавать новый поток под каждую задачу, так как создание потоков — дорогая операция.

Реализация простого пула потоков:

class ThreadPool {
private:
    std::vector<std::thread> workers_;
    ThreadSafeQueue queue_;
    bool stop_ = false;

public:
    ThreadPool(size_t num_threads) {
        for (size_t i = 0; i < num_threads; ++i) {
            workers_.emplace_back([this]() {
                while (true) {
                    auto task = queue_.pop();
                    if (!task) break;  // Сигнал остановки
                    task();
                }
            });
        }
    }

    void enqueue(std::function<void()> task) {
        queue_.push(task);
    }

    ~ThreadPool() {
        queue_.stop();
        for (auto &worker : workers_) {
            if (worker.joinable()) {
                worker.join();
            }
        }
    }
};

Как это работает:

1. При создании пула запускается N рабочих потоков.
2. Каждый поток вызывает queue_.pop(), который блокируется, пока не появится задача.
3. Основной поток (или любой другой) добавляет задачи через enqueue().
4. При уничтожении пула вызывается queue_.stop(), все потоки получают nullptr и завершаются.

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

int main() {
    ThreadPool pool(4);  // 4 рабочих потока

    for (int i = 0; i < 10; ++i) {
        pool.enqueue([i]() {
            std::cout << "Задача " << i << " выполняется в потоке "
                      << std::this_thread::get_id() << "\n