Очереди задач в C++: реализация, применение и оптимизация

Работа с очередями задач в C++: от простого к сложному

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

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

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

Рассмотрим базовый пример синхронизированной очереди с использованием мьютекса и условной переменной:

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

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

    void push(Task task) {
        std::lock_guard<std::mutex> lock(mutex_);
        queue_.push(std::move(task));
        cond_var_.notify_one();
    }

    Task pop() {
        std::unique_lock<std::mutex> lock(mutex_);
        cond_var_.wait(lock, [this]() { return !queue_.empty(); });
        Task task = std::move(queue_.front());
        queue_.pop();
        return task;
    }

    bool empty() const {
        std::lock_guard<std::mutex> lock(mutex_);
        return queue_.empty();
    }

private:
    std::queue<Task> queue_;
    mutable std::mutex mutex_;
    std::condition_variable cond_var_;
};

В этом коде push() добавляет задачу и уведомляет один ожидающий поток. pop() блокирует поток, пока очередь не станет непустой. Такой подход решает проблему синхронизации, но не масштабируется на множество потребителей без доработок.

2. Паттерн Producer-Consumer

Классическая задача producer-consumer (производитель-потребитель) идеально ложится на очередь задач. Один или несколько потоков-производителей генерируют задания, а потоки-потребители их выполняют. Важно правильно управлять размером очереди, чтобы избежать переполнения или простоя.

Пример с несколькими потребителями:

#include <thread>
#include <vector>
#include <iostream>

void worker(TaskQueue& queue, int id) {
    while (true) {
        auto task = queue.pop();
        if (!task) break;  // сигнал завершения
        std::cout << "Worker " << id << " executing task\n";
        task();
    }
}

int main() {
    TaskQueue queue;
    std::vector<std::thread> workers;

    // Запускаем 4 потока-потребителя
    for (int i = 0; i < 4; ++i) {
        workers.emplace_back(worker, std::ref(queue), i);
    }

    // Производитель добавляет задачи
    for (int i = 0; i < 10; ++i) {
        queue.push([i]() {
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
            std::cout << "Task " << i << " done\n";
        });
    }

    // Сигнал завершения (nullptr задача)
    for (auto& w : workers) {
        queue.push(nullptr);
    }

    for (auto& w : workers) {
        w.join();
    }
    return 0;
}

Здесь каждый потребитель получает задачу, выполняет её и ждёт следующую. Сигнал завершения (nullptr) останавливает цикл. Такой паттерн часто используют в серверных приложениях, обработчиках событий и игровых движках.

3. Thread Pool — пул потоков для очереди задач

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

Реализация простого thread pool на C++17:

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

class ThreadPool {
public:
    explicit ThreadPool(size_t numThreads) {
        for (size_t i = 0; i < numThreads; ++i) {
            workers_.emplace_back([this]() {
                while (true) {
                    std::function<void()> 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();
                }
            });
        }
    }

    template<class F, class... Args>
    auto enqueue(F&& f, Args&&... args) 
        -> std::future<typename std::invoke_result_t<F, Args...>> 
    {
        using return_type = typename std::invoke_result_t<F, Args...>;
        auto task = std::make_shared<std::packaged_task<return_type()>>(
            std::bind(std::forward<F>(f), std::forward<Args>(args)...)
        );
        std::future<return_type> result = task->get_future();
        {
            std::lock_guard<std::mutex> lock(mutex_);
            if (stop_) throw std::runtime_error("enqueue on stopped ThreadPool");
            tasks_.emplace([task]() { (*task)(); });
        }
        condition_.notify_one();
        return result;
    }

    ~ThreadPool() {
        {
            std::lock_guard<std::mutex> lock(mutex_);
            stop_ = true;
        }
        condition_.notify_all();
        for (auto& worker : workers_) {
            worker.join();
        }
    }

private:
    std::vector<std::thread> workers_;
    std::queue<std::function<void()>> tasks_;
    std::mutex mutex_;
    std::condition_variable condition_;
    bool stop_ = false;
};

Использование пула:

int main() {
    ThreadPool pool(4);
    std::vector<std::future<int>> results;

    for (int i = 0