分组并行执行任务

Question

我用一个简单的例子来描述我的问题，然后描述一个更接近的问题。

想象我们在 box1 中有 n 个项目 [i1,i2,i3,i4,...,in] 并且我们有一个 box2 可以处理 m 个项目来执行它们（m 通常远小于 n）。每个项目所需的时间是不同的。我想一直做 m 个工作项目，直到所有项目都进行完。

一个更接近的问题是，例如，您有一个包含 n 个文件字符串（URL 地址）的 list1，我们希望有一个系统可以同时下载 m 个文件（例如通过 httpclient.getAsync() 方法）。每当 m 项中的一项下载完成时，必须尽快替换 list1 中的另一个剩余项，并且必须计算到所有 List1 项都继续进行。 （n 和 m 的数量由用户在运行时输入）

如何做到这一点？

Answer 1

这是您可以使用的通用方法。

当您调用此 TIn 时，它将是字符串（URL 地址），并且 asyncProcessor 将是您的异步​​方法，它将 URL 地址作为输入并返回一个任务。

此方法使用的 SlimSemaphore 将只允许实时并发异步 I/O 请求的数量，一旦一个完成，另一个请求就会执行。类似于滑动窗口模式。

public static Task ForEachAsync<TIn>(
            IEnumerable<TIn> inputEnumerable,
            Func<TIn, Task> asyncProcessor,
            int? maxDegreeOfParallelism = null)
        {
            int maxAsyncThreadCount = maxDegreeOfParallelism ?? DefaultMaxDegreeOfParallelism;
            SemaphoreSlim throttler = new SemaphoreSlim(maxAsyncThreadCount, maxAsyncThreadCount);

            IEnumerable<Task> tasks = inputEnumerable.Select(async input =>
            {
                await throttler.WaitAsync().ConfigureAwait(false);
                try
                {
                    await asyncProcessor(input).ConfigureAwait(false);
                }
                finally
                {
                    throttler.Release();
                }
            });

            return Task.WhenAll(tasks);
        }

Answer 2

您应该查看TPL Dataflow，将System.Threading.Tasks.Dataflow NuGet 包添加到您的项目中，然后您想要的就这么简单

private static HttpClient _client = new HttpClient();
public async Task<List<MyClass>> ProcessDownloads(IEnumerable<string> uris, 
                                                  int concurrentDownloads)
{
    var result = new List<MyClass>();

    var downloadData = new TransformBlock<string, string>(async uri =>
    {
        return await _client.GetStringAsync(uri); //GetStringAsync is a thread safe method.
    }, new ExecutionDataflowBlockOptions{MaxDegreeOfParallelism = concurrentDownloads});

    var processData = new TransformBlock<string, MyClass>(
          json => JsonConvert.DeserializeObject<MyClass>(json), 
          new ExecutionDataflowBlockOptions {MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded});

    var collectData = new ActionBlock<MyClass>(
          data => result.Add(data)); //When you don't specifiy options dataflow processes items one at a time.

    //Set up the chain of blocks, have it call `.Complete()` on the next block when the current block finishes processing it's last item.
    downloadData.LinkTo(processData, new DataflowLinkOptions {PropagateCompletion = true});
    processData.LinkTo(collectData, new DataflowLinkOptions {PropagateCompletion = true});

    //Load the data in to the first transform block to start off the process.
    foreach (var uri in uris)
    {
        await downloadData.SendAsync(uri).ConfigureAwait(false);
    }
    downloadData.Complete(); //Signal you are done adding data.

    //Wait for the last object to be added to the list.
    await collectData.Completion.ConfigureAwait(false);

    return result;
}

在上面的代码中，只有concurrentDownloads 数量的 HttpClient 在任何给定时间将处于活动状态，无限线程将处理接收到的字符串并将它们转换为对象，单个线程将获取这些对象并将它们添加到一个列表。

更新：这是一个简化的示例，仅执行您在问题中要求的内容

private static HttpClient _client = new HttpClient();
public void ProcessDownloads(IEnumerable<string> uris, int concurrentDownloads)
{
    var downloadData = new ActionBlock<string>(async uri =>
    {
        var response = await _client.GetAsync(uri); //GetAsync is a thread safe method.
        //do something with response here.
    }, new ExecutionDataflowBlockOptions{MaxDegreeOfParallelism = concurrentDownloads});


    foreach (var uri in uris)
    {
       downloadData.Post(uri);
    }
    downloadData.Complete();

    downloadData.Completion.Wait();
}

Answer 3

一个简单的节流解决方案是SemaphoreSlim。
编辑
稍作改动后，代码现在会在需要时创建任务

var client = new HttpClient();
SemaphoreSlim semaphore = new SemaphoreSlim(m, m); //set the max here
var tasks = new List<Task>();

foreach(var url in urls)
{
    // moving the wait here throttles the foreach loop
    await semaphore.WaitAsync();
    tasks.Add(((Func<Task>)(async () =>
    {
        //await semaphore.WaitAsync();
        var response = await client.GetAsync(url); // possibly ConfigureAwait(false) here
        // do something with response
        semaphore.Release();
    }))());
}

await Task.WhenAll(tasks);

这是另一种方法

var client = new HttpClient();
var tasks = new HashSet<Task>();

foreach(var url in urls)
{
    if(tasks.Count == m)
    {
        tasks.Remove(await Task.WhenAny(tasks));            
    }

    tasks.Add(((Func<Task>)(async () =>
    {
        var response = await client.GetAsync(url); // possibly ConfigureAwait(false) here
        // do something with response            
    }))());
}

await Task.WhenAll(tasks);

Answer 4

并行处理项目，限制同时作业的数量：

string[] strings = GetStrings();  // Items to process.
const int m = 2;  // Max simultaneous jobs.

Parallel.ForEach(strings, new ParallelOptions {MaxDegreeOfParallelism = m}, s =>
{
    DoWork(s);
});