Task Parallel Library (TPL) und Parallelprogrammierung in C#

In .NET vereinfacht die Task Parallel Library (TPL) die parallele Ausführung, indem sie die Vorteile von Mehrkernprozessoren nutzt. Mit TPL können Sie Tools wie Task, Parallel, Concurrent-Sammlungen und PLINQ verwenden, um CPU-intensive Aufgaben zu beschleunigen. Dieser Artikel behandelt die grundlegenden Konzepte, geeignete Anwendungsfälle, Abbruch- und Ausnahmebehandlung sowie praktische Beispiele.

Was ist TPL? Einstieg mit Task

Eine Task repräsentiert eine Arbeitseinheit. Task.Run übergibt CPU-intensive Aufgaben an den Thread-Pool. Mehrere Aufgaben können gleichzeitig gestartet und mit Task.WhenAll gemeinsam abgewartet werden.

using System;
using System.Threading.Tasks;

class Program
{
    static async Task Main()
    {
        Task t1 = Task.Run(() => Arbeit("A", 1000));
        Task t2 = Task.Run(() => Arbeit("B", 800));
        Task t3 = Task.Run(() => Arbeit("C", 1200));

        await Task.WhenAll(t1, t2, t3);
        Console.WriteLine("Alle Aufgaben abgeschlossen.");
    }

    static void Arbeit(string name, int ms)
    {
        Console.WriteLine($"{name} gestartet");
        Task.Delay(ms).Wait(); // Beispielhafte Wartezeit (nicht CPU-intensiv)
        Console.WriteLine($"{name} beendet");
    }
}

Hinweis: Für echte I/O-Wartevorgänge sollte async/await bevorzugt werden; Task.Run ist für CPU-intensive Arbeiten gedacht.

Parallel.For / Parallel.ForEach

Die Parallel-Klasse teilt Schleifen automatisch auf und verteilt sie auf Threads. Beim Zugriff auf gemeinsame Variablen sollten thread-sichere Techniken wie Interlocked verwendet werden.

using System;
using System.Threading;
using System.Threading.Tasks;

class Program
{
    static void Main()
    {
        int summe = 0;

        Parallel.For(0, 1_000_000, i =>
        {
            // Verwendung von Interlocked, um Race Conditions zu vermeiden
            Interlocked.Add(ref summe, 1);
        });

        Console.WriteLine($"Summe: {summe}");
    }
}

using System;
using System.Collections.Generic;
using System.Threading.Tasks;

class Program
{
    static void Main()
    {
        var zahlen = new List<int> { 1, 2, 3, 4, 5 };
        Parallel.ForEach(zahlen, n =>
        {
            Console.WriteLine($"{n} verarbeitet (Thread: {Environment.CurrentManagedThreadId})");
        });
    }
}

ParallelOptions: Parallelitätsgrad und Abbruch

Mit MaxDegreeOfParallelism lässt sich der Grad der Parallelität steuern, und mit CancellationToken kann ein Abbruch unterstützt werden.

using System;
using System.Threading;
using System.Threading.Tasks;

class Program
{
    static void Main()
    {
        var cts = new CancellationTokenSource();
        var po = new ParallelOptions
        {
            MaxDegreeOfParallelism = Environment.ProcessorCount - 1, // nach Anzahl der Kerne
            CancellationToken = cts.Token
        };

        // Abbruch nach 2 Sekunden
        Task.Run(async () => { await Task.Delay(2000); cts.Cancel(); });

        try
        {
            Parallel.For(0, 1000, po, i =>
            {
                po.CancellationToken.ThrowIfCancellationRequested();
                CpuArbeit(i); // CPU-intensive Aufgabe
            });
        }
        catch (OperationCanceledException)
        {
            Console.WriteLine("Operationen wurden abgebrochen.");
        }
    }

    static void CpuArbeit(int i)
    {
        // Beispielhafte Berechnung
        double x = 0;
        for (int k = 0; k < 50_000; k++) x += Math.Sqrt(k + i);
    }
}

Aggregation mit lokalen Sammlern (Local Init/Finally)

Anstatt eine gemeinsame Variable zu sperren, kann jeder Thread einen lokalen Sammler verwenden und die Ergebnisse am Ende zusammenführen.

using System;
using System.Threading.Tasks;

class Program
{
    static void Main()
    {
        long globaleSumme = 0;

        Parallel.For<long>(0, 10_000_000,
            () => 0, // Lokale Summe pro Thread
            (i, loop, lokaleSumme) =>
            {
                lokaleSumme += i % 10;
                return lokaleSumme;
            },
            lokaleSumme => { System.Threading.Interlocked.Add(ref globaleSumme, lokaleSumme); });

        Console.WriteLine($"Summe: {globaleSumme}");
    }
}

Concurrent-Kollektionen

In Producer/Consumer-Szenarien bieten ConcurrentBag<T>, ConcurrentQueue<T> und ConcurrentDictionary<TKey,TValue> Sperr-freien oder Sperr-armen Zugriff für Thread-Sicherheit.

using System;
using System.Collections.Concurrent;
using System.Threading.Tasks;

class Program
{
    static void Main()
    {
        var bag = new ConcurrentBag<int>();

        Parallel.For(0, 1000, i => bag.Add(i));

        int zaehler = 0;
        while (bag.TryTake(out _)) zaehler++;

        Console.WriteLine($"Entnommene Elemente: {zaehler}");
    }
}

PLINQ (Parallel LINQ)

Verwenden Sie AsParallel(), um Abfragen über große Sammlungen parallel auszuführen. Wenn die Reihenfolge wichtig ist, fügen Sie AsOrdered() hinzu; andernfalls ist die unsortierte Ausführung schneller.

using System;
using System.Linq;

class Program
{
    static void Main()
    {
        var array = Enumerable.Range(1, 1_000_000).ToArray();

        var ergebnis = array
            .AsParallel()
            .Where(x => x % 3 == 0)
            .Select(x => x * x)
            .Take(10)
            .ToArray();

        Console.WriteLine(string.Join(", ", ergebnis));
    }
}

Fehlerbehandlung

In Parallel-Aufrufen können mehrere Ausnahmen auftreten; diese werden mit AggregateException behandelt. In Task-basierten Workflows treten Ausnahmen während des await auf.

using System;
using System.Threading.Tasks;

class Program
{
    static void Main()
    {
        try
        {
            Parallel.Invoke(
                () => throw new InvalidOperationException("A"),
                () => throw new ApplicationException("B")
            );
        }
        catch (AggregateException ex)
        {
            foreach (var e in ex.InnerExceptions)
                Console.WriteLine($"Fehler: {e.GetType().Name} - {e.Message}");
        }
    }
}

async/await vs Parallel

• async/await: Für I/O-intensive Operationen (Datei, Netzwerk, DB), um die UI reaktionsfähig zu halten.
• Parallel/TPL: Für CPU-intensive Aufgaben, um Mehrkernprozessoren optimal zu nutzen.
• Vermeiden Sie es, I/O-Aufgaben unnötig mit Task.Run in den Thread-Pool zu verschieben; der wirkliche Nutzen liegt bei CPU-intensiver Arbeit.

Leistungstipps und bewährte Praktiken

• Minimieren Sie gemeinsame Zustände; kombinieren Sie Ergebnisse mit lokalen Sammlern, wenn möglich.
• Vermeiden Sie unnötige Sperren; verwenden Sie Interlocked oder Concurrent-Sammlungen bei Bedarf.
• Verwenden Sie MaxDegreeOfParallelism, um übermäßigen Kontextwechsel zu vermeiden.
• Vermeiden Sie blockierende Aufrufe (.Wait()/.Result) in UI-Anwendungen.

Beispiel: Parallele Bildverarbeitung

Das folgende Beispiel simuliert die parallele Verarbeitung von Bilddateien in einem Ordner. In einem echten Szenario könnten hier CPU-intensive Operationen wie Filterung oder Thumbnail-Erstellung durchgeführt werden.

using System;
using System.IO;
using System.Threading;
using System.Threading.Tasks;

class Program
{
    static void Main()
    {
        string ordner = "C:\\\\Bilder";
        var dateien = Directory.Exists(ordner)
            ? Directory.GetFiles(ordner, "*.jpg")
            : Array.Empty<string>();

        var po = new ParallelOptions { MaxDegreeOfParallelism = Environment.ProcessorCount };
        Parallel.ForEach(dateien, po, datei =>
        {
            // Simulation einer CPU-intensiven Filterung
            var name = Path.GetFileName(datei);
            Console.WriteLine($"Verarbeite: {name}");
            Thread.SpinWait(3000000); // Simulation
            Console.WriteLine($"Fertig: {name}");
        });

        Console.WriteLine("Alle Dateien wurden verarbeitet.");
    }
}

TL;DR

Ähnliche Artikel