.NET 同時実行: 適切なツールの選択肢を決める方法
どのようなときにこのスキルを使うか
このスキルは以下の状況で使用します：
- .NET で同時実行処理をどう扱うか決めるとき
- async/await、Channels、Akka.NET、その他の抽象化を使うか評価するとき
- ロック、セマフォ、その他の同期プリミティブを使いたくなったとき
- バックプレッシャー、バッチ処理、デバウンスを伴うデータストリームを処理する必要があるとき
- 複数の同時実行エンティティにまたがる状態を管理するとき
参考ファイル
- advanced-concurrency.md: Akka.NET Streams、Reactive Extensions、Akka.NET Actors（エンティティ毎アクター、ステートマシン、クラスタシャーディング）、非同期ローカル関数パターン
哲学
シンプルに始め、本当に必要なときだけ高度な手段に進む。
ほとんどの同時実行の問題は async/await で解決できます。async/await ではうまく扱えない明確な要件がある場合にのみ、より洗練されたツールを検討してください。
共有可変状態は避けるように設計する。 同時実行に対処する最善の方法は、設計段階でその問題を排除することです。イミュータブルなデータ、メッセージパッシング、独立した状態（アクターなど）は、バグのカテゴリ全体を排除します。
ロックは例外であり、原則ではない。 共有可変状態をどうしても避けられない場合：
- 最初の選択肢： 回避するように再設計（イミュータブル、メッセージパッシング、アクター分離）
- 次の選択肢：
System.Collections.Concurrentを使用（ConcurrentDictionary など） - その次の選択肢：
Channel<T>を使ってメッセージパッシングによりアクセスを直列化 - 最後の手段： 単純で短いクリティカルセクションに
lockを使用
決定木
何をしようとしているか？
│
├─► I/O（HTTP、データベース、ファイル）を待つ？
│ └─► async/await を使う
│
├─► コレクションを並列処理する（CPU バウンド）？
│ └─► Parallel.ForEachAsync を使う
│
├─► Producer/Consumer パターン（ワークキュー）？
│ └─► System.Threading.Channels を使う
│
├─► UI イベント処理（デバウンス、スロットル、結合）？
│ └─► Reactive Extensions (Rx) を使う
│
├─► サーバーサイドのストリーム処理（バックプレッシャー、バッチ）？
│ └─► Akka.NET Streams を使う
│
├─► 複雑な遷移を持つステートマシン？
│ └─► Akka.NET Actors (Become パターン) を使う
│
├─► 多数の独立したエンティティの状態を管理する？
│ └─► Akka.NET Actors (エンティティ毎アクター) を使う
│
├─► 複数の非同期操作を調整する？
│ └─► Task.WhenAll / Task.WhenAny を使う
│
└─► 上記のどれにも当てはまらない？
└─► 「本当に共有可変状態が必要か？」と自問する
├─► Yes → それを避けるように再設計を検討
└─► どうしても避けられない → 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（Producer/Consumer）
使用場面： ワークキュー、Producer/Consumer パターン、プロデューサーとコンシューマーの分離。
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（バックグラウンドタスクとして実行）
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); }
}
// 良い例: Actor または 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
│
├─► Producer/Consumer が必要？ → Channel<T>
│
├─► UI イベント合成が必要？ → Reactive Extensions
│
├─► サーバーサイドのストリーム処理が必要？ → Akka.NET Streams
│
└─► ステートマシンまたはエンティティ管理が必要？ → Akka.NET Actors
具体的な要件がある場合にのみエスカレーションする。 「念のため」にアクターやストリームに手を出さないように。