csharp-concurrency-patterns

.NET の並行処理：適切なツールの選び方

このスキルを使用する場面

次のような場合に、このスキルを使用してください。

• .NET で並行処理を行う方法を決めるとき
• async/await、Channels、Akka.NET、その他の抽象化を使用すべきか評価するとき
• ロック、セマフォ、その他の同期プリミティブを使おうとしているとき
• バックプレッシャー、バッチ処理、またはデバウンスを伴うデータストリームを処理する必要があるとき
• 複数の並行エンティティ間で状態を管理するとき

参考ファイル

• advanced-concurrency.md：Akka.NET Streams、Reactive Extensions、Akka.NET Actors（エンティティごとのアクター、ステートマシン、クラスターシャーディング）、および非同期ローカル関数のパターン

基本方針

まずはシンプルに始め、必要な場合にのみ高度な方法を選びましょう。

ほとんどの並行処理の問題は async/await で解決できます。async/await では適切に対処できない具体的な要件がある場合にのみ、より高度なツールを使用してください。

共有された可変状態はできるだけ避けてください。 並行処理に対処する最善の方法は、設計によってそれを不要にすることです。不変データ、メッセージパッシング、アクターのような分離された状態を使うことで、多くの種類のバグを根本から排除できます。

ロックは原則ではなく、例外として使うべきです。 共有された可変状態を避けられない場合：

1. 第一候補： 設計を見直して回避します（不変性、メッセージパッシング、アクターによる分離）
2. 第二候補： System.Collections.Concurrent（ConcurrentDictionary など）を使用します
3. 第三候補： Channel<T> を使用し、メッセージパッシングでアクセスを直列化します
4. 最終手段： 単純で短時間のクリティカルセクションには lock を使用します

---

判断フロー

What are you trying to do?
│
├─► Wait for I/O (HTTP, database, file)?
│   └─► Use async/await
│
├─► Process a collection in parallel (CPU-bound)?
│   └─► Use Parallel.ForEachAsync
│
├─► Producer/consumer pattern (work queue)?
│   └─► Use System.Threading.Channels
│
├─► UI event handling (debounce, throttle, combine)?
│   └─► Use Reactive Extensions (Rx)
│
├─► Server-side stream processing (backpressure, batching)?
│   └─► Use Akka.NET Streams
│
├─► State machines with complex transitions?
│   └─► Use Akka.NET Actors (Become pattern)
│
├─► Manage state for many independent entities?
│   └─► Use Akka.NET Actors (entity-per-actor)
│
├─► Coordinate multiple async operations?
│   └─► Use Task.WhenAll / Task.WhenAny
│
└─► None of the above fits?
    └─► Ask yourself: "Do I really need shared mutable state?"
        ├─► Yes → Consider redesigning to avoid it
        └─► Truly unavoidable → Use Channels or Actors to serialize access

---

レベル 1：async/await（デフォルトの選択肢）

用途： I/O バウンドの操作、ノンブロッキングの待機、日常的な並行処理の大半。

// Simple async I/O
public async Task<Order> GetOrderAsync(string orderId, CancellationToken ct)
{
    var order = await _database.GetAsync(orderId, ct);
    var customer = await _customerService.GetAsync(order.CustomerId, ct);
    return order with { Customer = customer };
}

// Parallel async operations (when independent)
public async Task<Dashboard> LoadDashboardAsync(string userId, CancellationToken ct)
{
    var ordersTask = _orderService.GetRecentOrdersAsync(userId, ct);
    var notificationsTask = _notificationService.GetUnreadAsync(userId, ct);
    var statsTask = _statsService.GetUserStatsAsync(userId, ct);

    await Task.WhenAll(ordersTask, notificationsTask, statsTask);

    return new Dashboard(
        Orders: await ordersTask,
        Notifications: await notificationsTask,
        Stats: await statsTask);
}

重要な原則： 常に CancellationToken を受け取ってください。ライブラリコードでは ConfigureAwait(false) を使用してください。非同期コードをブロックしないでください。

---

レベル 2：Parallel.ForEachAsync（CPU バウンドの並列処理）

用途： CPU バウンドの処理や並行処理数を制御する必要がある場合に、コレクションを並列処理します。

public async Task ProcessOrdersAsync(
    IEnumerable<Order> orders,
    CancellationToken ct)
{
    await Parallel.ForEachAsync(
        orders,
        new ParallelOptions
        {
            MaxDegreeOfParallelism = Environment.ProcessorCount,
            CancellationToken = ct
        },
        async (order, token) =>
        {
            await ProcessOrderAsync(order, token);
        });
}

使用しない場面： I/O 操作のみの場合、順序が重要な場合、バックプレッシャーが必要な場合。

---

レベル 3：System.Threading.Channels（プロデューサー／コンシューマー）

用途： 作業キュー、プロデューサー／コンシューマーパターン、プロデューサーとコンシューマーの分離。

public class OrderProcessor
{
    private readonly Channel<Order> _channel;

    public OrderProcessor()
    {
        _channel = Channel.CreateBounded<Order>(new BoundedChannelOptions(100)
        {
            FullMode = BoundedChannelFullMode.Wait
        });
    }

    // Producer
    public async Task EnqueueOrderAsync(Order order, CancellationToken ct)
    {
        await _channel.Writer.WriteAsync(order, ct);
    }

    // Consumer (run as background task)
    public async Task ProcessOrdersAsync(CancellationToken ct)
    {
        await foreach (var order in _channel.Reader.ReadAllAsync(ct))
        {
            await ProcessOrderAsync(order, ct);
        }
    }

    public void Complete() => _channel.Writer.Complete();
}

Channels が適している用途： 処理速度の差を吸収すること、バックプレッシャーを伴うバッファリング、複数のワーカーへのファンアウト、バックグラウンドキュー。

Channels が適していない用途： 複雑なストリーム操作（バッチ処理、ウィンドウ処理）、エンティティごとの状態を持つ処理、高度な監視機能。

---

レベル 4 以上：Akka.NET Streams、Reactive Extensions、Actors

ストリーム処理、UI イベントの合成、または状態を持つエンティティの管理が必要な高度なシナリオについては、advanced-concurrency.md を参照してください。

Akka.NET Streams は、サーバー側のバッチ処理、スロットリング、バックプレッシャーに優れています。Reactive Extensions は、UI イベントの合成に最適です。Akka.NET Actors は、エンティティごとのアクターパターン、Become() を使ったステートマシン、クラスターシャーディングによる分散システムに対応します。

---

アンチパターン：避けるべきこと

ビジネスロジックでのロック

// BAD: Using locks to protect shared state
private readonly object _lock = new();
private Dictionary<string, Order> _orders = new();

public void UpdateOrder(string id, Action<Order> update)
{
    lock (_lock) { if (_orders.TryGetValue(id, out var order)) update(order); }
}

// GOOD: Use an actor or Channel to serialize access

手動でのスレッド管理

// BAD: Creating threads manually
var thread = new Thread(() => ProcessOrders());
thread.Start();

// GOOD: Use Task.Run or better abstractions
_ = Task.Run(() => ProcessOrdersAsync(cancellationToken));

非同期コード内でのブロッキング

// BAD: Blocking on async - deadlock risk!
var result = GetDataAsync().Result;

// GOOD: Async all the way
var result = await GetDataAsync();

保護されていない共有可変状態

// BAD: Multiple tasks mutating shared state
var results = new List<Result>();
await Parallel.ForEachAsync(items, async (item, ct) =>
{
    var result = await ProcessAsync(item, ct);
    results.Add(result); // Race condition!
});

// GOOD: Use ConcurrentBag
var results = new ConcurrentBag<Result>();

---

クイックリファレンス：どのツールを使うか？

必要なこと	ツール	例
I/O の待機	async/await	HTTP 呼び出し、データベースクエリ
CPU 処理の並列実行	Parallel.ForEachAsync	画像処理、計算
作業キュー	Channel<T>	バックグラウンドジョブの処理
デバウンス／スロットリングを伴う UI イベント	Reactive Extensions	入力中の検索、自動保存
サーバー側のバッチ処理／スロットリング	Akka.NET Streams	イベントの集約、レート制限
ステートマシン	Akka.NET Actors	決済フロー、注文のライフサイクル
エンティティ状態の管理	Akka.NET Actors	注文管理、ユーザーセッション
複数の非同期処理を並行して実行	Task.WhenAll	ダッシュボードデータの読み込み
複数の非同期処理を競争させる	Task.WhenAny	フォールバック付きタイムアウト
定期的な処理	PeriodicTimer	ヘルスチェック、ポーリング

---

段階的なエスカレーション

async/await (start here)
    │
    ├─► Need parallelism? → Parallel.ForEachAsync
    │
    ├─► Need producer/consumer? → Channel<T>
    │
    ├─► Need UI event composition? → Reactive Extensions
    │
    ├─► Need server-side stream processing? → Akka.NET Streams
    │
    └─► Need state machines or entity management? → Akka.NET Actors

具体的な必要性がある場合にのみエスカレーションしてください。「念のため」にアクターやストリームを使おうとしないでください。