使用 async-await 并行多线程下载

Question

我在我的 Windows 服务 - C# 中有 100 多个要从 Web 下载的大文件。要求是一次维护 - 最多 4 个并行 Web 文件下载。

我可以使用异步等待实现并发/并行下载，还是必须使用BackgroundWorker 进程或线程？ async-await 是多线程的吗？ 请参阅下面使用 async-await 的示例程序：

 static int i = 0;
 
 Timer_tick() {
   while (i < 4) {
     i++;
     model = GetNextModel();
     await Download(model);
   }
 }
 
 private async Download(XYZ model) {
   Task<FilesetResult> t = DoWork(model);
   result = await t;
   //Use Result
 }
 
 private async Task<FilesetResult> Work(XYZ model) {
   fileresult = await api.Download(model.path)
   i--;
   return filesetresult;
 }

Answer 1

您可以使用SemaphoreSlim 类限制并行运行的异步任务的数量。比如：

List<DownloadRequest> requests = Enumerable.Range(0, 100).Select(x => new DownloadRequest()).ToList();
using (var throttler = new SemaphoreSlim(4))
{
    Task<DownloadResult>[] downloadTasks = requests.Select(request => Task.Run(async () =>
    {
        await throttler.WaitAsync();
        try
        {
            return await DownloadTaskAsync(request);
        }
        finally
        {
            throttler.Release();
        }
    })).ToArray();
    await Task.WhenAll(downloadTasks);
}

更新：感谢 cmets，已修复问题。

Update2：动态请求列表的示例解决方案

public class DownloadManager : IDisposable
{
    private readonly SemaphoreSlim _throttler = new SemaphoreSlim(4);

    public async Task<DownloadResult> DownloadAsync(DownloadRequest request)
    {
        await _throttler.WaitAsync();
        try
        {
            return await api.Download(request);
        }
        finally
        {
            _throttler.Release();
        }
    }

    public void Dispose()
    {
        _throttler?.Dispose();
    }
}

Answer 2

手工操作似乎非常复杂。

var files = new List<Uri>();

Parallel.ForEach(files, 
                 new ParallelOptions { MaxDegreeOfParallelism = 4 },
                 this.Download);

现在你只需要一个单一的、正常的、同步的方法private void Download(Uri file)，你就可以开始了。

如果您需要生产者/消费者模式，最简单的版本可能是BlockingCollection：

using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;

namespace ConsoleApp11
{
    internal class Program
    {
        internal static void Main()
        {
            using (var queue = new BlockingCollection<Uri>())
            {
                // starting the producer task:
                Task.Factory.StartNew(() =>
                {
                    for (int i = 0; i < 100; i++)
                    {
                        // faking read from message queue... we get a new Uri every 100 ms
                        queue.Add(new Uri("http://www.example.com/" + i));

                        Thread.Sleep(100);
                    }

                    // just to end this program... you don't need to end this, just listen to your message queue
                    queue.CompleteAdding();
                });

                // run the consumers:
                Parallel.ForEach(queue.GetConsumingEnumerable(), new ParallelOptions { MaxDegreeOfParallelism = 4 }, Download);
            }
        }

        internal static void Download(Uri uri)
        {
            // download your file here

            Console.WriteLine($"Downloading {uri} [..        ]");
            Thread.Sleep(1000);
            Console.WriteLine($"Downloading {uri} [.....     ]");
            Thread.Sleep(1000);
            Console.WriteLine($"Downloading {uri} [.......   ]");
            Thread.Sleep(1000);
            Console.WriteLine($"Downloading {uri} [......... ]");
            Thread.Sleep(1000);
            Console.WriteLine($"Downloading {uri} [..........]");
            Thread.Sleep(1000);
            Console.WriteLine($"Downloading {uri} OK");
        }
    }
}