异步编程最佳实践

SunnyFan大约 14 分钟约 4273 字

异步编程最佳实践

简介

C# 的 async/await 简化了异步编程，但不当使用会导致死锁、内存泄漏和性能问题。理解异步最佳实践，有助于编写高效可靠的异步代码。async/await 本质上是编译器将一个方法转换为状态机，通过回调方式避免阻塞线程。理解这个本质是掌握异步编程的关键。

本文从同步上下文与死锁、异常处理、取消模式、性能优化、资源管理五个维度全面讲解异步编程最佳实践。

特点

1.ConfigureAwait — 控制同步上下文
2.避免异步 void — 使用 async Task
3.取消令牌 — 支持协作式取消
4.ValueTask — 减少热路径的堆分配
5.异步锁 — SemaphoreSlim 替代 lock
6.异常处理 — 正确捕获和传播异步异常
7.并发控制 — 限流、超时、重试

异步编程核心原理

async/await 的本质

// async/await 不是"多线程"，而是"非阻塞等待"
// 当 await 一个未完成的 Task 时：
// 1. 当前方法的状态被保存到状态机
// 2. 控制权返回给调用者
// 3. Task 完成后，通过回调（延续）恢复执行
// 4. 恢复时可能回到原始同步上下文，也可能不回

// 简单示例 — 理解执行流程
async Task DemonstrateAsync()
{
    Console.WriteLine($"1. 线程: {Thread.CurrentThread.ManagedThreadId}");
    await Task.Delay(100); // 这里方法被挂起，线程被释放
    // 100ms 后，可能在另一个线程上恢复
    Console.WriteLine($"2. 线程: {Thread.CurrentThread.ManagedThreadId}");
}

// ==========================================
// 编译器生成的状态机（简化版）
// ==========================================
// 编译器将上面的方法大致转换为：
/*
struct DemonstrateAsyncStateMachine : IAsyncStateMachine
{
    public int _state;
    public AsyncTaskMethodBuilder _builder;
    private TaskAwaiter _awaiter;

    void IAsyncStateMachine.MoveNext()
    {
        switch (_state)
        {
            case 0:
                Console.WriteLine("1...");
                _awaiter = Task.Delay(100).GetAwaiter();
                if (!_awaiter.IsCompleted)
                {
                    _state = 1;
                    _builder.AwaitUnsafeOnCompleted(ref _awaiter, ref this);
                    return; // 控制权返回
                }
                goto case 1;
            case 1:
                Console.WriteLine("2...");
                _builder.SetResult();
                break;
        }
    }
}
*/

同步上下文详解

// SynchronizationContext 是理解异步的关键
// WinForms:   WindowsFormsSynchronizationContext — 所有回调回到 UI 线程
// WPF:        DispatcherSynchronizationContext — 所有回调回到 UI 线程
// ASP.NET Core: 没有 SynchronizationContext — 回调在线程池线程上执行
// Console:    没有 SynchronizationContext — 回调在线程池线程上执行

// 检查当前是否有同步上下文
var hasSyncContext = SynchronizationContext.Current != null;
Console.WriteLine($"有同步上下文: {hasSyncContext}");
// ASP.NET Core 和 Console 应用: false
// WinForms/WPF: true

// ==========================================
// 死锁的经典场景
// ==========================================
// 场景: ASP.NET Framework（有同步上下文）中
public class OrderController : ApiController
{
    // ❌ 死锁！
    public IHttpActionResult GetOrder(int id)
    {
        // 1. UI/请求线程调用 Result（阻塞线程）
        // 2. async 方法在后台完成，尝试回到同步上下文
        // 3. 但同步上下文的线程被 Result 阻塞了
        // 4. 互相等待 → 死锁
        var order = _service.GetOrderAsync(id).Result; // 死锁！
        return Ok(order);
    }
}

// ✅ 正确做法 — 全链路 async
public async Task<IHttpActionResult> GetOrder(int id)
{
    var order = await _service.GetOrderAsync(id);
    return Ok(order);
}

// ✅ 如果必须同步等待（极少数情况）
public IHttpActionResult GetOrder(int id)
{
    var order = _service.GetOrderAsync(id)
        .GetAwaiter().GetResult(); // 比 .Result 更好，异常不被包装
    return Ok(order);
}
// GetAwaiter().GetResult() 不会导致死锁吗？
// 如果 async 方法内部使用了 ConfigureAwait(false)，就不会死锁
// 因为 async 方法不会尝试回到同步上下文

实现

ConfigureAwait 与同步上下文

// 库代码始终使用 ConfigureAwait(false)
public class MyService
{
    public async Task<string> GetDataAsync()
    {
        // 不捕获 SynchronizationContext，避免死锁
        using var client = new HttpClient();
        var json = await client.GetStringAsync("https://api.example.com/data")
            .ConfigureAwait(false);
        return json;
    }
}

// UI/WPF 应用仅在需要回到 UI 线程时不用 ConfigureAwait(false)
public class ViewModel
{
    public async Task LoadDataAsync()
    {
        var data = await _service.GetDataAsync(); // 回到 UI 线程
        Label.Text = data; // 安全访问 UI 控件
    }
}

// ==========================================
// ConfigureAwait(false) 的完整理解
// ==========================================
// ConfigureAwait(false) 做了两件事：
// 1. 不捕获 SynchronizationContext（不尝试回到原始上下文）
// 2. 完成后在线程池线程上继续执行

// 好处：
// - 避免死锁
// - 避免不必要的线程切换（性能更好）
// - 减少死锁风险

// 什么时候用 ConfigureAwait(false)：
// - 所有库代码（DLL、NuGet 包）
// - ASP.NET Core 应用（反正没有同步上下文）
// - 后台服务、控制台应用

// 什么时候不用 ConfigureAwait(false)：
// - UI 代码中需要访问 UI 控件时
// - 需要特定线程上下文的代码

// .NET 8+ 的项目级配置 — 全局禁用同步上下文
// 在 .csproj 中添加（仅影响 ASP.NET Core）
/*
<PropertyGroup>
  <EnableConfigureAwaitAnalyzer>true</EnableConfigureAwaitAnalyzer>
</PropertyGroup>
*/

// ==========================================
// ConfigureAwait 的链式问题
// ==========================================
// ConfigureAwait 只影响单个 await
// 如果链中有多个 await，每个都需要加
public async Task<string> ChainedAsync()
{
    var step1 = await Step1Async().ConfigureAwait(false);
    var step2 = await Step2Async(step1).ConfigureAwait(false);
    var step3 = await Step3Async(step2).ConfigureAwait(false);
    return step3;
}

// 使用 Task.WhenAll 并行执行
public async Task<string> ParallelAsync()
{
    var task1 = Step1Async();
    var task2 = Step2Async();
    var task3 = Step3Async();

    await Task.WhenAll(task1, task2, task3).ConfigureAwait(false);

    return $"{task1.Result}, {task2.Result}, {task3.Result}";
}

避免 Async Void

// ❌ 异步 void — 异常无法捕获，会导致进程崩溃
public async void Button_Click(object sender, EventArgs e)
{
    await DoSomethingAsync(); // 异常直接抛到 SynchronizationContext
}

// ✅ 异步 Task — 异常可捕获
public async Task Button_ClickAsync(object sender, EventArgs e)
{
    try { await DoSomethingAsync(); }
    catch (Exception ex) { Logger.Error(ex); }
}

// 唯一允许 async void 的场景: 事件处理器（确保内部 try-catch）
public async void OnLoaded(object sender, RoutedEventArgs e)
{
    try { await InitializeAsync(); }
    catch (Exception ex) { MessageBox.Show(ex.Message); }
}

// ==========================================
// async void 的具体问题
// ==========================================
// 问题 1: 调用者无法知道何时完成
async void FireAndForget()
{
    await Task.Delay(1000);
    Console.WriteLine("完成"); // 调用者无法 await 这个方法
}

// 问题 2: 异常直接导致进程崩溃（在同步上下文上）
async void CrashOnException()
{
    throw new InvalidOperationException("崩溃！");
    // 在 WinForms/WPF 中会弹出未处理异常对话框
    // 在 ASP.NET 中会导致进程终止
}

// 问题 3: 无法组合
// 你不能 await 一个 void 返回的方法
// 你不能用 Task.WhenAll 组合多个 void 方法

// ==========================================
// 安全的 Fire-and-Forget 模式
// ==========================================
public static class FireAndForgetExtensions
{
    // 安全的 fire-and-forget，捕获异常并记录
    public static async void SafeFireAndForget(
        this Task task,
        Action<Exception>? errorHandler = null)
    {
        try
        {
            await task.ConfigureAwait(false);
        }
        catch (Exception ex)
        {
            errorHandler?.Invoke(ex);
            // 或者使用全局异常日志
            Console.Error.WriteLine($"Fire-and-forget 异常: {ex}");
        }
    }
}

// 使用
public void StartBackgroundWork()
{
    DoBackgroundWorkAsync().SafeFireAndForget(
        ex => _logger.LogError(ex, "后台任务失败"));
}

// ==========================================
// 事件处理器的异步包装
// ==========================================
public class AsyncEventHandler
{
    public event Func<Task>? SomethingHappened;

    // 触发事件并等待所有处理器完成
    public async Task RaiseEventAsync()
    {
        if (SomethingHappened is null) return;

        var handlers = SomethingHappened.GetInvocationList()
            .Cast<Func<Task>>()
            .ToArray();

        var tasks = handlers.Select(h => h()).ToArray();
        await Task.WhenAll(tasks);
    }
}

取消令牌传播

public class OrderService
{
    // 所有异步方法接受 CancellationToken
    public async Task<Order> CreateOrderAsync(CreateOrderRequest request, CancellationToken ct)
    {
        ct.ThrowIfCancellationRequested();

        var customer = await _customerRepo.GetAsync(request.CustomerId, ct);
        var items = await _inventoryRepo.ReserveAsync(request.Items, ct);
        var payment = await _paymentService.ChargeAsync(customer, items.Total, ct);

        return new Order(customer.Id, items, payment.Id);
    }
}

// ASP.NET Core 自动传播取消
[HttpPost("/orders")]
public async Task<IActionResult> Create(
    [FromBody] CreateOrderRequest request,
    CancellationToken ct) // 从 HttpContext 自动获取
{
    var order = await _orderService.CreateOrderAsync(request, ct);
    return Created($"/orders/{order.Id}", order);
}

// 超时取消
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
var result = await _service.LongRunningAsync(cts.Token);

// ==========================================
// CancellationToken 的使用模式
// ==========================================

// 模式 1: 链式取消源
using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(
    externalToken, timeoutToken);
// 任一 token 取消，linkedCts 都会被取消

// 模式 2: 带取消的延迟
try
{
    await Task.Delay(5000, cancellationToken);
}
catch (OperationCanceledException)
{
    Console.WriteLine("操作被取消");
}

// 模式 3: 取消后清理
public async Task ProcessBatchAsync(
    IReadOnlyList<Item> items,
    CancellationToken ct)
{
    var processedItems = new List<Item>();
    try
    {
        foreach (var item in items)
        {
            ct.ThrowIfCancellationRequested();
            await ProcessItemAsync(item, ct);
            processedItems.Add(item);
        }
    }
    catch (OperationCanceledException)
    {
        // 清理已处理的部分结果
        _logger.LogInformation(
            "已取消，已处理 {Count}/{Total} 项",
            processedItems.Count, items.Count);
        await RollbackAsync(processedItems);
        throw; // 重新抛出，让上层知道操作被取消
    }
}

// 模式 4: 取消感知的轮询
public async Task<T> PollUntilReadyAsync<T>(
    Func<Task<T>> checkFunc,
    TimeSpan interval,
    CancellationToken ct)
{
    while (!ct.IsCancellationRequested)
    {
        var result = await checkFunc();
        if (result is not null)
            return result;

        try
        {
            await Task.Delay(interval, ct);
        }
        catch (OperationCanceledException)
        {
            ct.ThrowIfCancellationRequested();
        }
    }

    throw new OperationCanceledException(ct);
}

// 模式 5: 取消令牌注册回调
var cts = new CancellationTokenSource();
cts.Token.Register(() =>
{
    Console.WriteLine("取消回调触发 — 释放资源");
    _resource.Release();
});

ValueTask 优化热路径

// ValueTask 避免异步完成时的 Task 堆分配
public class CacheService
{
    private readonly ConcurrentDictionary<string, string> _cache = new();

    // 同步完成不分配 Task 对象
    public ValueTask<string> GetOrAddAsync(string key, Func<string, Task<string>> factory)
    {
        if (_cache.TryGetValue(key, out var cached))
            return new ValueTask<string>(cached); // 同步返回，零分配

        return new ValueTask<string>(FactoryCore(key, factory));
    }

    private async Task<string> FactoryCore(string key, Func<string, Task<string>> factory)
    {
        var value = await factory(key);
        _cache[key] = value;
        return value;
    }
}

// ==========================================
// ValueTask 的使用规则
// ==========================================
// 规则 1: ValueTask 只能 await 一次
// ValueTask<string> vt = GetOrAddAsync("key");
// await vt; // 正确
// await vt; // 未定义行为！可能抛异常或返回错误结果

// 如果需要多次 await，先转为 Task
ValueTask<string> vt = GetOrAddAsync("key");
Task<string> task = vt.AsTask(); // 确保只转换一次
var result1 = await task;
var result2 = await task; // 安全

// 规则 2: 不要在没有 await 的情况下返回 ValueTask
// 反面示例
public ValueTask<int> BadMethod()
{
    // 如果调用者没有 await，ValueTask 的内部 Task 不会被释放
    // 可能导致资源泄漏
    return DoSomethingAsync();
}

// 规则 3: ValueTask 适合"大部分情况同步完成"的场景
// 如果大部分情况都是异步完成，直接用 Task 更简单

// ==========================================
// ValueTask 的实际应用 — 缓存读取
// ==========================================
public class DataCache
{
    private readonly ConcurrentDictionary<string, Data> _cache = new();
    private readonly SemaphoreSlim _loadLock = new(1, 1);

    public ValueTask<Data> GetDataAsync(string key)
    {
        // 热路径: 缓存命中，同步返回
        if (_cache.TryGetValue(key, out var data))
            return new ValueTask<Data>(data);

        // 冷路径: 缓存未命中，异步加载
        return new ValueTask<Data>(LoadAsync(key));
    }

    private async Task<Data> LoadAsync(string key)
    {
        await _loadLock.WaitAsync();
        try
        {
            // 双重检查锁定
            if (_cache.TryGetValue(key, out var cached))
                return cached;

            var data = await FetchFromDatabaseAsync(key);
            _cache[key] = data;
            return data;
        }
        finally
        {
            _loadLock.Release();
        }
    }

    private async Task<Data> FetchFromDatabaseAsync(string key)
    {
        await Task.Delay(100); // 模拟数据库查询
        return new Data(key);
    }
}

record Data(string Key);

异步锁与并发控制

// ❌ lock 不能用于异步代码
// lock (_sync) { await DoAsync(); } // 编译错误

// ✅ SemaphoreSlim
public class AsyncLock
{
    private readonly SemaphoreSlim _semaphore = new(1, 1);

    public async Task<IDisposable> LockAsync(CancellationToken ct = default)
    {
        await _semaphore.WaitAsync(ct);
        return new Releaser(_semaphore);
    }

    private class Releaser : IDisposable
    {
        private readonly SemaphoreSlim _sem;
        public Releaser(SemaphoreSlim sem) => _sem = sem;
        public void Dispose() => _sem.Release();
    }
}

// 使用
using (await _asyncLock.LockAsync())
{
    await UpdateSharedResourceAsync();
}

// ==========================================
// 异步互斥锁的完整实现
// ==========================================
public sealed class AsyncMutex : IDisposable
{
    private readonly SemaphoreSlim _semaphore = new(1, 1);
    private int _disposed;

    public async Task<IDisposable> AcquireAsync(CancellationToken ct = default)
    {
        ObjectDisposedException.ThrowIf(_disposed != 0, this);
        await _semaphore.WaitAsync(ct);
        return new MutexReleaser(this);
    }

    public void Dispose()
    {
        if (Interlocked.Exchange(ref _disposed, 1) == 0)
        {
            _semaphore.Dispose();
        }
    }

    private sealed class MutexReleaser : IDisposable
    {
        private readonly AsyncMutex _owner;
        private int _released;

        public MutexReleaser(AsyncMutex owner) => _owner = owner;

        public void Dispose()
        {
            if (Interlocked.Exchange(ref _released, 1) == 0)
            {
                _owner._semaphore.Release();
            }
        }
    }
}

// ==========================================
// 并发限流器
// ==========================================
public class ConcurrencyLimiter
{
    private readonly SemaphoreSlim _semaphore;

    public ConcurrencyLimiter(int maxConcurrency)
    {
        _semaphore = new SemaphoreSlim(maxConcurrency, maxConcurrency);
    }

    public async Task<IDisposable> AcquireAsync(CancellationToken ct = default)
    {
        await _semaphore.WaitAsync(ct);
        return new Releaser(_semaphore);
    }

    private class Releaser : IDisposable
    {
        private readonly SemaphoreSlim _sem;
        public Releaser(SemaphoreSlim sem) => _sem = sem;
        public void Dispose() => _sem.Release();
    }
}

// 使用 — 限制同时执行的任务数
public async Task ProcessAllAsync(IEnumerable<WorkItem> items)
{
    using var limiter = new ConcurrencyLimiter(maxConcurrency: 5);

    var tasks = items.Select(async item =>
    {
        using (await limiter.AcquireAsync())
        {
            await ProcessItemAsync(item);
        }
    });

    await Task.WhenAll(tasks);
}

// ==========================================
// 异步重试策略
// ==========================================
public static class AsyncRetry
{
    public static async Task<T> ExecuteAsync<T>(
        Func<Task<T>> action,
        int maxRetries = 3,
        TimeSpan? delay = null,
        CancellationToken ct = default)
    {
        delay ??= TimeSpan.FromSeconds(1);

        for (int attempt = 0; attempt <= maxRetries; attempt++)
        {
            try
            {
                return await action();
            }
            catch (Exception ex) when (attempt < maxRetries && !IsFatal(ex))
            {
                var actualDelay = delay.Value * Math.Pow(2, attempt); // 指数退避
                _logger.LogWarning(
                    ex, "第 {Attempt} 次重试，等待 {Delay}ms",
                    attempt + 1, actualDelay.TotalMilliseconds);

                await Task.Delay(actualDelay, ct);
            }
        }

        throw new InvalidOperationException("超过最大重试次数");
    }

    private static bool IsFatal(Exception ex)
        => ex is OutOfMemoryException or StackOverflowException;
}

// 使用
var result = await AsyncRetry.ExecuteAsync(
    () => _api.FetchDataAsync(),
    maxRetries: 3,
    delay: TimeSpan.FromSeconds(2));

异步异常处理模式

// ==========================================
// 异步异常的传播行为
// ==========================================
// async Task 方法: 异常被捕获在 Task 对象中
// await 时异常被重新抛出
// Task.Exception 包含所有异常（AggregateException）

// 当多个异常同时发生时
public async Task MultiExceptionAsync()
{
    var task1 = ThrowAsync("错误1");
    var task2 = ThrowAsync("错误2");

    try
    {
        await Task.WhenAll(task1, task2);
    }
    catch (Exception ex)
    {
        // ex 是第一个异常
        // 要获取所有异常，需要检查 Task.Exception
        Console.WriteLine($"捕获: {ex.Message}");
    }
}

static async Task ThrowAsync(string message)
{
    await Task.Yield();
    throw new InvalidOperationException(message);
}

// ==========================================
// 正确的异常处理策略
// ==========================================

// 策略 1: 按异常类型分别处理
try
{
    await _service.DoWorkAsync(ct);
}
catch (OperationCanceledException)
{
    // 取消是预期行为，不需要记录为错误
    _logger.LogInformation("操作被用户取消");
}
catch (HttpRequestException ex) when (ex.StatusCode == System.Net.HttpStatusCode.NotFound)
{
    // 404 是业务逻辑
    return NotFound();
}
catch (Exception ex)
{
    // 其他异常记录并传播
    _logger.LogError(ex, "操作失败");
    throw;
}

// 策略 2: 捕获聚合异常中的所有异常
public async Task ProcessWithAllErrorsAsync()
{
    var tasks = items.Select(ProcessItemAsync).ToArray();
    var allTasks = Task.WhenAll(tasks);

    try
    {
        await allTasks;
    }
    catch
    {
        // 收集所有失败的异常
        var exceptions = allTasks.Exception!.InnerExceptions;
        foreach (var ex in exceptions)
            _logger.LogError(ex, "子任务失败");

        throw new AggregateException("多个子任务失败", exceptions);
    }
}

// 策略 3: 使用 ExceptionFilter (.NET 6+)
try
{
    await riskyOperation();
}
catch (Exception ex) when (LogAndCheckIfHandled(ex))
{
    // 只有 when 返回 true 时才进入此 catch 块
    // 且异常不会被视为"已处理"，栈不会被展开
}

static bool LogAndCheckIfHandled(Exception ex)
{
    _logger.LogError(ex, "发生异常");
    return false; // 不处理，继续抛出
}

// ==========================================
// Task 的 IsFaulted 和 IsCanceled
// ==========================================
var task = DoSomethingAsync();

try
{
    await task;
}
catch
{
    // 已被 await 处理
}

// 或者不 await，直接检查状态
if (task.IsFaulted)
{
    _logger.LogError(task.Exception, "任务失败");
}
else if (task.IsCanceled)
{
    _logger.LogInformation("任务被取消");
}

优点

1.避免死锁 — ConfigureAwait(false) 防止死锁
2.可靠取消 — CancellationToken 协作取消
3.性能优化 — ValueTask 减少分配
4.异常安全 — async Task 可捕获异常
5.资源高效 — 非阻塞避免线程浪费
6.可组合 — Task.WhenAll/WhenAny 灵活组合

缺点

1.心智负担 — 需理解同步上下文
2.调试困难 — 异步堆栈不易阅读
3.传染性 — async 传染调用链
4.过度异步 — 简单同步操作不应异步
5.隐藏成本 — 状态机分配、闭包捕获
6.取消复杂 — 取消传播需要全链路支持

总结

库代码始终使用 ConfigureAwait(false) 避免 SynchronizationContext 死锁。避免 async void，使用 async Task。所有异步方法接受 CancellationToken 参数。热路径使用 ValueTask 减少分配。异步锁使用 SemaphoreSlim。建议在项目中统一异步编程规范，团队共享最佳实践文档。

核心原则总结：