保持异步运行特定数量的任务

Question

我目前正在开发一个并发文件下载器。

出于这个原因，我想参数化并发任务的数量。我不想等待所有任务完成，而是要保持相同的数量正在运行。

事实上，这个关于星溢出的线程给了我一个正确的线索，但我正在努力让它异步：

Keep running a specific number of tasks

这是我的代码：

public async Task StartAsync()
    {
        var semaphore = new SemaphoreSlim(1, _concurrentTransfers);
        var queueHasMessages = true;

        while (queueHasMessages)
        {
            try {
                await Task.Run(async () =>
                  {
                      await semaphore.WaitAsync();
                      await asyncStuff();
                 });
            }
            finally {
                semaphore.Release();
            };
        }
    }

但是代码一次只执行一个。我认为等待阻止了我生成所需数量的任务，但我不知道如何在尊重信号量建立的限制的同时避免它。

如果我将所有任务添加到列表中并创建一个 whenall，信号量会抛出异常，因为它已达到最大计数。

有什么建议吗？

Answer 1

我注意到删除线解决方案将删除执行期间发生的任何异常。这很糟糕。

这是一个不会丢弃异常的解决方案：

---

Task.Run 是用于创建任务的工厂方法。您可以使用智能感知返回值检查自己。您可以将返回的任务分配到您喜欢的任何位置。

"await" 是一个操作符，它将等待它操作的任务完成。您可以将任何任务与 await 运算符一起使用。

public static async Task RunTasksConcurrently()
{
    IList<Task> tasks = new List<Task>();

    for (int i = 1; i < 4; i++)
    {
        tasks.Add(RunNextTask());
    }

    foreach (var task in tasks) {
        await task; 
    }
}

public static async Task RunNextTask()
{
    while(true) {
        await Task.Delay(500);
    }
}

通过将我们创建的任务的值添加到列表中，我们可以在稍后执行时等待它们。

---

下面的上一个答案

编辑：通过澄清，我认为我理解得更好。

您希望启动 3 个任务，而不是一次运行每个任务，并在一个任务完成后立即运行下一个任务。

我相信使用.ContinueWith(Action<Task>) 方法可以做到这一点。

看看这是否更接近您的预期解决方案。

    public void SpawnInitialTasks()
    {
        for (int i = 0; i < 3; i++)
        {
            RunNextTask();
        }
    }

    public void RunNextTask()
    {
        Task.Run(async () => await Task.Delay(500))
            .ContinueWith(t => RunNextTask());  
        // Recurse here to keep running tasks whenever we finish one.
    }

我们的想法是我们立即生成 3 个任务，然后每当一个任务完成时，我们就会生成下一个。如果需要保持任务之间的数据流动，可以使用参数：

RunNextTask(DataObject object)

Answer 2

您可以通过使用Parallel.ForEach() 以不使用await 的老式方式轻松完成此操作，它允许您指定要使用的最大并发线程数。

例如：

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace Demo
{
    class Program
    {
        public static void Main(string[] args)
        {
            IEnumerable<string> filenames = Enumerable.Range(1, 100).Select(x => x.ToString());

            Parallel.ForEach(
                filenames,
                new ParallelOptions { MaxDegreeOfParallelism = 4},
                download
            );
        }

        static void download(string filepath)
        {
            Console.WriteLine("Downloading " + filepath);
            Thread.Sleep(1000); // Simulate downloading time.
            Console.WriteLine("Downloaded " + filepath);
        }
    }
}

如果你运行它并观察输出，你会看到“文件”正在被批量“下载”。

更好的模拟是更改download()，以便处理每个“文件”需要随机的时间，如下所示：

static Random rng = new Random();

static void download(string filepath)
{
    Console.WriteLine("Downloading " + filepath);
    Thread.Sleep(500 + rng.Next(1000)); // Simulate random downloading time.
    Console.WriteLine("Downloaded " + filepath);
}

尝试一下，看看输出的不同。

---

但是，如果您想要一种更现代的方法来执行此操作，您可以查看 the Dataflow part of the TPL (Task Parallel Library) - 这与 async 方法配合得很好。

处理起来要复杂得多，但功能要强大得多。您可以使用ActionBlock 来执行此操作，但描述如何执行此操作有点超出我可以在这里给出的答案的范围。

看看this other answer on StackOverflow;它给出了一个简短的例子。

另请注意，TPL 并非内置于 .Net - 您必须从 NuGet 获取它。