Aufgabenwarteschlangen für C++: Implementierung und Nutzung

Aufgabenwarteschlangen für C++: Von der Theorie zur Praxis

Eine Aufgabenwarteschlange (Task Queue) ist eines der Schlüsselmuster der Multithread-Programmierung in C++. Sie ermöglicht es, Arbeit effizient auf Threads zu verteilen, Datenrennen und übermäßigen Ressourcenverbrauch zu vermeiden. In diesem Artikel werden wir uns ansehen, wie man Aufgabenwarteschlangen in C++ erstellt und verwendet, mit Fokus auf Leistung und Sicherheit.

Wozu eine Aufgabenwarteschlange?

In modernen Anwendungen besteht oft die Notwendigkeit, viele kleine Aufgaben parallel auszuführen. Das direkte Erstellen von Threads für jede Aufgabe führt zu Overhead: Das Erstellen eines Threads kostet Zeit und Speicher. Eine Aufgabenwarteschlange löst dieses Problem durch einen Thread-Pool:

• Wiederverwendung von Threads — Threads werden nach der Ausführung einer Aufgabe nicht zerstört, sondern kehren in den Pool zurück.
• Lastausgleich — Aufgaben werden gleichmäßig verteilt.
• Ressourcenverwaltung — Begrenzung der Anzahl gleichzeitig laufender Threads.

Basisimplementierung einer Aufgabenwarteschlange in C++

Betrachten wir eine einfache, aber threadsichere Aufgabenwarteschlange unter Verwendung von std::queue und Mutexen. Dazu verwenden wir die Standardbibliothek ab 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(); // Aufgabenausführung        }    }

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

Anwendungsbeispiel

Nun zeigen wir, wie man unsere Klasse TaskQueue zur parallelen Ausführung einfacher Berechnungen verwendet:

#include <iostream>

int main() {    TaskQueue tq(4); // Pool aus 4 Threads

    for (int i = 0; i < 10; ++i) {        tq.enqueue([i] {            std::cout << "Aufgabe " << i << " ausgeführt von Thread "                      << std::this_thread::get_id() << std::endl;        });    }

    // Der Destruktor von tq wartet auf die Fertigstellung aller Aufgaben    return 0;}

Erweiterte Möglichkeiten: Prioritätswarteschlange

Manchmal müssen Aufgaben in der Reihenfolge ihrer Wichtigkeit verarbeitet werden. Dazu kann man std::priority_queue anstelle einer normalen Warteschlange verwenden. Fügen wir Unterstützung für Prioritäten hinzu:

#include <queue>#include <functional>

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

    struct PrioritizedTask {        int priority;        Task t