csharp-concurrency-patterns

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

このスキルを使うタイミング

次のような場合にこのスキルを使います。

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

基本方針

シンプルに始め、必要な場合にだけ段階的に複雑にします。

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

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

ロックは例外的な手段であり、常用するものではありません。 共有可変状態を避けられない場合、必要に応じてロックを使うこと自体は問題ありません。ただし、頻繁に lock、SemaphoreSlim、その他の同期プリミティブを使おうとしているなら、一度立ち止まって設計を見直してください。

共有可変状態が本当に必要な場合：

1. 第一候補： それを避ける設計に変更します（不変性、メッセージパッシング、アクターによる分離）
2. 第二候補： System.Collections.Concurrent（ConcurrentDictionary、ConcurrentQueue など）を使います
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
│
├─► Need to protect shared mutable state with synchronization?
│   └─► Is the shared state a single scalar (int, long, reference)?
│       YES -> Use Interlocked (lock-free, lowest overhead)
│
│       Is the shared state a key-value lookup or queue?
│       YES -> Use ConcurrentDictionary / ConcurrentQueue (thread-safe by design)
│
│       Does the critical section contain `await`?
│       YES -> Use SemaphoreSlim (async-compatible via WaitAsync)
│       NO  -> Does the critical section need many readers, few writers?
│                YES -> Use ReaderWriterLockSlim (only if profiling shows lock contention)
│                NO  -> Use lock (simplest, lowest cognitive overhead)
│
│       Is the critical section extremely short (< 100 ns) with high contention?
│       YES -> Consider SpinLock (advanced, measure first)
│
└─► 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 バウンドの操作、ブロックしない待機、日常的な並行処理の大半。

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) を使います
• 非同期コードをブロックして待たないでください（.Result や .Wait() は使いません）

---

レベル 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);
        });
}

public async Task<IReadOnlyList<ProcessedImage>> ProcessImagesAsync(
    IEnumerable<string> imagePaths,
    CancellationToken ct)
{
    var results = new ConcurrentBag<ProcessedImage>();

    await Parallel.ForEachAsync(
        imagePaths,
        new ParallelOptions { MaxDegreeOfParallelism = 4, CancellationToken = ct },
        async (path, token) =>
        {
            var image = await File.ReadAllBytesAsync(path, token);
            var processed = ProcessImage(image);
            results.Add(processed);
        });

    return results.ToList();
}

使うべきでない場合：

• 純粋な I/O 操作（async/await で十分です）
• 順序が重要な場合（Parallel は順序を保持しません）
• バックプレッシャーやフロー制御が必要な場合

---

レベル 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();
}

public class WorkerPool
{
    private readonly Channel<WorkItem> _channel;
    private readonly List<Task> _workers = new();

    public WorkerPool(int workerCount)
    {
        _channel = Channel.CreateUnbounded<WorkItem>();

        for (int i = 0; i < workerCount; i++)
        {
            _workers.Add(Task.Run(() => ConsumeAsync()));
        }
    }

    private async Task ConsumeAsync()
    {
        await foreach (var item in _channel.Reader.ReadAllAsync())
        {
            await ProcessAsync(item);
        }
    }

    public ValueTask EnqueueAsync(WorkItem item)
        => _channel.Writer.WriteAsync(item);
}

Channels が適している用途：

• プロデューサーとコンシューマーの処理速度を分離する
• バックプレッシャーを伴う作業のバッファリング
• 複数のワーカーへのシンプルなファンアウト
• バックグラウンド処理キュー

Channels が適していない用途：

• 複雑なストリーム操作（バッチ処理、ウィンドウ処理、マージ）
• エンティティごとの状態を使った処理
• 高度なエラー処理や監督が必要な場合

---

レベル 4：Akka.NET Streams（複雑なストリーム処理）

用途： バックプレッシャー、バッチ処理、デバウンス、スロットリング、ストリームのマージ、複雑な変換。

using Akka.Streams;
using Akka.Streams.Dsl;

public Source<IReadOnlyList<Event>, NotUsed> BatchEvents(
    Source<Event, NotUsed> events)
{
    return events
        .GroupedWithin(100, TimeSpan.FromSeconds(1))
        .Select(batch => batch.ToList() as IReadOnlyList<Event>);
}

public Source<Request, NotUsed> ThrottleRequests(
    Source<Request, NotUsed> requests)
{
    return requests
        .Throttle(10, TimeSpan.FromSeconds(1), 5, ThrottleMode.Shaping);
}

public Source<ProcessedItem, NotUsed> ProcessWithParallelism(
    Source<Item, NotUsed> items)
{
    return items
        .SelectAsync(4, async item => await ProcessAsync(item));
}

public IRunnableGraph<Task<Done>> CreatePipeline(
    Source<RawEvent, NotUsed> events,
    Sink<ProcessedEvent, Task<Done>> sink)
{
    return events
        .Where(e => e.IsValid)
        .GroupedWithin(50, TimeSpan.FromMilliseconds(500))
        .SelectAsync(4, batch => ProcessBatchAsync(batch))
        .SelectMany(results => results)
        .ToMaterialized(sink, Keep.Right);
}

---

レベル 4b：Reactive Extensions（UI とイベントの合成）

用途： UI イベント処理、イベントストリームの合成、クライアントアプリケーションでの時間ベースの操作。

using System.Reactive.Linq;

public class SearchViewModel
{
    public SearchViewModel(ISearchService searchService)
    {
        SearchResults = SearchText
            .Throttle(TimeSpan.FromMilliseconds(300))
            .DistinctUntilChanged()
            .Where(text => text.Length >= 3)
            .SelectMany(text => searchService.SearchAsync(text).ToObservable())
            .ObserveOn(RxApp.MainThreadScheduler);
    }

    public IObservable<string> SearchText { get; }
    public IObservable<IList<SearchResult>> SearchResults { get; }
}

public IObservable<bool> CanSubmit =>
    Observable.CombineLatest(
        UsernameValid,
        PasswordValid,
        EmailValid,
        (user, pass, email) => user && pass && email);

public IObservable<Point> DoubleClicks =>
    MouseClicks
        .Buffer(TimeSpan.FromMilliseconds(300))
        .Where(clicks => clicks.Count >= 2)
        .Select(clicks => clicks.Last());

public IDisposable AutoSave =>
    DocumentChanges
        .Throttle(TimeSpan.FromSeconds(2))
        .Subscribe(async doc => await SaveAsync(doc));

Rx と Akka.NET Streams の比較：

シナリオ	Rx	Akka.NET Streams
UI イベント	最適な選択肢	過剰
クライアント側の合成	最適な選択肢	過剰
サーバー側のパイプライン	動作するが制約あり	より優れたバックプレッシャー
分散処理	その用途には設計されていない	この用途向けに構築
ホットオブザーバブル	ネイティブでサポート	追加のセットアップが必要

---

レベル 5：Akka.NET Actors（状態を持つ並行処理）

用途： 複数のエンティティの状態管理、ステートマシン、プッシュベースの更新、複雑な調整、監督とフォールトトレランス。

エンティティごとにアクターを割り当てるパターン

public class OrderActor : ReceiveActor
{
    private OrderState _state;

    public OrderActor(string orderId)
    {
        _state = new OrderState(orderId);

        Receive<AddItem>(msg =>
        {
            _state = _state.AddItem(msg.Item);
            Sender.Tell(new ItemAdded(msg.Item));
        });

        Receive<Checkout>(msg =>
        {
            if (_state.CanCheckout)
            {
                _state = _state.Checkout();
                Sender.Tell(new CheckoutSucceeded(_state.Total));
            }
            else
            {
                Sender.Tell(new CheckoutFailed("Cart is empty"));
            }
        });

        Receive<GetState>(_ => Sender.Tell(_state));
    }
}

Become を使ったステートマシン

public class PaymentActor : ReceiveActor
{
    private PaymentData _payment;

    public PaymentActor(string paymentId)
    {
        _payment = new PaymentData(paymentId);
        Pending();
    }

    private void Pending()
    {
        Receive<AuthorizePayment>(msg =>
        {
            _payment = _payment with { Amount = msg.Amount };
            Become(Authorizing);
            Self.Tell(new ProcessAuthorization());
        });

        Receive<CancelPayment>(_ =>
        {
            Become(Cancelled);
            Sender.Tell(new PaymentCancelled(_payment.Id));
        });
    }

    private void Authorizing()
    {
        Receive<ProcessAuthorization>(async _ =>
        {
            var result = await _gateway.AuthorizeAsync(_payment);
            if (result.Success)
            {
                _payment = _payment with { AuthCode = result.AuthCode };
                Become(Authorized);
            }
            else
            {
                Become(Failed);
            }
        });

        Receive<CancelPayment>(_ =>
        {
            Sender.Tell(new PaymentError("Cannot cancel during authorization"));
        });
    }

    private void Authorized()
    {
        Receive<CapturePayment>(_ =>
        {
            Become(Capturing);
            Self.Tell(new ProcessCapture());
        });

        Receive<VoidPayment>(_ =>
        {
            Become(Voiding);
            Self.Tell(new ProcessVoid());
        });
    }

    private void Capturing() { }
    private void Voiding() { }
    private void Cancelled() { }
    private void Failed() { }
}

---

同期プリミティブ

共有可変状態を使わなければならない場合は、要件を満たす最もシンプルなプリミティブを選んでください。

クイックリファレンステーブル

プリミティブ	非同期セーフ	再入可能	ユースケース
lock / Monitor	いいえ	はい（同じスレッド）	await を含まない短いクリティカルセクション
SemaphoreSlim	はい（WaitAsync）	いいえ	非同期対応の相互排他、スロットリング
Interlocked	該当なし（ロックフリー）	該当なし	スカラー値のアトミック操作（インクリメント、比較交換）
ConcurrentDictionary<K,V>	該当なし（スレッドセーフ）	該当なし	スレッドセーフなキーと値のキャッシュ／検索
ConcurrentQueue<T>	該当なし（スレッドセーフ）	該当なし	スレッドセーフな FIFO キュー
ReaderWriterLockSlim	いいえ	オプション（LockRecursionPolicy）	読み取りが多く書き込みが少ない場合（プロファイルに基づく場合のみ）
SpinLock	いいえ	いいえ	極端な競合が発生する状況でのごく短いクリティカルセクション

lock と Monitor

public sealed class Counter
{
    private readonly object _lock = new();
    private int _count;

    public void Increment()
    {
        lock (_lock)
        {
            _count++;
        }
    }

    public int GetCount()
    {
        lock (_lock)
        {
            return _count;
        }
    }
}

ロックオブジェクトのルール：

• 専用のプライベートな object フィールドを使用します
• this をロックしないでください
• typeof(T) をロックしないでください
• 文字列リテラルをロックしないでください
• 値型をロックしないでください

SemaphoreSlim

await をサポートする .NET 組み込みの同期プリミティブはこれだけです。

public sealed class AsyncCache
{
    private readonly SemaphoreSlim _semaphore = new(1, 1);
    private readonly Dictionary<string, object> _cache = new();

    public async Task<T> GetOrAddAsync<T>(string key,
        Func<CancellationToken, Task<T>> factory,
        CancellationToken ct = default)
    {
        await _semaphore.WaitAsync(ct);
        try
        {
            if (_cache.TryGetValue(key, out var existing))
                return (T)existing;

            var value = await factory(ct);
            _cache[key] = value!;
            return value;
        }
        finally
        {
            _semaphore.Release();
        }
    }
}

Interlocked 操作

スカラー値に対するロックフリーのアトミック操作：

private int _counter;
private long _totalBytes;
private object? _current;

Interlocked.Increment(ref _counter);
Interlocked.Decrement(ref _counter);
Interlocked.Add(ref _totalBytes, bytesRead);
var previous = Interlocked.Exchange(ref _current, newValue);
var original = Interlocked.CompareExchange(ref _counter, newValue: 10, comparand: 0);

ConcurrentDictionary

private readonly ConcurrentDictionary<int, Widget> _cache = new();

var widget = _cache.GetOrAdd(id, key => LoadWidget(key));
var updated = _cache.AddOrUpdate(id,
    addValueFactory: key => CreateDefault(key),
    updateValueFactory: (key, existing) => existing with { LastAccessed = DateTime.UtcNow });

if (_cache.TryRemove(id, out var removed))
{
}

重要： 競合が発生すると、GetOrAdd のファクトリーデリゲートは複数回実行されることがあります。厳密に一度だけ実行するには、Lazy<T> でラップしてください。

---

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

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

// 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: Blocking on async
var result = GetDataAsync().Result;
GetDataAsync().Wait();

// 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 or collect results differently
var results = new ConcurrentBag<Result>();

async メソッド内で lock を使用しない

lock はスレッドに紐づいています。await の後の継続処理は別のスレッドで再開される場合があり、その結果 SynchronizationLockException が発生することがあります。代わりに SemaphoreSlim.WaitAsync を使用してください。

---

非同期のローカル関数を優先する

Task.Run(async () => ...) や ContinueWith() の代わりに、非同期のローカル関数を使用してください。

private void HandleCommand(MyCommand cmd)
{
    async Task<WorkCompleted> ExecuteAsync()
    {
        var result = await DoWorkAsync();
        return new WorkCompleted(result);
    }

    ExecuteAsync().PipeTo(Self);
}

---

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

必要なこと	ツール	例
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	ヘルスチェック、ポーリング
単一のスカラー値を保護する	Interlocked	カウンター、フラグ
キーと値の状態を保護する	ConcurrentDictionary	キャッシュ、検索
非同期対応のミューテックス	SemaphoreSlim	非同期のクリティカルセクション
シンプルな同期ミューテックス	lock	await を含まない短いクリティカルセクション

---

段階的な選択肢

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

具体的な必要性がある場合にのみ、次の段階へ進んでください。 アクターやストリームを「念のため」に使わず、まずは async/await から始めて、よりシンプルな方法では対応できない場合にのみ次へ進んでください。

---

References

• Threading in C# (Joseph Albahari)
• Concurrency in C# Cookbook (Stephen Cleary)
• System.Threading.Interlocked
• ConcurrentDictionary best practices
• SemaphoreSlim class
• ReaderWriterLockSlim class
• Pattern Matching
• Span and Memory
• Async Best Practices