在任何给定时间运行 X 个 Task<T>，同时保持 UI 响应

Question

我有一个使用 TPL 的 C# WinForms (.NET 4.5.2) 应用程序。该工具有一个同步函数，它被传递给任务工厂X 的次数（使用不同的输入参数），其中X 是用户在开始该过程之前声明的数字。任务启动并存储在List<Task>。

假设用户输入了5，我们在async 按钮点击处理程序中有这个：

for (int i = 0; i < X; i++)
{
    var progress = Progress(); // returns a new IProgress<T>
    var task = Task<int>.Factory.StartNew(() => MyFunction(progress), TaskCreationOptions.LongRunning);
    TaskList.Add(task);
}

每个 progress 实例都会更新 UI。

现在，一旦完成一项任务，我就想启动一个新任务。本质上，该进程应该无限期地运行，在任何给定时间运行X 任务，除非用户通过 UI 取消（我将为此使用取消令牌）。我尝试使用以下方法来实现这一点：

while (TaskList.Count > 0)
{
    var completed = await Task.WhenAny(TaskList.ToArray());                                  

    if (completed.Exception == null)
    {
        // report success
    }
    else
    {
        // flatten AggregateException, print out, etc
    }
    // update some labels/textboxes in the UI, and then:
    TaskList.Remove(completed);
    var task = Task<int>.Factory.StartNew(() => MyFunction(progress), TaskCreationOptions.LongRunning);
    TaskList.Add(task);
}

这会使 UI 陷入困境。有没有更好的方法来实现此功能，同时保持 UI 响应？

在 cmets 中提出了使用 TPL 数据流的建议，但由于时间限制和规格，欢迎使用替代解决方案

更新

我不确定进度报告是否有问题？这是它的样子：

private IProgress<string> Progress()
{
    return new Progress<string>(msg =>
    {
        txtMsg.AppendText(msg);
    });
}

Answer 1

现在，一旦完成一项任务，我想启动一个新任务。本质上，该进程应该无限期运行，在任何给定时间运行 X 个任务

在我看来你想要一个无限循环在你的任务：

for (int i = 0; i < X; i++)
{
  var progress = Progress(); // returns a new IProgress<T>
  var task = RunIndefinitelyAsync(progress);
  TaskList.Add(task);
}

private async Task RunIndefinitelyAsync(IProgress<T> progress)
{
  while (true)
  {
    try
    {
      await Task.Run(() => MyFunction(progress));
      // handle success
    }
    catch (Exception ex)
    {
      // handle exceptions
    }
    // update some labels/textboxes in the UI
  }
}

但是，我怀疑“使 UI 陷入困境”可能在 // handle success 和/或 // handle exceptions 代码中。如果我的怀疑是正确的，那么将尽可能多的逻辑推入Task.Run。

Answer 2

据我了解，您只需要具有定义的并行化程度的并行执行。有很多方法可以实现你想要的。我建议使用阻塞收集和并行类而不是任务。

所以当用户点击按钮时，你需要创建一个新的阻塞集合作为你的数据源：

BlockingCollection<IProgress> queue = new BlockingCollection<IProgress>();
CancellationTokenSource source = new CancellationTokenSource();

现在您需要一个可以并行执行的运行器：

Task.Factory.StartNew(() =>
    Parallel.For(0, X, i =>
    {
        foreach (IProgress p in queue.GetConsumingEnumerable(source.Token))
        {
            MyFunction(p);
        }
    }), source.Token);

或者您可以选择更正确的分区方式。所以你需要一个分区器类：

private class BlockingPartitioner<T> : Partitioner<T>
{
    private readonly BlockingCollection<T> _Collection;
    private readonly CancellationToken _Token;

    public BlockingPartitioner(BlockingCollection<T> collection, CancellationToken token)
    {
        _Collection = collection;
        _Token = token;
    }

    public override IList<IEnumerator<T>> GetPartitions(int partitionCount)
    {
        throw new NotImplementedException();
    }

    public override IEnumerable<T> GetDynamicPartitions()
    {
        return _Collection.GetConsumingEnumerable(_Token);
    }

    public override bool SupportsDynamicPartitions
    {
        get { return true; }
    }
}

跑步者看起来像这样：

ParallelOptions Options = new ParallelOptions();
Options.MaxDegreeOfParallelism = X;

Task.Factory.StartNew(
    () => Parallel.ForEach(
        new BlockingPartitioner<IProgress>(queue, source.Token),
        Options,
        p => MyFunction(p)));

因此，您现在只需要在queue 中填写必要的数据即可。您可以随时进行。

最后，当用户取消操作时，你有两种选择：

• 首先您可以使用source.Cancel 调用中断执行，
• 或者您可以通过标记收集完成 (queue.CompleteAdding) 来优雅地停止执行，在这种情况下，运行器将执行所有已排队的数据并完成。

当然，您需要额外的代码来处理异常、进度、状态等。但主要思想在这里。