csharp-concurrency-patterns

.NET 并发：选择合适的工具

何时使用此技能

在以下情况下使用此技能：

• 决定如何处理 .NET 中的并发操作
• 评估是否使用 async/await、Channels、Akka.NET 或其他抽象
• 考虑使用锁、信号量或其他同步原语
• 需要处理具有背压、批处理或防抖功能的数据流
• 管理多个并发实体之间的状态

理念

从简单方案开始，仅在必要时升级。

大多数并发问题都可以通过 async/await 解决。只有在你有 async/await 无法妥善处理的特定需求时，才使用更复杂的工具。

尽量避免共享可变状态。 处理并发的最佳方式是从设计上消除并发问题。不可变数据、消息传递和隔离状态（例如 Actor）可以消除整类错误。

锁应当是例外，而非规则。 如果无法避免共享可变状态，偶尔使用锁也不是什么大问题。但如果你发现自己经常使用 lock、SemaphoreSlim 或其他同步原语，不妨退一步重新考虑设计。

当你确实需要共享可变状态时：

1. 首选： 重新设计以避免共享可变状态（不可变性、消息传递、Actor 隔离）
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（有状态并发）

适用于： 管理多个实体的状态、状态机、推送式更新、复杂协调、监督和容错。

每个实体一个 Actor 模式

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

只有在有明确需求时才升级。 不要为了“以防万一”而使用 actor 或流——从 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