【C#】非同期・並列処理の実践パターン集｜Fan-out/Rate Limit/Pipeline/Worker Pool/エラーハンドリングまで

// 100 ユーザーの情報を並行取得
public async Task<List<User>> FetchUsersAsync(IEnumerable<int> userIds)
{
    var tasks = userIds.Select(id => FetchUserAsync(id));
    User[] users = await Task.WhenAll(tasks);
    return users.ToList();
}

// NG: foreach + await で逐次実行になってしまう
public async Task<List<User>> FetchUsersBadAsync(IEnumerable<int> userIds)
{
    var result = new List<User>();
    foreach (var id in userIds)
        result.Add(await FetchUserAsync(id));  // 100倍遅い
    return result;
}

// Task.WhenAny: 最初に終わった1つだけ取る（競争）
public async Task<string> FetchFirstOfAsync(params string[] urls)
{
    var tasks = urls.Select(FetchAsync).ToList();
    Task<string> first = await Task.WhenAny(tasks);
    return await first;   // 最速で応答したエンドポイントの結果
}

public sealed class RateLimitedClient
{
    private readonly SemaphoreSlim _gate = new(initialCount: 10);  // 同時 10 まで
    private readonly HttpClient _http = new();

    public async Task<string> FetchAsync(string url, CancellationToken ct)
    {
        await _gate.WaitAsync(ct);
        try
        {
            return await _http.GetStringAsync(url, ct);
        }
        finally
        {
            _gate.Release();  // 必ず解放（例外時も）
        }
    }

    public async Task<List<string>> FetchAllAsync(
        IEnumerable<string> urls, CancellationToken ct)
    {
        var tasks = urls.Select(u => FetchAsync(u, ct));
        return (await Task.WhenAll(tasks)).ToList();
    }
}

// 代替: Parallel.ForEachAsync の MaxDegreeOfParallelism（.NET 6+）
public async Task FetchAllAsync2(IEnumerable<string> urls, CancellationToken ct)
{
    await Parallel.ForEachAsync(urls,
        new ParallelOptions { MaxDegreeOfParallelism = 10, CancellationToken = ct },
        async (url, c) => await _http.GetStringAsync(url, c));
}

using System.Threading.RateLimiting;

// 1秒あたり100リクエストまで許可（.NET 7+ の System.Threading.RateLimiting）
var limiter = new TokenBucketRateLimiter(new TokenBucketRateLimiterOptions
{
    TokenLimit        = 100,
    TokensPerPeriod   = 100,
    ReplenishmentPeriod = TimeSpan.FromSeconds(1),
    AutoReplenishment = true,
    QueueLimit        = 1000,
});

public async Task<string> FetchWithLimitAsync(string url, CancellationToken ct)
{
    using RateLimitLease lease = await limiter.AcquireAsync(permitCount: 1, ct);
    if (!lease.IsAcquired) throw new InvalidOperationException("レート制限超過");

    return await _http.GetStringAsync(url, ct);
}

using System.Threading.Tasks.Dataflow;

// ダウンロード（I/O）→ 変換（CPU）→ 保存（I/O）
// 各ステージで適切な並列度を別々に設定できる
public sealed class DocumentPipeline
{
    private readonly TransformBlock<string, string> _download;
    private readonly TransformBlock<string, Document> _parse;
    private readonly ActionBlock<Document> _save;

    public DocumentPipeline(HttpClient http, IDatabase db)
    {
        _download = new(async url =>
        {
            return await http.GetStringAsync(url);
        }, new ExecutionDataflowBlockOptions
        {
            MaxDegreeOfParallelism = 20,   // I/O は多めに並列
            BoundedCapacity         = 100, // バックプレッシャー
        });

        _parse = new(html => ParseDocument(html),
            new ExecutionDataflowBlockOptions
            {
                MaxDegreeOfParallelism = Environment.ProcessorCount, // CPU コア数
                BoundedCapacity         = 50,
            });

        _save = new(async doc => await db.SaveAsync(doc),
            new ExecutionDataflowBlockOptions
            {
                MaxDegreeOfParallelism = 4,   // DB の接続プール上限
                BoundedCapacity         = 30,
            });

        var link = new DataflowLinkOptions { PropagateCompletion = true };
        _download.LinkTo(_parse, link);
        _parse.LinkTo(_save, link);
    }

    public async Task ProcessAsync(IEnumerable<string> urls)
    {
        foreach (var u in urls) await _download.SendAsync(u);
        _download.Complete();
        await _save.Completion;
    }
}

using System.Threading.Channels;

public sealed class WorkerPool<T>
{
    private readonly Channel<T> _queue;
    private readonly Func<T, CancellationToken, Task> _handler;
    private readonly int _workerCount;

    public WorkerPool(int workerCount, int queueCapacity,
        Func<T, CancellationToken, Task> handler)
    {
        _workerCount = workerCount;
        _handler     = handler;
        _queue       = Channel.CreateBounded<T>(new BoundedChannelOptions(queueCapacity)
        {
            FullMode = BoundedChannelFullMode.Wait,
            SingleReader = false,
            SingleWriter = false,
        });
    }

    public ValueTask EnqueueAsync(T item, CancellationToken ct)
        => _queue.Writer.WriteAsync(item, ct);

    public void Complete() => _queue.Writer.Complete();

    public async Task RunAsync(CancellationToken ct)
    {
        var workers = Enumerable.Range(0, _workerCount)
            .Select(_ => Task.Run(async () =>
            {
                await foreach (var item in _queue.Reader.ReadAllAsync(ct))
                    await _handler(item, ct);
            }, ct))
            .ToArray();

        await Task.WhenAll(workers);
    }
}

// 使用例
var pool = new WorkerPool<WorkItem>(
    workerCount: 8,
    queueCapacity: 100,
    handler: async (item, ct) => await ProcessAsync(item, ct));

var runTask = pool.RunAsync(cts.Token);
foreach (var work in LoadWorkItems())
    await pool.EnqueueAsync(work, cts.Token);
pool.Complete();
await runTask;

// プロデューサが速すぎてメモリが膨張するのを防ぐ
// BoundedChannel で上限を設けると、満杯時は Writer 側が自動待機
var channel = Channel.CreateBounded<LogEntry>(new BoundedChannelOptions(10_000)
{
    FullMode = BoundedChannelFullMode.Wait,  // 上限到達時はプロデューサが待機
    // 代替: DropOldest / DropNewest / DropWrite（損失を許容する場合）
});

// プロデューサ: ログを高速生成
Task producer = Task.Run(async () =>
{
    while (IsRunning)
    {
        LogEntry entry = await ReadFromSocketAsync();
        await channel.Writer.WriteAsync(entry);  // 満杯なら自動で待つ
    }
    channel.Writer.Complete();
});

// コンシューマ: DB への書き込みは遅い
Task consumer = Task.Run(async () =>
{
    await foreach (var entry in channel.Reader.ReadAllAsync())
        await db.InsertAsync(entry);
});

await Task.WhenAll(producer, consumer);

// NG: 毎回ロック → 直列化して並列のメリットが消える
long sum = 0;
object gate = new();
Parallel.For(0, items.Length, i =>
{
    long local = Compute(items[i]);
    lock (gate) { sum += local; }  // 競合
});

// OK: スレッドローカル状態で Map→Reduce
long sumFast = 0;
Parallel.For(0, items.Length,
    localInit: () => 0L,
    body: (i, state, local) => local + Compute(items[i]),
    localFinally: local => Interlocked.Add(ref sumFast, local));

// PLINQ 版（シンプルだが制御しにくい）
long sumPlinq = items.AsParallel()
                     .Select(Compute)
                     .Sum();

// 複数集計が必要な場合（平均・分散・最大最小）
var stats = items.AsParallel()
    .Aggregate(
        seedFactory: () => new Stats(),
        updateAccumulatorFunc: (acc, x) => acc.Add(x),
        combineAccumulatorsFunc: (a, b) => a.Merge(b),
        resultSelector: acc => acc.ToResult());

public sealed record FetchResult<T>(T? Value, Exception? Error, string Source)
{
    public bool IsSuccess => Error is null;
}

public async Task<List<FetchResult<string>>> FetchAllSafeAsync(
    IEnumerable<string> urls, CancellationToken ct)
{
    async Task<FetchResult<string>> TryFetch(string url)
    {
        try
        {
            var body = await _http.GetStringAsync(url, ct);
            return new FetchResult<string>(body, null, url);
        }
        catch (Exception ex)
        {
            return new FetchResult<string>(null, ex, url);
        }
    }

    var results = await Task.WhenAll(urls.Select(TryFetch));
    return results.ToList();
}

// 使用側で成功/失敗を振り分け
var results = await FetchAllSafeAsync(urls, ct);
var succeeded = results.Where(r => r.IsSuccess).Select(r => r.Value!).ToList();
var failed    = results.Where(r => !r.IsSuccess).ToList();

foreach (var f in failed)
    logger.LogWarning(f.Error, "Failed to fetch {Url}", f.Source);

// ① 全体に 30 秒のタイムアウト
public async Task RunAllWithGlobalTimeoutAsync(IEnumerable<string> urls)
{
    using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
    try
    {
        var tasks = urls.Select(u => FetchAsync(u, cts.Token));
        await Task.WhenAll(tasks);
    }
    catch (OperationCanceledException) when (cts.IsCancellationRequested)
    {
        Console.WriteLine("全体タイムアウト");
    }
}

// ② 個別に 5 秒のタイムアウト（1つ遅くても全体は止めない）
public async Task<string?> FetchWithTimeoutAsync(string url, TimeSpan timeout)
{
    using var cts = new CancellationTokenSource(timeout);
    try
    {
        return await _http.GetStringAsync(url, cts.Token);
    }
    catch (OperationCanceledException)
    {
        return null;
    }
}

// ③ .NET 6+: Task.WaitAsync でタイムアウトを手軽に
public async Task<string> FetchAsync(string url)
{
    return await _http.GetStringAsync(url)
        .WaitAsync(TimeSpan.FromSeconds(10));  // 10秒超過で TimeoutException
}

using Polly;

// 指数バックオフ: 1秒 → 2秒 → 4秒 で最大3回リトライ
var policy = Policy
    .Handle<HttpRequestException>()
    .WaitAndRetryAsync(
        retryCount: 3,
        sleepDurationProvider: attempt => TimeSpan.FromSeconds(Math.Pow(2, attempt - 1)),
        onRetry: (ex, delay, attempt, _) =>
            Console.WriteLine($"Retry {attempt} after {delay}: {ex.Message}"));

string result = await policy.ExecuteAsync(
    ct => _http.GetStringAsync("https://api.example.com", ct),
    cancellationToken: ct);

// .NET 8+: Microsoft.Extensions.Resilience で標準のレジリエンス
services.AddHttpClient<MyApiClient>()
    .AddStandardResilienceHandler();  // リトライ・サーキットブレーカー・タイムアウトが全部入り

// 自前実装（Polly なしでシンプルに）
public async Task<T> RetryAsync<T>(
    Func<CancellationToken, Task<T>> action,
    int maxAttempts,
    CancellationToken ct)
{
    Exception? lastError = null;
    for (int i = 0; i < maxAttempts; i++)
    {
        try { return await action(ct); }
        catch (Exception ex) when (ex is not OperationCanceledException)
        {
            lastError = ex;
            var delay = TimeSpan.FromMilliseconds(Math.Pow(2, i) * 100);
            await Task.Delay(delay, ct);
        }
    }
    throw new InvalidOperationException($"{maxAttempts}回試行して失敗", lastError);
}

public async Task<T> RobustCallAsync<T>(
    Func<CancellationToken, Task<T>> action,
    CancellationToken userToken,
    TimeSpan timeout)
{
    // アプリ停止用（AppLifetime 等から）
    CancellationToken appShutdownToken = _appLifetime.ApplicationStopping;

    // 3つを束ねた1つのトークンを作る
    using var linked = CancellationTokenSource.CreateLinkedTokenSource(
        userToken, appShutdownToken);
    linked.CancelAfter(timeout);

    try
    {
        return await action(linked.Token);
    }
    catch (OperationCanceledException)
    {
        // どの要因でキャンセルされたか判別
        if (userToken.IsCancellationRequested)      throw new UserCancelledException();
        if (appShutdownToken.IsCancellationRequested) throw new AppShutdownException();
        throw new TimeoutException($"{timeout} 経過");
    }
}

using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Running;

[MemoryDiagnoser]
public class FanOutBench
{
    private readonly int[] _ids = Enumerable.Range(0, 100).ToArray();

    [Benchmark(Baseline = true)]
    public async Task Sequential()
    {
        foreach (var id in _ids) await FetchAsync(id);
    }

    [Benchmark]
    public async Task FanOut_Unlimited()
    {
        await Task.WhenAll(_ids.Select(FetchAsync));
    }

    [Benchmark]
    public async Task FanOut_Limited_10()
    {
        using var gate = new SemaphoreSlim(10);
        async Task Gated(int id)
        {
            await gate.WaitAsync();
            try { await FetchAsync(id); } finally { gate.Release(); }
        }
        await Task.WhenAll(_ids.Select(Gated));
    }
}

// 実行: BenchmarkRunner.Run<FanOutBench>();