批量异步调用，在没有信号量的情况下限制并发调用的数量

相当简单的任务 - 有 N 个有序请求必须异步执行，同时执行的请求数量有限制，然后从它们中取出它们以及 N 个有序响应。

实际上，当有很多请求时，这可以与网络或数据库一起使用，但您不想为您的计算机、网络或服务器安排灾难。因此，对同时活动的异步任务的数量进行了限制。

正面的解决方案看起来很明显。

public async Task<Tout[]> RunSemaphoreAsync<Tin, Tout>(IEnumerable<Tin> items, Func<Tin, Task<Tout>> func, int degree)
{
    using SemaphoreSlim semaphore = new(degree);
    return await Task.WhenAll(items.Select(async item => {
        await semaphore.WaitAsync();
        try
        {
            return await func(item);
        }
        finally
        {
            semaphore.Release();
        }
    }));
}

使用示例：

让它成为一项接受某事、做某事并回馈某事的工作任务。

public async Task<int> RunJobAsync(int n)
{
    await Task.Yield(); 
    return n + 1;
}

这是一个例子。

IEnumerable<int> numbers = Enumerable.Range(0, 100);
int[] result = await RunSemaphoreAsync(numbers, RunJobAsync, Environment.ProcessorCount * 2);
Console.WriteLine(string.Join(",", result));

一切都像发条一样，快速且按预期工作。

上面的一切都很好，但是有一天，当我再次阅读生产者/消费者模式的各种实现时，我想到了使用工人而不是信号量的想法。其实，为什么不呢。

这是方法。

public async Task<Tout[]> RunWorkersAsync<Tin, Tout>(IEnumerable<Tin> items, Func<Tin, Task<Tout>> func, int degree)
{
    List<Task<Tout>> tasks = new();
    using (var source = items.GetEnumerator())
    {
        Task[] jobs = new Task[degree];
        for (int i = 0; i < degree; i++)
        {
            jobs[i] = ((Func<Task>)(async () =>
            {
                while (true)
                {
                    Task<Tout> task;
                    lock (source)
                    {
                        if (source.MoveNext())
                        {
                            task = func(source.Current);
                            tasks.Add(task);
                        }
                        else
                            break;
                    }
                    await task;
                }
            }))();
        }
        await Task.WhenAll(jobs);
    }
    return tasks.Select(t => t.Result).ToArray();
}

和第一个候选人一样漂亮。那么什么更好呢？

我决定测量开销。
我不是编写基准测试的高手，但这什么时候阻止了任何人？:)

class Program
{
    static void Main(string[] args)
    {
        var result = BenchmarkRunner.Run<MyBenchmarks>();
        Console.ReadKey();
    }
}

[MemoryDiagnoser]
public class MyBenchmarks
{
    private readonly List<int> numbers = Enumerable.Range(0, 2000).ToList();
    private readonly int degree = Environment.ProcessorCount * 2;

    [Benchmark]
    public Task SemaphoreTest()
    {
        return RunSemaphoreAsync(numbers, RunJobAsync, degree);
    }

    [Benchmark]
    public Task WorkersTest()
    {
        return RunWorkersAsync(numbers, RunJobAsync, degree);
    }

    public async Task<int> RunJobAsync(int n)
    {
        await Task.Yield(); 
        return n + 1;
    }

    public async Task<Tout[]> RunSemaphoreAsync<Tin, Tout>(IEnumerable<Tin> items, Func<Tin, Task<Tout>> func, int degree)
    {
        using SemaphoreSlim semaphore = new(degree);
        return await Task.WhenAll(items.Select(async item => {
            await semaphore.WaitAsync();
            try
            {
                return await func(item);
            }
            finally
            {
                semaphore.Release();
            }
        }));
    }

    public async Task<Tout[]> RunWorkersAsync<Tin, Tout>(IEnumerable<Tin> items, Func<Tin, Task<Tout>> func, int degree)
    {
        List<Task<Tout>> tasks = new();
        using (var source = items.GetEnumerator())
        {
            Task[] jobs = new Task[degree];
            for (int i = 0; i < degree; i++)
            {
                jobs[i] = ((Func<Task>)(async () =>
                {
                    while (true)
                    {
                        Task<Tout> task;
                        lock (source)
                        {
                            if (source.MoveNext())
                            {
                                task = func(source.Current);
                                tasks.Add(task);
                            }
                            else
                                break;
                        }
                        await task;
                    }
                }))();
            }
            await Task.WhenAll(jobs);
        }
        return tasks.Select(t => t.Result).ToArray();
    }
}

在这里我得到了一个有趣的结果

BenchmarkDotNet=v0.13.0, OS=Windows 10.0.19043.1081 (21H1/May2021Update)
Intel Core i7-4700HQ CPU 2.40GHz (Haswell), 1 CPU, 8 logical and 4 physical cores
.NET SDK=5.0.301
  [Host]     : .NET 5.0.7 (5.0.721.25508), X64 RyuJIT
  DefaultJob : .NET 5.0.7 (5.0.721.25508), X64 RyuJIT

眼睛拒绝相信。为什么带有信号量的方法只是像这样合并，或者可能是一个歪曲的测试或歪曲的实现？请考虑。

PS我没有立即在SO上运行，但首先我寻找关于缓慢的投诉SemaphoreSlim......并没有找到它，但我发现了这个：https ://github.com/dotnet/runtime/pull/55262 。换句话说，红绿灯在 .NET 6 中会飞一点。