在 C# .NET 中管理多线程，控制每个操作的线程数

Question

认为我给出的标题有点令人困惑，但很难表达我想要问的内容。

基本上，我正在使用 .NET 用 C# 编写一个程序，该程序使用 Google 云 API 来上传数据。

我试图以一种有效的方式做到这一点，并且成功地使用了 parallel.foreach，但我需要更好的控制。我将要上传的文件收集到一个列表中，我想按文件大小对其进行排序，然后分成 3 个相同大小（以千兆字节而不是文件数计）的列表。

其中一个列表将包含总上传大小的三分之一，但包含最大的文件（以 GB 为单位）但因此文件数最少，下一个列表将与第一个列表的总 GB 相同但由更多的小​​文件组成，最后一个列表将由许多小文件组成，但总大小也应与其他子列表相同。

然后我想为上传过程分配一组线程。 （例如，我希望最大的文件列表分配 5 个线程，中间分配 3 个线程，小文件列表只有 2 个线程。）是否可以将这 3 个列表设置为并行迭代，同时控制如何分配了很多线程？

这样做的最佳方法是什么？

Answer 1

Parallel.ForEach 和 PLINQ 用于数据并行 - 使用多个内核处理大块数据。它适用于您在内存中有 1GB 数据（或非常快的 IEnumerable 源）并希望使用所有内核处理它的情况。在这种情况下，您需要将数据分成独立的块，并让一名工作人员一次处理一项，以限制同步开销。

您所描述的是大量文件的并发上传。那是纯粹的 IO，而不是数据并行。大部分时间将用于从磁盘加载数据或将其写入网络。这是Task.Run 和async/await 的工作。要同时上传多个文件，您可以使用ActionBlock 或Channel 将文件排队并异步上传。使用通道，您必须编写一些工人样板，但您可以获得更大的控制权，特别是在您想要使用相同的客户端实例进行多次调用的情况下。 ActionBlock 本质上是无状态的。

最后，您根据大小描述具有不同 DOP 的队列，当您同时拥有大文件和小文件时，这是一个非常的好主意。您可以通过使用多个 ActionBlock 实例（每个实例具有不同的 DOP）或多个 Channel 工作人员（每个具有不同的 DOP）来做到这一点。

数据流

假设您已经有一个通过路径名上传文件的方法：

//Adopted from the Google SDK example
async Task UploadFile(DriveService service,FileInfo file)
{
    var fileName=Path.GetFileName(filePath);

    using var uploadStream = file.OpenRead();
    var request insertRequest = service.Files.Insert(
        new File { Title = file.Name },
        uploadStream,
        "image/jpeg");

    await insert.UploadAsync();
}

您可以创建三个不同的 ActionBlock 实例，每个实例都有不同的 DOP：

var small=new ActionBlock<FileInfo>(
                  file=>UploadFile(service,file),
                  new ExecutionDataflowBlockOptions
                  {
                      MaxDegreeOfParallelism = 15
                  });
var medium=new ActionBlock<FileInfo>(
                  file=>UploadFile(service,file),
                  new ExecutionDataflowBlockOptions
                  {
                      MaxDegreeOfParallelism = 10
                  });

var big=new ActionBlock<FileInfo>(
                  path=>UploadFile(service,file),
                  new ExecutionDataflowBlockOptions
                  {
                      MaxDegreeOfParallelism = 2
                  });

并根据大小将不同的文件发布到不同的块：

var directory=new DirectoryInfo(...);
var files=directory.EnumerateFiles(...);
foreach(var file in files)
{
    switch (file.Length)
    {
        case int x when x < 1024:
          small.Post(file);
          break;
        case int x when x < 10240:
          medium.Post(file);
          break;
        default:
          big.Post(file);
          break;
    }    
}

或者，在 C# 8 中：

foreach(var file in files)
{

  var block = file.Length switch {
                long x when x < 1024 => small,
                long x when x < 10240=> medium,
                _                    => big
  };
  block.Post(file)
}

当迭代完成时，我们需要通过在每个块上调用 Complete() 并等待它们全部完成来告诉我们已经完成：

small.Complete();
medium.Complete();
big.Complete();

await Task.WhenAll(small.Completion, medium.Completion, big.Completion);

Answer 2

这是另一个想法。您可以有一个列表，但上传具有动态并行度的文件。如果SemaphoreSlim 类有一个WaitAsync 方法可以将CurrentCount 减少1 以外的值，这将很容易实现。然后你可以用一个大的initialCount 来初始化SemaphoreSlim，比如1000，然后用一个相对于每个文件大小的值来调用WaitAsync。让我们称这个值为 weight。信号量将保证当前上传的所有文件的总权重不会超过1000。这可能是单个重量为1000 的大文件，或10 每个重量为100 的中等文件，或者是总重量约为1000 的小、中和大文件的混合。并行度会根据列表中下一个文件的权重不断变化。

这是您必须编写的代码示例：

var semaphore = new SemaphoreSlim(1000);
var tasks = Directory.GetFiles(@"D:\FilesToUpload")
    .Select(async filePath =>
    {
        var fi = new FileInfo(filePath);
        var weight = (int)Math.Min(1000, fi.Length / 1_000_000);
        await semaphore.WaitAsync(weight); // Imaginary overload that accepts weight
        try
        {
            await cloudService.UploadFile(filePath);
        }
        finally
        {
            semaphore.Release(weight);
        }
    })
    .ToArray();
await Task.WhenAll(tasks);

下面是一个自定义的AsyncSemaphorePlus 类，它提供了缺失的重载。它基于博文Building Async Coordination Primitives, Part 5: AsyncSemaphore 中的Stephen Toub 的AsyncSemaphore 类。它略微现代化，具有 Task.CompletedTask 和 TaskCreationOptions.RunContinuationsAsynchronously 等功能，这些功能在撰写博客文章时不可用。

public class AsyncSemaphorePlus
{
    private readonly object _locker = new object();
    private readonly Queue<(TaskCompletionSource<bool>, int)> _queue
        = new Queue<(TaskCompletionSource<bool>, int)>();
    private int _currentCount;

    public int CurrentCount { get { lock (_locker) return _currentCount; } }

    public AsyncSemaphorePlus(int initialCount)
    {
        if (initialCount < 0)
            throw new ArgumentOutOfRangeException(nameof(initialCount));
        _currentCount = initialCount;
    }

    public Task WaitAsync(int count)
    {
        lock (_locker)
        {
            if (_currentCount - count >= 0)
            {
                _currentCount -= count;
                return Task.CompletedTask;
            }
            else
            {
                var tcs = new TaskCompletionSource<bool>(
                    TaskCreationOptions.RunContinuationsAsynchronously);
                _queue.Enqueue((tcs, count));
                return tcs.Task;
            }
        }
    }

    public void Release(int count)
    {
        lock (_locker)
        {
            _currentCount += count;
            while (_queue.Count > 0)
            {
                var (tcs, weight) = _queue.Peek();
                if (weight > _currentCount) break;
                (tcs, weight) = _queue.Dequeue();
                _currentCount -= weight;
                tcs.SetResult(true);
            }
        }
    }

}

---

更新：此方法适用于上传中/大量文件。它不适合非常大量的文件，因为所有上传任务都是预先创建的。如果必须上传的文件是 100,000,000 个，那么存储所有这些任务的状态所需的内存可能会超过机器的可用 RAM。对于上传这么多文件the solution proposed by Panagiotis Kanavos 可能更可取，因为可以使用bounded 数据流块轻松修改in，并通过SendAsync 而不是Post 提供它们，以便保留整个操作所需的内存受到控制。