在组的任务

Question

我描述了我的问题在一个简单的例子，然后描述更贴近问题的并行执行。假设我们在box1中有n项[i1，i2，i3，i4，...，in]，并且我们有一个可以处理m个项目的box2（m通常远小于n）。每个项目所需的时间是不同的。我希望在所有项目都进行之前，一直都在做m个工作项目。

更加紧密的问题是，例如，您有一个包含n个字符串（URL地址）的文件列表1，我们希望系统具有m个文件并发下载（例如，通过httpclient.getAsync（）方法） 。的m个项目之一，每当完成下载，从列表1另一剩余项目必须尽快取代，这必须countinued直到所有的列表1项进行。

（n和m的数量通过在运行时用户输入指定的）如何这可以做什么？并行

Answer 1

工艺项目，限制并发作业的数量：

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

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

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数HttpClients的将被激活，在任何给定时间，无限线程将被处理接收到的字符串和在物体转动它们，和一个单独的线程将采取茨艾伦e对象并将它们添加到列表中。

更新：这是一个简单的例子，只有做你问对问题

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

这里是你可以使用一个通用的方法。

当你调用这个锡将字符串（URL地址）和asyncProcessor将是你的异步方法，它的URL地址作为输入并返回任务。

的SlimSemaphore由该方法中使用是要尽快允许实时并发异步I/O请求，只有n个作为一个完成其他请求将执行。像滑动窗口模式。

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 4

一个简单的节流解决方案是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);