file-upload - 在小型 azure 实例中使用 Parallel.Foreach

Question

我有一个在小型实例上运行的 WebRole。该WebRole有一个将大量文件上传到BLOB存储的方法。根据 Azure 实例规范，小型实例只有1 个核心。那么，在上传这些 blob 时，Parallel.Foreach 会比常规 Foreach 给我带来什么好处吗？

Answer 1

专注于使用 aysnc 版本的 blob 存储 API 和/或 Stream API 会得到更好的服务，这样您就可以受 I/O 限制而不是 CPU 限制。任何有 BeginXXX API 的地方，您都应该通过用 Task.Factory.FromAsync 包装它来使用它。并从那里继续使用。在您的具体情况下，您应该利用 CloudBlob.BeginUploadFromStream 。最初如何获取流也同样重要，因此也要在这一端寻找异步 API。

在此之后，唯一可能阻碍您使用小型实例的因素是它的上限为 100Mbps，而中型实例的上限为 200Mbps。然后，当您需要更多处理时，您始终可以利用弹性因子并增加角色数量，并在事情平静下来时再次缩减规模。

下面是如何使用 FromAsync 调用 BeginUploadFromStream 的示例。现在，就协调并发处理而言，由于您现在正在启动异步任务，因此您不能指望 Parallel::ForEach 来限制最大并发量。这意味着您将在原始线程上使用常规的 foreach 并使用信号量来限制并发性。这将提供相当于 MaxDegreeOfParallelism 的功能:

// Setup a semaphore to constrain the max # of concurrent "thing"s we will process
int maxConcurrency = ... read from config ...
Semaphore maxConcurrentThingsToProcess = new Semaphore(maxConcurrency, maxConcurrency);

// Current thread will enumerate and dispatch I/O work async, this will be the only CPU resource we're holding during the async I/O
foreach(Thing thing in myThings)
{
    // Make sure we haven't reached max concurrency yet
    maxConcurrentThingsToProcess.WaitOne();

    try
    {
        Stream mySourceStream = ... get the source stream from somewhere ...;
        CloudBlob myCloudBlob = ... get the blob from somewhere ...;

        // Begin uploading the stream asynchronously
        Task uploadStreamTask = Task.Factory.FromAsync(
            myCloudBlob.BeginUploadFromStream,
            myCloudBlob.EndUploadFromStream,
            mySourceStream,
            null);

        // Setup a continuation that will fire when the upload completes (regardless of success or failure)
        uploadStreamTask.ContinueWith(uploadStreamAntecedent =>
        {
            try
            {
                // upload completed here, do any cleanup/post processing
            }
            finally
            {
                // Release the semaphore so the next thing can be processed
                maxConcurrentThingsToProcess.Release();
            }
        });
    }
    catch
    {
        // Something went wrong starting to process this "thing", release the semaphore
        maxConcurrentThingsToProcess.Release();

        throw;
    }
}

现在，在这个示例中，我没有展示您应该如何异步获取源流，但是，例如，如果您从其他地方的 URL 下载该流，您也希望异步启动它，并且将此处异步上传的开始链接到其上的延续。

相信我，我知道这不仅仅是执行简单的 Parallel::ForEach 代码，但是 Parallel::ForEach 的存在是为了使 CPU 密集型任务的并发变得容易。对于 I/O，使用异步 API 是在最小化 CPU 资源的同时实现最大 I/O 吞吐量的唯一方法。