处理大量内存中的任务答案

【问题标题】：Handling very large number of Tasks in memory处理大量内存中的任务
【发布时间】：2018-03-05 10:35:15
【问题描述】：

我正在使用以下模式执行大量操作（可能数百万）

var allTasks = new List<Task>();
var throttler = new SemaphoreSlim(initialCount: 8);

foreach (var file in filesToUpload)
{
    await throttler.WaitAsync();

    allTasks.Add(
        Task.Run(async () =>
        {
            try
            {
                await UploadFileAsync(file)
            }
            finally
            {
                throttler.Release();
            }
        }));
}

await Task.WhenAll(allTasks);

但是我担心在allTasks 集合中积累大量Task 对象。从一些诊断运行来看，我似乎已经为 ~100k Task 对象建立了大约 1Gb 的内存。

是否可以对上述模式进行任何更改以逐步淘汰已完成的任务，但仍保留整体模式的节流效果？

我自己唯一能想到的就是对整个数据集进行分区/批处理，以便上面的代码只能在上面运行，例如1000 个元素。这是最合适的方法吗？

更新

因此，根据 Henk 您的建议，我实施了以下措施；

var uploadFileBlock = new ActionBlock<string>(async file =>
{
    await UploadFileAsync(file)
}, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 8 });

foreach (var file in filePaths)
{
    await uploadFileBlock.SendAsync(file);
}

uploadFileBlock.Completion.Wait();

这似乎工作正常，并且整个时间的内存配置文件都相对较低。你觉得这个实现还可以吗？

【问题讨论】：

不是一个具体的答案，但是 a) Task.WhenAll() 将所有内容保存在内存中，b) 你执行 Task.Run(async void ...) 来执行一些异步 I/O。使用不必要的线程。
consensus 是使用 TPL 数据流。另请阅读该答案下方的 cmets。
由于调用了 Add()，代码在内存中保留了这么多任务。尽管其中 100K - 8 个任务已经完成，但不是很有用。也没有 Clear() 可见。考虑使用 CountdownEvent 类计算任务，而不是 WhenAll。
@HenkHolterman 我已经用 TPL DataFlow 的代码示例更新了这个问题 - 它看起来像适当的用法吗？感谢您迄今为止的指点，非常感谢。
我不是 DataFlow 方面的专家，但它看起来还不错。我将测试它如何处理错误，例如插入无效的文件名。另外，当您使用 HttpClient 时，请参阅我在 this answer 下的评论

标签： c# semaphore

【解决方案1】：

这与another recent SO questions 非常相似。与那个问题一样，一种可能有效的方法（尽管我自己没有测试过）是：

private async Task Test()
{
  var allTasks = new List<Task>();
  foreach (var file in filesToUpload)
  {
    await WaitList(allTasks, 1000);
    allTasks.Add(UploadFileAsync(file));
  }
  await Task.WhenAll(allTasks);
}

private async Task WaitList(IList<Task> tasks, int maxSize)
{
  while (tasks.Count > maxSize)
  {
    var completed = await Task.WhenAny(tasks).ConfigureAwait(false);
    tasks.Remove(completed);
  }
}

这样的批处理不仅有助于提高内存，还有助于防止您无意中创建拒绝服务附加。

其他方法可能会利用 .Net 类（例如 BlockingCollection）利用生产者/消费者模式

【讨论】：