限制异步任务

我想运行一堆异步任务，并限制在任何给定时间可以等待完成的任务数量。

假设您有 1000 个 URL，并且您只想一次打开 50 个请求；但是，一旦一个请求完成，您就会打开与列表中下一个 URL 的连接。这样，每次始终打开 50 个连接，直到 URL 列表耗尽为止。

如果可能的话，我还想利用给定数量的线程。

我想出了一个扩展方法，ThrottleTasksAsync这就是我想要的。是否已经有更简单的解决方案？我认为这是一个常见的情况。

Usage:

class Program
{
    static void Main(string[] args)
    {
        Enumerable.Range(1, 10).ThrottleTasksAsync(5, 2, async i => { Console.WriteLine(i); return i; }).Wait();

        Console.WriteLine("Press a key to exit...");
        Console.ReadKey(true);
    }
}

这是代码：

static class IEnumerableExtensions
{
    public static async Task<Result_T[]> ThrottleTasksAsync<Enumerable_T, Result_T>(this IEnumerable<Enumerable_T> enumerable, int maxConcurrentTasks, int maxDegreeOfParallelism, Func<Enumerable_T, Task<Result_T>> taskToRun)
    {
        var blockingQueue = new BlockingCollection<Enumerable_T>(new ConcurrentBag<Enumerable_T>());

        var semaphore = new SemaphoreSlim(maxConcurrentTasks);

        // Run the throttler on a separate thread.
        var t = Task.Run(() =>
        {
            foreach (var item in enumerable)
            {
                // Wait for the semaphore
                semaphore.Wait();
                blockingQueue.Add(item);
            }

            blockingQueue.CompleteAdding();
        });

        var taskList = new List<Task<Result_T>>();

        Parallel.ForEach(IterateUntilTrue(() => blockingQueue.IsCompleted), new ParallelOptions { MaxDegreeOfParallelism = maxDegreeOfParallelism },
        _ =>
        {
            Enumerable_T item;

            if (blockingQueue.TryTake(out item, 100))
            {
                taskList.Add(
                    // Run the task
                    taskToRun(item)
                    .ContinueWith(tsk =>
                        {
                            // For effect
                            Thread.Sleep(2000);

                            // Release the semaphore
                            semaphore.Release();

                            return tsk.Result;
                        }
                    )
                );
            }
        });

        // Await all the tasks.
        return await Task.WhenAll(taskList);
    }

    static IEnumerable<bool> IterateUntilTrue(Func<bool> condition)
    {
        while (!condition()) yield return true;
    }
}

该方法利用BlockingCollection and SemaphoreSlim使其发挥作用。节流器在一个线程上运行，所有异步任务在另一线程上运行。为了实现并行性，我添加了一个 maxDegreeOfParallelism 参数，该参数传递给Parallel.ForEach循环重新用作while loop.

旧版本是：

foreach (var master = ...)
{
    var details = ...;
    Parallel.ForEach(details, detail => {
        // Process each detail record here
    }, new ParallelOptions { MaxDegreeOfParallelism = 15 });
    // Perform the final batch updates here
}

但是，线程池很快就会耗尽，你不能这样做async/await.

Bonus:为了解决这个问题BlockingCollection抛出异常的地方Take() when CompleteAdding()被称为，我正在使用TryTake超时过载。如果我没有使用超时TryTake，这会违背使用的目的BlockingCollection since TryTake不会阻止。有没有更好的办法？理想情况下，会有一个TakeAsync method.

按照建议，使用TPL数据流 https://learn.microsoft.com/en-us/dotnet/standard/parallel-programming/dataflow-task-parallel-library.

A TransformBlock<TInput, TOutput> https://learn.microsoft.com/en-us/dotnet/api/system.threading.tasks.dataflow.transformblock-2可能就是您正在寻找的。

你定义一个MaxDegreeOfParallelism限制可以并行转换的字符串数量（即可以下载多少个 url）。然后，您将 URL 发布到该块，完成后，您告诉该块您已完成添加项目并获取响应。

var downloader = new TransformBlock<string, HttpResponse>(
        url => Download(url),
        new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 50 }
    );

var buffer = new BufferBlock<HttpResponse>();
downloader.LinkTo(buffer);

foreach(var url in urls)
    downloader.Post(url);
    //or await downloader.SendAsync(url);

downloader.Complete();
await downloader.Completion;

IList<HttpResponse> responses;
if (buffer.TryReceiveAll(out responses))
{
    //process responses
}

注：TransformBlock缓冲其输入和输出。那么，为什么我们需要将其链接到BufferBlock?

因为TransformBlock直到所有项目（HttpResponse）已被消耗，并且await downloader.Completion会挂起。相反，我们让downloader将其所有输出转发到专用缓冲区块 - 然后我们等待downloader完成并检查缓冲区块。