停止多个任务

Question

我正在使用以下方法创建新的Tasks，并在后台进行长时间的操作。如果满足某个条件，我需要完全停止所有任务并向用户显示消息。

dowork()
{

    mylist = new List<DataModel.CheckData>();
    int index = 0;
    foreach (var line in mylist)
    {
        mylist.Add(new DataModel.CheckData() { RawLine = line, data = line,FileName=virtualfilelist[index].ToString() });
        index++;
    }

    BlockingCollection<DataModel.CheckData> ujobs = new BlockingCollection<DataModel.CheckData>();
    timerRefreshUi.Start();

    Task.Factory.StartNew(() =>
    {
        _dtRows.Clear();
        uiQueue.Clear();
        uiQueueBad.Clear();
        uiQueueGood.Clear();

        for (int i = 0; i < mylist.Count; i++)
        {
             AddResultRow(mylist[i].data, "Waiting...",mylist[i].FileName, Color.White);
             ujobs.TryAdd(new DataModel.CheckData() { RowId = i, data = mylist[i].data }, 1000);

         }
         List<Task> openCheckTasks = new List<Task>();

         while (ujobs.Count > 0)
         {
             while (openCheckTasks.Where(task => task.Status == TaskStatus.Running).ToList().Count >= threadcount)
              System.Threading.Thread.Sleep(250);

              Task t = new Task(new Action(() =>
              {

              }));

              openCheckTasks.Add(t);
              t.Start();
         }
         Task.WaitAll(openCheckTasks.ToArray());

    }).ContinueWith(task => {
        _benchmark.Stop();
        this.BeginInvoke(new Action(() =>
        {

        }));

    });

}

我曾尝试在 while 循环中使用 Cancellation Token 和 break。但它无法正常工作。请建议停止所有线程的最佳方法。我对多线程编程的经验很少。

Answer 1

CancellationToken 是正确的选择。

您是否试图控制一次运行多少个任务？这就是 TPL 已经在做的事情，而且做得很好。

请参阅此示例，启动许多 CPU 密集型任务，然后在三秒后取消所有任务：

public static void Main()
{
    var delay = TimeSpan.FromSeconds(1);
    var cts = new CancellationTokenSource();
    var tasks = Enumerable.Range(0, 100).Select(i => Task.Run(() => SlowSqrt/*Async*/(i, delay, cts.Token), cts.Token)).ToArray();
    Thread.Sleep(3000);
    cts.Cancel();
}

public static double SlowSqrt(double arg, TimeSpan delay, CancellationToken token)
{
    Console.WriteLine($"Calculating Sqrt({arg})...");
    var burnCpuTimeUntil = DateTime.Now + delay;
    while (DateTime.Now < burnCpuTimeUntil) token.ThrowIfCancellationRequested();
    var result = Math.Sqrt(arg);
    Console.WriteLine($"Sqrt({arg}) is {result}.");
    return result;
}

public static async Task<double> SlowSqrtAsync(double arg, TimeSpan delay, CancellationToken token)
{
    Console.WriteLine($"Calculating Sqrt({arg})...");
    await Task.Delay(delay, token);
    var result = Math.Sqrt(arg);
    Console.WriteLine($"Sqrt({arg}) is {result}.");
    return result;
}

其输出为：

Calculating Sqrt(1)...
Calculating Sqrt(2)...
Calculating Sqrt(0)...
Calculating Sqrt(3)...
Sqrt(2) is 1.4142135623731.
Calculating Sqrt(4)...
Sqrt(0) is 0.
Calculating Sqrt(5)...
Sqrt(3) is 1.73205080756888.
Calculating Sqrt(6)...
Sqrt(1) is 1.
Calculating Sqrt(7)...
Sqrt(5) is 2.23606797749979.
Calculating Sqrt(8)...
Sqrt(4) is 2.
Calculating Sqrt(9)...
Sqrt(6) is 2.44948974278318.
Calculating Sqrt(10)...
Sqrt(7) is 2.64575131106459.
Calculating Sqrt(11)...

由于我的机器上有四个内核，因此一次只有 4 个任务处于活动状态。取消令牌（12..99）时尚未开始的任务甚至永远不会启动。 token.ThrowIfCancellationRequested() 中已启动的任务 (8..11) 出错。它们都以TaskStatus.Canceled 状态结束。

现在，如果您将上面的代码更改为调用SlowSqrtAsync，则 1 秒延迟不会使用 CPU，因此 TPL 会激活所有 100 个任务，试图最大化 CPU 使用率。大约一秒钟后，您将获得全部 100 个结果。如果您在 Task.Delay 内取消任务，它会像 Token.ThrowIfCancellationRequested() 一样抛出 OperationCanceledException。

Calculating Sqrt(0)...
Calculating Sqrt(1)...
:
:
Calculating Sqrt(92)...
Calculating Sqrt(89)...
(about 1 second later:)
Sqrt(19) is 4.35889894354067.
Sqrt(5) is 2.23606797749979.
:
:
Sqrt(99) is 9.9498743710662.
Sqrt(92) is 9.59166304662544.

Answer 2

要停止多个正在进行的任务，您可以使用await Task.WhenAll(...)，传入一组任务。每个任务都获得对CancellationToken (System.Threading) 的引用。取决于每个任务（因为他们正在做自己的工作）不时检查令牌以查看是否已请求取消 - 如果是这样，并且可以取消，则抛出 OperationCanceledException - 等待线程将捕获因此。

此外，但不是必须取消，如果您想将每个任务的进度报告回等待线程，您可以传入 System.IProgress。

使用CancellationToken 和IProgress（.NET 4.5.2 控制台应用程序）查看下面的工作演示。

编码愉快。

private static CancellationTokenSource _tokenSource;
private static Progress<string> _progress = new Progress<string>((msg) => Console.WriteLine(msg));
private static IProgress<string> _progressReporter = _progress as IProgress<string>;

static void Main(string[] args)
{
  Task.Run(async () => await Run()).GetAwaiter().GetResult();
}

private static async Task Run()
{
  try
  {
    Console.CancelKeyPress += Console_CancelKeyPress;

    using (_tokenSource = new CancellationTokenSource())
    {
      _progressReporter.Report("Running 3 async tasks... press [Ctrl+C] to cancel.\r\n");

      await Task.WhenAll
      (
          DoWorkAsync(1, 6, _tokenSource.Token, _progress),
          DoWorkAsync(2, 3, _tokenSource.Token, _progress),
          DoWorkAsync(3, 4, _tokenSource.Token, _progress)
      );

      _progressReporter.Report($"\r\nRun complete: All tasks finished successfully (without user cancelling)");
    }
  }
  catch (OperationCanceledException)
  {
    _progressReporter.Report($"\r\nRun complete: User cancelled one or more ongoing tasks");
  }
  finally
  {
    Console.ReadLine(); // pause
  }
}

private static void Console_CancelKeyPress(object sender, ConsoleCancelEventArgs e)
{
  _progressReporter.Report("\r\nCancelling tasks...\r\n");
  _tokenSource.Cancel();
  e.Cancel = true;
}

private static async Task DoWorkAsync(int taskNumber, int durationInSeconds, CancellationToken ct, IProgress<string> progress)
{
  await Task.Run(() => 
  {
    if (ct.IsCancellationRequested)
    {
      progress.Report($"Task {taskNumber} cancelled before it started");
      return;
    }
    else
      progress.Report($"Started task {taskNumber} (duration: {durationInSeconds} sec. steps)");

    for (int i = 0; i < durationInSeconds; i++)
    {
      progress.Report($"Task {taskNumber} is working (on step {i + 1}/{durationInSeconds})");

      Thread.Sleep(1000); // simulate work in 1 sec steps

      if (ct.IsCancellationRequested)
      {
        progress.Report($"Task {taskNumber} cancelled (upon request) after it started (on step {i + 1}/{durationInSeconds})");
        ct.ThrowIfCancellationRequested();
      }
    }

    progress.Report($"Finished task {taskNumber}");
  });
}