ByteFisher

钓鱼爱好者的编程世界

C# 高级教程 - 15 高级异步模式

发表于 分类于 ， Waline：
C#、高级: 基础教程中我们学会了 async/await 的基本用法。但在高性能场景下，不当的异步编程会导致性能反而比同步更差。本文将深入探讨 ValueTask、IAsyncEnumerable、Channel、管道模式、以及异步同步上下文等高级话题，帮助你编写真正高效的异步代码。

基础教程中我们学会了 async/await 的基本用法。但在高性能场景下，不当的异步编程会导致性能反而比同步更差。本文将深入探讨 ValueTask、IAsyncEnumerable、Channel、管道模式、以及异步同步上下文等高级话题，帮助你编写真正高效的异步代码。

ValueTask —— 减少异步分配

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
// Task 总是分配堆对象（即使同步完成）
public async Task<int> GetCachedValueAsync()
{
    if (_cache.TryGetValue("key", out int value))
        return value; // 仍然分配了 Task<int> 对象
    return await FetchFromDbAsync();
}

// ValueTask —— 同步完成时零分配
public ValueTask<int> GetCachedValueOptimizedAsync()
{
    if (_cache.TryGetValue("key", out int value))
        return new ValueTask<int>(value); // 零分配

    return new ValueTask<int>(FetchFromDbAsync());
}

// IAsyncStateMachine 约束：ValueTask 只能 await 一次
// 不能：await task; await task;（多次 await）
// 不能：Task.WhenAll(valueTask1, valueTask2)

何时使用 ValueTask

场景 推荐
方法经常同步完成（如缓存命中） ✅ ValueTask
热路径调用（每秒数万次） ✅ ValueTask
需要多次 await ❌ Task
需要 Task.WhenAll ❌ Task

IAsyncEnumerable —— 异步流

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
// 异步生成序列
async IAsyncEnumerable<int> ProduceNumbersAsync(
    [EnumeratorCancellation] CancellationToken token = default)
{
    for (int i = 0; i < 10; i++)
    {
        await Task.Delay(100, token);
        yield return i;
    }
}

// 消费异步流
await foreach (var number in ProduceNumbersAsync())
{
    Console.WriteLine(number);
}

// 带取消
using var cts = new CancellationTokenSource(500);
await foreach (var number in ProduceNumbersAsync(cts.Token))
{
    Console.WriteLine(number);
}

实际应用：分页数据加载

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
async IAsyncEnumerable<Order> GetOrdersPagedAsync(
    int pageSize = 100, [EnumeratorCancellation] CancellationToken token = default)
{
    int page = 0;
    bool hasMore = true;

    while (hasMore)
    {
        token.ThrowIfCancellationRequested();

        var pageData = await FetchPageAsync(page++, pageSize, token);
        foreach (var order in pageData)
            yield return order;

        hasMore = pageData.Count == pageSize;
    }
}

// 消费
await foreach (var order in GetOrdersPagedAsync().Take(500))
{
    Process(order);
}

异步 LINQ（System.Linq.Async）

1
2
3
4
5
6
7
8
// 需要安装: dotnet add package System.Linq.Async

using System.Linq;

var result = await GetOrdersPagedAsync()
    .WhereAwait(async order => await ValidateOrderAsync(order))
    .Select(order => new { order.Id, order.Amount })
    .ToListAsync();

Channel —— 生产者-消费者进阶

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
using System.Threading.Channels;

// 有界通道 —— 流量控制
var options = new BoundedChannelOptions(100)
{
    FullMode = BoundedChannelFullMode.Wait, // 满时等待
    SingleWriter = false,
    SingleReader = false,
};
var channel = Channel.CreateBounded<Order>(options);

// 无界通道 —— 无限增长（可能导致内存溢出）
var unbounded = Channel.CreateUnbounded<string>(new UnboundedChannelOptions
{
    SingleWriter = true, // 单生产者时优化
    SingleReader = true, // 单消费者时优化
});

// 批量消费者
async Task BatchConsumerAsync(ChannelReader<Order> reader, int batchSize = 10)
{
    var buffer = new List<Order>(batchSize);
    await foreach (var order in reader.ReadAllAsync())
    {
        buffer.Add(order);
        if (buffer.Count >= batchSize)
        {
            await ProcessBatchAsync(buffer);
            buffer.Clear();
        }
    }

    if (buffer.Count > 0)
        await ProcessBatchAsync(buffer);
}

管道模式（Pipeline）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
// 使用 Dataflow 或 Channel 构建处理管道
public class ImagePipeline
{
    private readonly Channel<string> _download = Channel.CreateBounded<string>(10);
    private readonly Channel<byte[]> _process = Channel.CreateBounded<byte[]>(10);
    private readonly Channel<string> _save = Channel.CreateBounded<string>(10);

    public async Task StartAsync(string[] urls)
    {
        var downloader = DownloadAsync();
        var processor = ProcessAsync();
        var saver = SaveAsync();

        foreach (var url in urls)
            await _download.Writer.WriteAsync(url);

        _download.Writer.Complete();

        await Task.WhenAll(downloader, processor, saver);
    }

    private async Task DownloadAsync()
    {
        await foreach (var url in _download.Reader.ReadAllAsync())
        {
            var data = await new HttpClient().GetByteArrayAsync(url);
            await _process.Writer.WriteAsync(data);
        }
        _process.Writer.Complete();
    }

    private async Task ProcessAsync()
    {
        await foreach (var data in _process.Reader.ReadAllAsync())
        {
            var processed = await ProcessImageDataAsync(data);
            await _save.Writer.WriteAsync($"processed_{Guid.NewGuid()}.jpg");
        }
        _save.Writer.Complete();
    }

    private async Task SaveAsync()
    {
        await foreach (var name in _save.Reader.ReadAllAsync())
        {
            Console.WriteLine($"保存: {name}");
        }
    }
}

ConfigureAwait 与同步上下文

1
2
3
4
5
6
7
8
9
10
11
12
13
14
// ConfigureAwait(false) —— 避免回到原始同步上下文
public async Task<string> ReadFileAsync(string path)
{
    using var file = File.OpenText(path);

    // 在 WinForms/WPF 中：await 后回到 UI 线程
    string content = await file.ReadToEndAsync().ConfigureAwait(false);

    // 此时在 ThreadPool 线程上
    return content;
}

// ASP.NET Core 无同步上下文，ConfigureAwait(false) 非必需但仍推荐
// 库代码必须使用 ConfigureAwait(false)，避免死锁

异步锁（SemaphoreSlim）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
public class AsyncResource
{
    private readonly SemaphoreSlim _semaphore = new(1, 1);

    public async Task AccessAsync()
    {
        await _semaphore.WaitAsync();
        try
        {
            await Task.Delay(100);
            Console.WriteLine("异步临界区");
        }
        finally
        {
            _semaphore.Release();
        }
    }
}

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

    public async Task<IDisposable> LockAsync()
    {
        await _semaphore.WaitAsync();
        return new Releaser(this);
    }

    private sealed class Releaser : IDisposable
    {
        private readonly AsyncLock _lock;
        public Releaser(AsyncLock l) => _lock = l;
        public void Dispose() => _lock._semaphore.Release();
    }
}

// 使用
private readonly AsyncLock _lock = new();
async Task SafeAccessAsync()
{
    using await _lock.LockAsync();
    // 异步安全的临界区
}

异步初始化与缓存

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
// 延迟异步初始化
public class AsyncInitializedService
{
    private readonly Lazy<Task<ExpensiveResource>> _resource;

    public AsyncInitializedService()
    {
        _resource = new Lazy<Task<ExpensiveResource>>(
            () => InitializeAsync(),
            LazyThreadSafetyMode.ExecutionAndPublication);
    }

    public async Task<ExpensiveResource> GetResourceAsync()
        => await _resource.Value;

    private async Task<ExpensiveResource> InitializeAsync()
    {
        await Task.Delay(1000); // 模拟耗时初始化
        return new ExpensiveResource();
    }
}

// 异步缓存
public class AsyncCache<TKey, TValue> where TKey : notnull
{
    private readonly ConcurrentDictionary<TKey, Task<TValue>> _cache = new();
    private readonly Func<TKey, Task<TValue>> _factory;

    public AsyncCache(Func<TKey, Task<TValue>> factory) => _factory = factory;

    public Task<TValue> GetOrAddAsync(TKey key)
    {
        return _cache.GetOrAdd(key, (k, fac) => fac(k), _factory);
    }
}

超时与取消模式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
// 带超时的异步操作
public async Task<T> WithTimeout<T>(Task<T> task, TimeSpan timeout)
{
    using var cts = new CancellationTokenSource(timeout);
    var completed = await Task.WhenAny(task, Task.Delay(timeout, cts.Token));

    if (completed == task)
        return await task; // 可能抛出异常，await 传播

    throw new TimeoutException($"操作超时 ({timeout.TotalSeconds}s)");
}

// 使用 CancellationTokenSource.CreateLinkedTokenSource
public async Task ProcessAsync(CancellationToken externalToken)
{
    using var timeoutCts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
    using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(
        externalToken, timeoutCts.Token);

    await LongOperationAsync(linkedCts.Token);
}

异步性能基准

1
2
3
4
// 小操作同步完成的场景（10 万次调用）
// 同步方法:          ~3ms (零分配)
// async Task 方法:  ~15ms (分配 Task 对象)
// async ValueTask:  ~4ms (零分配，同步路径)

总结：高级异步编程的核心是理解”分配”和”调度”。ValueTask 消除热路径的堆分配，Channel 提供高性能的生产者-消费者通信，ConfigureAwait(false) 避免死锁，而管道模式则让复杂的工作流变得清晰可控。掌握这些模式后，你的异步代码不仅能正确运行，更能以最少的资源消耗实现最大的吞吐量。

ByteFisher
分享编程技术 · 记录钓鱼乐趣
扫码关注
▸ 扫码关注 ◂
相关文章
C# 高级教程 - 20 性能优化与诊断2025-05-27C#、高级: 性能优化不是"过早优化"，而是基于数据的科学工程实践。.NET 生态提供了世界一流的诊断工具链——从 BenchmarkDotNet 到 OpenTelemetry，从 dotnet-trace 到 PerfView。本文将系统性地介绍性能分析、诊断工具和优化策略。C# 高级教程 - 19 互操作与源代码生成器2025-05-25C#、高级: 互操作（Interop）让 .NET 代码可以调用原生 C/C++ 库和 COM 组件，这在图像处理、硬件驱动、游戏引擎集成等场景中不可或缺。源代码生成器（Source Generator）则是 Roslyn 编译器的扩展能力，可以在编译时分析代码并生成额外的源文件。本文将深入探讨这两项技术。C# 高级教程 - 18 函数式编程2025-05-23C# 是一门多范式语言，在面向对象基础上不断吸收函数式编程（FP）的特性。从 LINQ 到模式匹配，从 record 到不可变数据，C# 的函数式表达能力持续增强。本文将深入探讨不可变性、纯函数、高阶函数、Either/Option 模式、以及如何在 C# 中实践函数式设计。C# 高级教程 - 17 高级设计模式2025-05-21C#、高级: 基础教程中我们了解过单例、工厂、观察者等经典设计模式。在实际的 C# 项目中，这些模式有更地道的 .NET 实现方式，同时也需要理解模式之间的组合与演进。本文将探讨设计模式在 C# 中的最佳实践以及 .NET 生态中的模式变体。C# 高级教程 - 16 依赖注入2025-05-19C#、高级: 依赖注入（DI）是现代 .NET 应用架构的核心模式。ASP.NET Core 内置了完整的 DI 容器，了解其工作原理、生命周期管理、高级注册方式和常见陷阱，对于构建可测试、可维护的企业级应用至关重要。本文将深入探讨 DI 容器的内部机制和最佳实践。C# 高级教程 - 14 网络编程2025-05-15C#、高级: 网络编程是后端开发的核心能力。C# 提供了从高层级的 HttpClient 到低层级的 Socket 的完整网络编程栈。本文将深入探讨 HTTP 请求调优、gRPC 通信、Socket 编程、以及 ASP.NET Core 中的网络性能优化。
C#高级教程 系列
第 15/20 篇
C# 高级教程 - 14 网络编程C# 高级教程 - 16 依赖注入
分享：
淡水鱼
淡水鱼
钓鱼爱好者 & Unity 游戏开发者，分享编程技术与钓鱼经验。