.NET並行処理: 適切なツールの選択

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

以下の場合にこのスキルを使用します:

• .NETで並行処理をどのように処理するかを決定する場合
• async/await、Channels、Akka.NET、または他の抽象化を使用すべきかを評価する場合
• ロック、セマフォ、または他の同期プリミティブを使用したい誘惑に駆られている場合
• バックプレッシャー、バッチング、またはデバウンスを使用してデータストリームを処理する必要がある場合
• 複数の並行エンティティ間でステートを管理する場合

参照ファイル

• advanced-concurrency.md: Akka.NET Streams、Reactive Extensions、Akka.NET Actors(エンティティ単位のアクター、ステートマシン、Cluster Sharding)、および非同期ローカル関数パターン

哲学

シンプルに始めて、必要な場合のみエスカレーションする。

ほとんどの並行処理の問題はasync/awaitで解決できます。async/awaitで明確に対応できない特定のニーズがある場合にのみ、より洗練されたツールを使用してください。

共有可変ステートを避けるようにしてください。 並行処理を扱う最良の方法は、設計段階で回避することです。イミュータブルなデータ、メッセージパッシング、および分離されたステート(アクターのような)は、バグの多くのカテゴリを排除します。

ロックは例外であり、ルールではありません。 共有可変ステートを避けられない場合:

1. 第一選択: 設計を変更して回避する(イミュータビリティ、メッセージパッシング、アクター分離)
2. 第二選択: System.Collections.Concurrentを使用(ConcurrentDictionary等)
3. 第三選択: Channel<T>を使用してメッセージパッシングでアクセスをシリアライズする
4. 最後の手段: lockを使用して、シンプルで短命なクリティカルセクションの場合のみ

---

決定フロー

何をしようとしていますか?
│
├─► I/O(HTTP、データベース、ファイル)を待機?
│   └─► async/awaitを使用
│
├─► コレクションを並列処理(CPU束縛)?
│   └─► Parallel.ForEachAsyncを使用
│
├─► プロデューサー・コンシューマーパターン(仕事キュー)?
│   └─► System.Threading.Channelsを使用
│
├─► UIイベント処理(デバウンス、スロットル、結合)?
│   └─► Reactive Extensions(Rx)を使用
│
├─► サーバー側ストリーム処理(バックプレッシャー、バッチング)?
│   └─► Akka.NET Streamsを使用
│
├─► 複雑な遷移を持つステートマシン?
│   └─► Akka.NET Actors(Becomeパターン)を使用
│
├─► 多数の独立したエンティティのステート管理?
│   └─► Akka.NET Actors(エンティティ単位のアクター)を使用
│
├─► 複数の非同期操作を調整?
│   └─► Task.WhenAll / Task.WhenAnyを使用
│
└─► 上記のいずれにも該当しない?
    └─► 自問してください: 「本当に共有可変ステートが必要ですか?」
        ├─► はい → 回避するように再設計を検討
        └─► 本当に避けられない → ChannelsまたはActorsを使用してアクセスをシリアライズ

---

レベル1: async/await(デフォルトの選択肢)

使用対象: I/O束縛の操作、非ブロッキング待機、日常的なほとんどの並行処理。

// シンプルな非同期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 };
}

// 並列非同期操作(独立している場合)
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
        });
    }

    // プロデューサー
    public async Task EnqueueOrderAsync(Order order, CancellationToken ct)
    {
        await _channel.Writer.WriteAsync(order, ct);
    }

    // コンシューマー(バックグラウンドタスクとして実行)
    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()を使用したステートマシン、およびCluster Shardingを介した分散システムを処理します。

---

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

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

// 悪い例: ロックを使用して共有ステートを保護
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); }
}

// 良い例: アクターまたはChannelを使用してアクセスをシリアライズ

手動スレッド管理

// 悪い例: スレッドを手動で作成
var thread = new Thread(() => ProcessOrders());
thread.Start();

// 良い例: Task.Runまたはより良い抽象化を使用
_ = Task.Run(() => ProcessOrdersAsync(cancellationToken));

非同期コードでのブロック

// 悪い例: 非同期をブロック - デッドロック風険!
var result = GetDataAsync().Result;

// 良い例: 非同期をずっと使用
var result = await GetDataAsync();

保護のない共有可変ステート

// 悪い例: 複数のタスクが共有ステートを変更
var results = new List<Result>();
await Parallel.ForEachAsync(items, async (item, ct) =>
{
    var result = await ProcessAsync(item, ct);
    results.Add(result); // レース条件!
});

// 良い例: 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(ここから始める)
    │
    ├─► 並列化が必要? → Parallel.ForEachAsync
    │
    ├─► プロデューサー・コンシューマーが必要? → Channel<T>
    │
    ├─► UIイベント合成が必要? → Reactive Extensions
    │
    ├─► サーバー側ストリーム処理が必要? → Akka.NET Streams
    │
    └─► ステートマシンまたはエンティティ管理が必要? → Akka.NET Actors

具体的なニーズがある場合のみエスカレーションしてください。 「念のため」でアクターやストリームを選ばないでください。