如何测试内部使用SemaphoreSlim实现并行限流的C#异步下载方法

现有方案的合理性评价 #

你现在用条件编译加内部事件的方案是可以跑通的，但存在几个明显的缺陷：

• 静态事件属于全局状态，多个单元测试并行执行时会出现事件触发串扰，必须额外加隔离逻辑才能保证测试结果准确
• 条件编译会导致测试构建和生产构建的执行逻辑存在差异，虽然差异很小，但仍存在测试结果无法完全代表生产行为的风险
• 额外增加了生产代码的复杂度，需要维护和业务逻辑无关的测试桩代码

无侵入的测试方案 #

完全不需要修改生产代码，只需要通过可控的successCallback就能完成并行度的所有断言，这也是测试这类限流逻辑的标准做法：

• 验证并行度上限：在回调中维护原子计数器，进入回调时递增，先阻塞回调的返回直到所有可进入的任务都触发回调，此时计数器的峰值就是实际的并行度，直接断言该值等于设置的maxDownloadsInParallel即可
• 验证非串行执行：给每个回调固定的延迟，统计总执行时间，如果总耗时远小于「任务数*单任务延迟」即可证明是并行执行

测试代码示例 #

[Test]
public async Task DownloadInParallel_RespectsMaxParallelismLimit()
{
    // 测试配置：10个任务，最大并行度3
    var testUris = Enumerable.Range(0, 10).Select(i => new Uri($"https://test.com/{i}"));
    const int expectedMaxParallelism = 3;
    var mockHandler = new MockHttpMessageHandler(); // 你现有用的模拟Handler即可
    var httpClient = new HttpClient(mockHandler);

    var currentRunning = 0;
    var peakRunning = 0;
    var releaseLatch = new TaskCompletionSource<VoidTaskResult>();

    Task TestSuccessCallback(Uri uri, int index, Stream stream)
    {
        var current = Interlocked.Increment(ref currentRunning);
        // 更新峰值
        if (current > Volatile.Read(ref peakRunning))
        {
            Volatile.Write(ref peakRunning, current);
        }
        // 阻塞直到手动释放
        return releaseLatch.Task;
    }

    // 启动下载任务不等待
    var downloadTask = testUris.DownloadInParallel(
        httpClient, 
        TestSuccessCallback, 
        expectedMaxParallelism, 
        CancellationToken.None);

    // 留足够时间让所有可进入的任务都触发回调
    await Task.Delay(1000);
    // 断言并行度符合预期
    Assert.That(peakRunning, Is.EqualTo(expectedMaxParallelism));

    // 释放所有阻塞的回调
    releaseLatch.SetResult(default);
    await downloadTask;
}

更适配的多线程构造 #

你目前使用SemaphoreSlim实现异步限流是完全合理的，这是.NET生态中处理异步并行限流的标准方案。如果想要减少样板代码、降低手动维护信号量的出错概率，可以选择以下封装好的官方实现：

1. TPL Dataflow ActionBlock #

属于微软官方的System.Threading.Tasks.Dataflow库，原生支持配置最大并行度，内部已经处理了限流、异常传播等逻辑，不需要自己维护SemaphoreSlim的生命周期和释放逻辑：

public static async Task DownloadInParallel(this IEnumerable<Uri> values, HttpClient httpClient, Func<Uri, int, Stream, Task> successCallback, int maxDownloadsInParallel, CancellationToken cancellationToken)
{
    var actionBlock = new ActionBlock<(Uri Uri, int Index)>(async item =>
    {
        using var response = await httpClient.GetAsync(item.Uri, HttpCompletionOption.ResponseHeadersRead, cancellationToken).ConfigureAwait(false);
        using var stream = await response.Content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false);
        await successCallback(item.Uri, item.Index, stream).ConfigureAwait(false);
    }, new ExecutionDataflowBlockOptions
    {
        MaxDegreeOfParallelism = maxDownloadsInParallel,
        CancellationToken = cancellationToken
    });

    var index = 0;
    foreach (var uri in values)
    {
        await actionBlock.SendAsync((uri, index++), cancellationToken).ConfigureAwait(false);
    }
    actionBlock.Complete();
    await actionBlock.Completion.ConfigureAwait(false);
}

2. System.Linq.Async ParallelForEachAsync #

属于System.Linq.Async库提供的扩展方法，专门针对异步枚举的并行执行，代码更加简洁：

public static async Task DownloadInParallel(this IEnumerable<Uri> values, HttpClient httpClient, Func<Uri, int, Stream, Task> successCallback, int maxDownloadsInParallel, CancellationToken cancellationToken)
{
    await values.ParallelForEachAsync(async (uri, index) =>
    {
        using var response = await httpClient.GetAsync(uri, HttpCompletionOption.ResponseHeadersRead, cancellationToken).ConfigureAwait(false);
        using var stream = await response.Content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false);
        await successCallback(uri, index, stream).ConfigureAwait(false);
    }, maxDegreeOfParallelism: maxDownloadsInParallel, cancellationToken: cancellationToken).ConfigureAwait(false);
}

内容的提问来源于stack exchange，提问作者Moss