【C#】非同期処理と並列処理の違い完全ガイド｜Task・Parallel・PLINQ・Dataflow・Channelsの使い分け

// ① I/O-bound: HTTP 呼び出し → CPU を一切使わずネットワーク待ち
// → async/await が正解。Task.Run は CPU スレッドを占有するだけで無意味
public async Task<string> FetchAsync(string url)
{
    using var http = new HttpClient();
    return await http.GetStringAsync(url);   // OK: I/O 待ち中はスレッドが解放される
}

// ② CPU-bound: 数値計算 → CPU が忙しい
// → Parallel や Task.Run で別スレッドに逃がす
public double ComputePi(int iterations)
{
    double sum = 0;
    for (int i = 0; i < iterations; i++)
        sum += 4.0 * (i % 2 == 0 ? 1 : -1) / (2 * i + 1);
    return sum;
}

// CPU-bound 処理を UI スレッドから追い出すときだけ Task.Run が有用
public async Task OnButtonClickAsync()
{
    double result = await Task.Run(() => ComputePi(100_000_000));
    Label.Text = result.ToString();   // UI スレッドで更新
}

// ③ 混在: 大量 URL を並行取得し、各結果をパース
// → Parallel.ForEachAsync（.NET 6+）が最適
var urls = GetUrls();
await Parallel.ForEachAsync(urls,
    new ParallelOptions { MaxDegreeOfParallelism = 10 },
    async (url, ct) =>
    {
        string html = await FetchAsync(url);       // I/O 部分は非同期
        var parsed  = ParseDocument(html);         // CPU 部分
        await SaveAsync(parsed, ct);
    });

// ① Parallel.For: インデックス 0〜N-1 の処理を並列化
var sums = new double[1000];
Parallel.For(0, sums.Length, i =>
{
    sums[i] = ExpensiveCompute(i);   // CPU 負荷の高い計算
});

// ② Parallel.ForEach: 既存コレクションの並列走査
var images = GetImageList();
Parallel.ForEach(images, img =>
{
    ProcessImage(img);
});

// ③ Parallel.Invoke: 異なる処理を同時実行
Parallel.Invoke(
    () => GenerateThumbnails(),
    () => UpdateDatabase(),
    () => CalculateStatistics());
// 3つの処理が並列に走り、すべて終わるまでブロック

// ④ Parallel.ForEachAsync（.NET 6+）: 非同期処理の並列化
//    MaxDegreeOfParallelism で同時実行数を制御できる
var urls = GetUrls();
await Parallel.ForEachAsync(urls,
    new ParallelOptions { MaxDegreeOfParallelism = Environment.ProcessorCount },
    async (url, ct) =>
    {
        await DownloadAsync(url, ct);
    });

using System.Linq;

var numbers = Enumerable.Range(1, 10_000_000);

// 通常 LINQ（シングルスレッド）
int count = numbers.Where(IsPrime).Count();

// PLINQ: AsParallel() で並列実行に切り替え
int countPar = numbers.AsParallel()
                      .Where(IsPrime)
                      .Count();

// 並列度を制御
int count2 = numbers.AsParallel()
                    .WithDegreeOfParallelism(4)       // 4 スレッドまで
                    .WithCancellation(cts.Token)      // キャンセル対応
                    .Where(IsPrime)
                    .Count();

// 順序を保持したい場合（ソート済みを保つ）
var sorted = numbers.AsParallel()
                    .AsOrdered()        // 元の順序を保証
                    .Where(IsPrime)
                    .Take(100)
                    .ToArray();

// ForAll: 結果の列挙を並列化（副作用あり処理用）
numbers.AsParallel().ForAll(n => Process(n));

static bool IsPrime(int n)
{
    if (n < 2) return false;
    for (int i = 2; i * i <= n; i++)
        if (n % i == 0) return false;
    return true;
}

using System.Threading.Tasks.Dataflow;

// ダウンロード → パース → 保存 のパイプライン
var downloadBlock = new TransformBlock<string, string>(
    async url =>
    {
        using var http = new HttpClient();
        return await http.GetStringAsync(url);
    },
    new ExecutionDataflowBlockOptions
    {
        MaxDegreeOfParallelism = 5,       // 同時ダウンロード数
        BoundedCapacity         = 20,     // バッファ上限
    });

var parseBlock = new TransformBlock<string, Document>(
    html => ParseDocument(html),            // CPU-bound
    new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4 });

var saveBlock = new ActionBlock<Document>(
    async doc => await SaveAsync(doc),
    new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 2 });

// ブロックをリンク（完了もパイプライン化）
var linkOpts = new DataflowLinkOptions { PropagateCompletion = true };
downloadBlock.LinkTo(parseBlock, linkOpts);
parseBlock.LinkTo(saveBlock, linkOpts);

// URLを投入
foreach (var url in GetUrls())
    await downloadBlock.SendAsync(url);
downloadBlock.Complete();

// 最後のブロックの完了を待つ
await saveBlock.Completion;

using System.Threading.Channels;

var channel = Channel.CreateBounded<Work>(new BoundedChannelOptions(100)
{
    FullMode = BoundedChannelFullMode.Wait,
});

// プロデューサ（通常1本〜数本）
var producer = Task.Run(async () =>
{
    foreach (var w in GetWorkItems())
        await channel.Writer.WriteAsync(w);
    channel.Writer.Complete();
});

// コンシューマ（複数本でスケール）
var consumers = Enumerable.Range(0, 4).Select(_ => Task.Run(async () =>
{
    await foreach (var w in channel.Reader.ReadAllAsync())
        await ProcessAsync(w);
})).ToArray();

await Task.WhenAll(new[] { producer }.Concat(consumers));

// デフォルト: Environment.ProcessorCount（論理コア数）
// CPU-bound 処理なら通常これが最適

var opts = new ParallelOptions
{
    MaxDegreeOfParallelism = Environment.ProcessorCount,  // CPU コア数
    CancellationToken      = cts.Token,
};

Parallel.ForEach(items, opts, item => ProcessCpu(item));

// I/O-bound 処理（Parallel.ForEachAsync）では CPU コア数より多くてよい
// → ネットワーク待ち中もスレッドが解放されるため、並列数を増やせる
await Parallel.ForEachAsync(urls,
    new ParallelOptions { MaxDegreeOfParallelism = 50 },   // 同時 50 コネクション
    async (url, ct) => await FetchAsync(url, ct));

// MaxDegreeOfParallelism = 1 → シーケンシャル実行（デバッグ用）
// MaxDegreeOfParallelism = -1 → 制限なし（通常は避ける）

// NG: 共有変数への競合書き込み → 非決定的な結果
int total = 0;
Parallel.For(0, 10_000_000, i =>
{
    total += i;  // データ競合！
});
Console.WriteLine(total);  // 毎回違う値

// OK ①: Interlocked で原子的にインクリメント（軽量）
long totalSafe = 0;
Parallel.For(0, 10_000_000, i =>
{
    Interlocked.Add(ref totalSafe, i);
});

// OK ②: lock で排他制御（重いが柔軟）
int total2 = 0;
object gate = new();
Parallel.For(0, 10_000_000, i =>
{
    lock (gate) { total2 += i; }
});

// OK ③: Parallel.For のローカル状態パターン（最速）
// 各スレッドが自分のローカル変数で集計し、最後だけまとめる
long total3 = 0;
Parallel.For(0, 10_000_000,
    localInit: () => 0L,                             // スレッドごとの初期値
    body:      (i, state, local) => local + i,       // ローカル加算
    localFinally: local => Interlocked.Add(ref total3, local)); // 最後に合算

// OK ④: 並行コレクションを使う（ConcurrentBag / ConcurrentDictionary 等）
var bag = new ConcurrentBag<int>();
Parallel.For(0, 1000, i => bag.Add(i * i));

using var cts = new CancellationTokenSource();

// Ctrl+C で中止できる
Console.CancelKeyPress += (_, e) => { e.Cancel = true; cts.Cancel(); };

// ① CancellationToken を全パスに伝える
try
{
    await Parallel.ForEachAsync(items,
        new ParallelOptions
        {
            MaxDegreeOfParallelism = 4,
            CancellationToken      = cts.Token,
        },
        async (item, ct) => await ProcessAsync(item, ct));
}
catch (OperationCanceledException)
{
    Console.WriteLine("処理が中止されました");
}

// ② IProgress<T> で UI に進捗を通知
var progress = new Progress<int>(p => Console.WriteLine($"{p}% 完了"));
int completed = 0;
int total     = items.Count;

await Parallel.ForEachAsync(items,
    async (item, ct) =>
    {
        await ProcessAsync(item, ct);
        int done = Interlocked.Increment(ref completed);
        ((IProgress<int>)progress).Report(done * 100 / total);
    });

// NG: async void は例外を呼び出し側で catch できない・テストしづらい
public async void OnClick(object sender, EventArgs e)
{
    await Task.Delay(1000);
    throw new Exception("エラー！");  // このまま AppDomain 全体が落ちる
}

// OK: イベントハンドラ以外は必ず async Task
public async Task OnClickHandler() { /* ... */ }

// イベントハンドラの async void でも try-catch で例外を握る
public async void OnClick2(object sender, EventArgs e)
{
    try { await DoWorkAsync(); }
    catch (Exception ex) { Logger.Error(ex); }
}

// NG: async メソッドを同期的に .Result で待つ
// → UI スレッドや ASP.NET Classic でデッドロック

public async Task<string> FetchAsync() { /* ... */ }

// 以下はデッドロックの可能性あり
string data = FetchAsync().Result;   // NG
FetchAsync().Wait();                 // NG

// OK: Task.Run でスレッドプールに逃がす、または全パスを async 化
string data2 = Task.Run(() => FetchAsync()).GetAwaiter().GetResult(); // やむを得ない時のみ
string data3 = await FetchAsync();   // 最良: 呼び出し側も async にする

// NG: Parallel.ForEach は同期メソッドを前提にしている
// async ラムダを渡すと「fire-and-forget」になり Wait されない
Parallel.ForEach(urls, async url =>
{
    await DownloadAsync(url); // ← ループは await されずに即終了する！
});

// OK: 非同期処理の並列化には Parallel.ForEachAsync を使う（.NET 6+）
await Parallel.ForEachAsync(urls, async (url, ct) =>
{
    await DownloadAsync(url, ct);
});

// OK: .NET 5 以前なら Task.WhenAll
await Task.WhenAll(urls.Select(DownloadAsync));

// Task.WhenAll は最初の例外だけ再スローするが
// AggregateException には全エラーが入っている
var tasks = urls.Select(FetchAsync);
try
{
    await Task.WhenAll(tasks);
}
catch (Exception)
{
    // この catch には最初の例外しか来ない
    // 全部のエラーを見たい場合は Task のプロパティから
    var all = tasks.Where(t => t.IsFaulted).Select(t => t.Exception!);
    foreach (var ex in all)
        Console.WriteLine(ex.InnerException!.Message);
}