csharp-concurrency-patterns

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

何时使用此技能

使用此技能的场景：

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

参考文件

• advanced-concurrency.md：Akka.NET Streams、Reactive Extensions、Akka.NET Actors（每个 Actor 对应一个实体、状态机、集群分片）以及异步本地函数模式

理念

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

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

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

锁应是例外，而不是常规做法。 无法避免共享可变状态时：

1. 首选： 重新设计以避免共享状态（不可变性、消息传递、Actor 隔离）
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 可用于每个 Actor 对应一个实体的模式、通过 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

只有在有明确需求时才升级。 不要“以防万一”就使用 actors 或 streams。