C#如何优雅地取消进程的执行之Cancellation详解

目录

• 概述
• 与取消线程相关的类型
• 代码举例
• 操作取消vs对象取消
• 监听并响应取消请求
• 轮询监听
• 通过回调注册进行监听
• 使用WaitHandle进行侦听
• 同时监听多个令牌
• 总结

概述

从.NET Framework 4开始，.NET使用统一的模型来协作取消异步或长时间运行的同步线程。该模型基于一个称为CancellationToken的轻量级对象。这个对象在调用一个或多个取消线程时(例如通过创建新线程或任务)，是通过将token传递给每个线程来完成的（通过链式的方式依次传递）。单个线程能够依次地将token的副本传递给其他线程。

之后，在适当的某个时机，创建token的对象就可以使用token来请求线程停止。只有请求对象可以发出取消请求，每个监听器负责监听到请求并以适当和及时的方式响应取消请求。

实现协作取消模型的一般模式是：

• 1、实例化一个CancellationTokenSource对象，该对象管理cancellation并将cancellation通知发送给单独的cancellation token。
• 2、CancellationTokenSource对象的Token属性，可以返回一个Token对象，我们可以将该Token对象发送给每个监听该cancellation的进程或Task。
• 3、为每个任务或线程提供响应取消的机制。
• 4、调用 CancellationTokenSource.Cancel() 方法，来取消线程或者Task。

【tips】我们在使用cancellation的token取消线程后，应该确保调用CancellationTokenSource.Dispose()方法，以便于释放它持有的任何非托管资源。。

下图展示出了CancellationTokenSource对象里的Token属性对象，是如何传递到其他的线程里的。

合作取消模型使创建取消感知的应用程序和库变得更容易，它支持以下功能：

• 1、取消是合式的，不会强加给监听器。监听器确定如何优雅地终止以响应取消请求。
• 2、请求不同于监听。调用可取消的线程的对象，可以控制何时(如果有的话)取消被请求。
• 3、请求的对象，可以通过仅使用一个方法，即可发送取消请求到所有的token副本中。
• 4、监听器可以通过将多个Token连接成一个linked Token，来同时监听多个Token。
• 5、用户代码可以注意到并响应library code的取消请求，而library code可以注意到并响应用户代码的取消请求。
• 6、可以通过轮询、回调注册或等待等待句柄的方式，来通知监听器执行取消请求。

与取消线程相关的类型

取消框架是作为一组相关类型实现的，这些类型在下表中列出。

取消模型以几种类型集成到.net中。

最重要的是System.Threading.Tasks.Parallel，System.Threading.Tasks.Task、System.Threading.Tasks.Task<TResult> 和 System.Linq.ParallelEnumerable。

建议使用所有新的库和应用代码来实现合作市取消模式。

代码举例

在下面的示例中，请求对象创建一个CancellationTokenSource对象，然后将该对象的Token属性传递给可取消的进程。

接收请求的线程通过轮询来监视Token的IsCancellationRequested属性的值。

当该值变为true时，侦听器可以以任何合适的方式终止。在本例中，方法只是退出，这是许多情况下所需要的全部内容。

using System;
using System.Threading;

public class Example
{
    public static void Main()
    {
        // Create the token source.
        CancellationTokenSource cts = new CancellationTokenSource();

        // Pass the token to the cancelable operation.
        ThreadPool.QueueUserWorkItem(new WaitCallback(DOSomeWork), cts.Token);
        Thread.Sleep(2500);

        // Request cancellation.
        cts.Cancel();
        Console.WriteLine("Cancellation set in token source...");
        Thread.Sleep(2500);
        // Cancellation should have happened, so call Dispose.
        cts.Dispose();
    }

    // Thread 2: The listener
    static void DoSomeWork(object? obj)
    {
        if (obj is null)
            return;

        CancellationToken token = (CancellationToken)obj;

        for (int i = 0; i < 100000; i++)
        {
            if (token.IsCancellationRequested)
            {
                Console.WriteLine("In iteration {0}, cancellation has been requested...",
                                  i + 1);
                // Perform cleanup if necessary.
                //...
                // Terminate the operation.
                break;
            }
            // Simulate some work.
            Thread.SpinWait(500000);
        }
    }
}
// The example displays output like the following:
//       Cancellation set in token source...
//       In iteration 1430, cancellation has been requested...

操作取消vs对象取消

在协作取消框架中，取消指的是操作（线程中执行的操作），而不是对象。取消请求意味着在执行任何所需的清理后，操作应尽快停止。一个cancellation token应该指向一个“可取消的操作”，无论该操作如何在您的程序中实现。

在token的IsCancellationRequested属性被设置为true之后，它不能被重置为false。因此，取消令牌在被取消后不能被重用。

如果您需要对象取消机制，您可以通过调用CancellationToken来基于操作取消机制。注册方法，如下例所示。

using System;
using System.Threading;

class CancelableObject
{
    public string id;

    public CancelableObject(string id)
    {
        this.id = id;
    }

    public void Cancel()
    {
        Console.WriteLine("Object {0} Cancel callback", id);
        // Perform object cancellation here.
    }
}

public class Example1
{
    public static void Main()
    {
        CancellationTokenSource cts = new CancellationTokenSource();
        CancellationToken token = cts.Token;

        // User defined Class with its own method for cancellation
        var obj1 = new CancelableObject("1");
        var obj2 = new CancelableObject("2");
        var obj3 = new CancelableObject("3");

        // Register the object's cancel method with the token's
        // cancellation request.
        token.Register(() => obj1.Cancel());
        token.Register(() => obj2.Cancel());
        token.Register(() => obj3.Cancel());

        // Request cancellation on the token.
        cts.Cancel();
        // Call Dispose when we're done with the CancellationTokenSource.
        cts.Dispose();
    }
}
// The example displays the following output:
//       javascriptObject 3 Cancel callback
//       Object 2 Cancel callback
//       Object 1 Cancel callback

如果一个对象支持多个并发的可取消操作，则可以给每个不同的可取消操作各自传入一个不同的token。这样，一个操作可以被取消而不会影响到其他操作。

监听并响应取消请求

在用户委托中，可取消操作的实现者决定如何终止该操作以响应取消请求。在许多情况下，用户委托可以只执行任何所需的清理，然后立即返回。

但是，在更复杂的情况下，可能需要用户委托通知库代码已发生cancellation。在这种情况下，终止操作的正确方法是委托调用ThrowIfCancellationRequested方法，这将导致抛出OperationCanceledException异常。库代码可以在用户委托线程上捕获此异常，并检查异常的token，以确定该异常是否表示协作取消或其他异常情况。

在这种情况下，终止操作的正确方法是委托调用ThrowIfCancellationRequested方法，这将导致抛出OperationCanceledException。库代码可以在用户委托线程上捕获此异常，并检查异常的token，以确定该异常是否表示协作取消或其他异常情况。

轮询监听

对于循环或递归的长时间运行的计算，可以通过定期轮询CancellationToken.IsCancellationRequested的值来监听取消请求。如果它的值为true，则该方法应该尽快清理并终止。轮询的最佳频率取决于应用程序的类型。开发人员可以为任何给定的程序确定最佳轮询频率。轮询本身不会显著影响性能。

下面的程序案例展示了一种可能的轮询方式。

static void NestedLoops(Rectangle rect, CancellationToken token)
{
   for (int col = 0; col < rect.columns && !token.IsCancellationRequested; col++) {
      // Assume that we know that the inner loop is very fast.
      // Therefore, polling once per column in the outer loop condition
      // is sufficient.
      for (int row = 0; row < rect.rows; row++) {
         // Simulating work.
         Thread.SpinWait(5_000);
         Console.Write("{0},{1} ", col, row);
      }
   }

   if (token.IsCancellationRequested) {
      // Cleanup or undo here if necessary...
      Console.WriteLine("\r\nOperation canceled");
      Console.WriteLine("Press any key to exit.");

      // If using Task:
      // token.ThrowIfCancellationRequested();
   }
}

下面的程序代码是一个详细的实现：

using System;
using System.Threading;

public class ServerClass
{
   public static void StaticMethod(object obj)
   {
      CancellationToken ct = (CancellationToken)obj;
      Console.WriteLine("ServerClass.StaticMethod is running on another thread.");

      // Simulate work that can be canceled.
      while (!ct.IsCancellationRequested) {
         Thread.SpinWait(50000);
      }
      Console.WriteLine("The worker thread has been canceled. Press any key to exit.");
      Console.ReadKey(true);
   }
}

public class Simple
{
   public static void Main()
   {
      // The Simple class controls Access to the token source.
      CancellationTokenSource cts = new CancellationTokenSource();

      Console.WriteLine("Press 'C' to terminate the application...\n");
      // Allow the UI thread to capture the token source, so that it
      // can issue the cancel command.
      Thread t1 = new Thread(() => { if (Console.ReadKey(true).KeyChar.ToString().ToUpperInvariant() == "C")
                                     cts.Cancel(); } );

      // ServerClass sees only the token, not the token source.
      Thread t2 = new Thread(new ParameterizedThreadStart(ServerClass.StaticMethod));
      // Start the UI thread.

      t1.Start();

      // Start the worker thread and pass it the token.
      t2.Start(cts.Token);

      t2.Join();
      cts.Dispose();
   }
}
// The example displays the following output:
//       Press 'C' to terminate the application...
//
//     js  ServerClass.StaticMethod is running on another thread.
//       The worker thread has been canceled. Press any key to exit.

通过回调注册进行监听

以这种方式进行的某些python操作可能会阻塞，从而无法及时检查cancellation token的值。对于这些情况，您可以注册一个回调方法，以便在收到取消请求时解除对该方法的阻塞。

Register方法返回一个专门用于此目的的CancellationTokenRegistration对象。下面的示例展示了如何使用Register方法来取消异步Web请求。

using System;
using System.Net;
using System.Threading;

class Example4
{
    static void Main()
    {
        CancellationTokenSource cts = new CancellationTokenSource();

        StartWebRequest(cts.Token);

        // cancellation will cause the web
        // request to be cancelled
        cts.Cancel();
    }

    static void StartWebRequest(CancellationToken token)
    {
        WebClient wc = new WebClient();
        wc.DownloadStringCompleted += (s, e) => Console.WriteLine("Request completed.");

        // Cancellation on the token will
        // call CancelAsync on the WebClient.
        token.Register(() =>
        {
            wc.CancelAsync();
            Console.WriteLine("Request cancelled!");
        });

        Console.WriteLine("Starting request.");
        wc.DownloadStringAsync(new Uri("http://www.contoso.com"));
    }
}

CancellationTokenRegistration对象管理线程同步，并确保回调将在精确的时间点停止执行。

为了确保系统响应ZfUIQeyZbn性并避免死锁，在注册回调时必须遵循以下准则:

1、回调方法应该是快速的，因为它是同步调用的，因此对Cancel的调用在回调返回之前不会返回。

2、如果在回调运行时调用Dispose，并且持有回调等待的锁，则程序可能会死锁。Dispose返回后，您可以释放回调所需的任何资源。

3、Callbacks 不应该执行任何手动线程或在回调中使用SynchronizationContext。如果回调必须在特定线程上运行，则使用System.Threading.CancellationTokenRegistration构造函数，该构造函数使您能够指定目标syncContext是活动的SynchronizationContext.Current。在回调中执行手动线程会导致死锁。

使用WaitHandle进行侦听

当一个可取消的操作在等待一个同步原语(如System.Threading. manualresetevent或System.Threading. Semaphore)时可能会阻塞。

你可以使用CancellationToken.WaitHandle属性，以使操作同时等待事件和取消请求。

CancellationToken的 等待句柄 将在响应取消请求时发出信号，该方法可以使用WaitAny()方法的返回值来确定发出信号的是否是cancellation token。然后操作可以直接退出，或者抛出OperationCanceledException异常。

// Wait on the event if it is not signaled.
int eventThatSignaledIndex =
       WaitHandle.WaitAny(new WaitHandle[] { mre, token.WaitHandle },
                          new TimeSpan(0, 0, 20));

System.Threading.ManualResetEventSlim和System.Threading.SemaphoreSlim都在它们的Wait()方法中支持取消框架。

您可以将CancellationToken传递给该方法，当请求取消时，事件将被唤醒并抛出OperationCanceledException。

try
{
    // mres is a ManualResetEventSlim
    mres.Wait(token);
}
catch (OperationCanceledException)
{
    // Throw immediately to be responsive. The
    // alternative is to do one more item of work,
    // and throw on next iteration, because
    // IsCancellationRequested will be true.
    Console.WriteLine("The wait operation was canceled.");
    throw;
}

Console.Write("Working...");
// Simulating work.
Thread.SpinWait(500000);

下面的示例使用ManualResetEvent来演示如何解除阻塞不支持统一取消的等待句柄。

using System;
using System.Threading;
using System.Threading.Tasks;

class CancelOldStyleEvents
{
    // Old-style MRE that doesn't support unified cancellation.
    static ManualResetEvent mre = new ManualResetEvent(false);

    static void Main()
    {
        var cts = new CancellationTokenSource();

        // Pass the same token source to the delegate and to the task instance.
        Task.Run(() => DoWork(cts.Token), cts.Token);
        Console.WriteLine("Press s to start/restart, p to pause, or c to cancel.");
        Console.WriteLine("Or any other key to exit.");

        // Old-style UI thread.
        bool goAgain = true;
        while (goAgain)
        {
            char ch = Console.ReadKey(true).KeyChar;

            switch (ch)
            {
                case 'c':
                    cts.Cancel();
                    break;
                case 'p':
                    mre.Reset();
                    break;
                case 's':
                    mre.Set();
                    break;
                default:
                    goAgain = false;
                    break;
            }

            Thread.Sleep(100);
        }
        cts.Dispose();
    }

    static void DoWork(CancellationToken token)
    {
        while (true)
        {
            // Wait on the event if it is not signaled.
            int eventThatSignaledIndex =
                   WaitHandle.WaitAny(new WaitHandle[] { mre, token.WaitHandle },
                                      new TimeSpan(0, 0, 20));

            // Were we canceled while waiting?
            if (eventThatSignaledIndex == 1)
            {
                Console.WriteLine("The wait operation was canceled.");
                throw new OperationCanceledException(token);
            }
            // Were we canceled while running?
            else if (token.IsCancellationRequested)
            {
                Console.WriteLine("I was canceled while running.");
                token.ThrowIfCancellationRequested();
            }
            // Did we time out?
            else if (eventThatSignaledIndex == WaitHandle.WaitTimeout)
            {
                Console.WriteLine("I timed out.");
                break;
            }
            else
            {
                Console.Write("Working... ");
                // Simulating work.
                Thread.SpinWait(5000000);
            }
        }
    }
}

下面的示例使用ManualResetEventSlim来演示如何解除支持统一取消的协调原语的阻塞。同样的方法也可以用于其他轻量级协调原语，如SemaphoreSlim和CountdownEvent。

using System;
using System.Threading;
using System.Threading.Tasks;

class CancelNewStyleEvents
{
   // New-style MRESlim that supports unified cancellation
   // in its Wait methods.
   static ManualResetEventSlim mres = new ManualResetEventSlim(false);

   static void Main()
   {
      var cts = new CancellationTokenSource();

      // Pass the same token source to the delegate and to the task instance.
      Task.Run(() => DoWork(cts.Token), cts.Token);
      Console.WriteLine("Press c to cancel, p to pause, or s to start/restart,");
      Console.WriteLine("or any other key to exit.");

      // New-style UI thread.
         bool goAgain = true;
        编程客栈 while (goAgain)
         {
             char ch = Console.ReadKey(true).KeyChar;

             switch (ch)
             {
                 case 'c':
                     // Token can only be canceled once.
                     cts.Cancel();
                     break;
                 case 'p':
                     mres.Reset();
                     break;
                 case 's':
                     mres.Set();
                     break;
                 default:
                     goAgain = false;
                     break;
             }

             Thread.Sleep(100);
         }
         cts.Dispose();
     }

     static void DoWork(CancellationToken token)
     {

         while (true)
         {
             if (token.IsCancellationRequested)
             {
                 Console.WriteLine("Canceled while running.");
                 token.ThrowIfCancellationRequested();
             }

             // Wait on the event to be signaled
             // or the token to be canceled,
             // whichever comes first. The token
             // will throw an exception if it is canceled
             // while the thread is waiting on the event.
             try
             {
                 // mres is a ManualResetEventSlim
                 mres.Wait(token);
             }
             catch (OperationCanceledException)
             {
                 // Throw immediately to be responsive. The
                 // alternative is to do one more item of work,
                 // and throw on next iteration, because
                 // IsCancellationRequested will be true.
                 Console.WriteLine("The wait operation was canceled.");
                 throw;
             }

             Console.Write("Working...");
             // Simulating work.
             Thread.SpinWait(500000);
         }
     }
 }

同时监听多个令牌

在某些情况下，侦听器必须同时侦听多个cancellation token。

例如，一个可取消操作除了监控通过方法形参传入的外部token之外，还可能必须监视内部的cancellation token。为此，创建一个linked token源，它可以将两个或多个token连接到一个token中，如下面的示例所示。

using System;
using System.Threading;
using System.Threading.Tasks;

class LinkedTokenSourceDemo
{
    static void Main()
    {
        WorkerWithTimer worker = new WorkerWithTimer();
        CancellationTokenSource cts = new CancellationTokenSource();

        // Task for UI thread, so we can call Task.Wait wait on the main thread.
        Task.Run(() =>
        {
            Console.WriteLine("Press 'c' to cancel within 3 seconds after work begins.");
            Console.WriteLine("Or let the task time out by doing nothing.");
            if (Console.ReadKey(true).KeyChar == 'c')
                cts.Cancel();
        });

        // Let the user read the UI message.
        Thread.Sleep(1000);

        // Start the worker task.
        Task task = Task.Run(() => worker.DoWork(cts.Token), cts.Token);

        try
        {
            task.Wait(cts.Token);
        }
        catch (OperationCanceledException e)
        {
            if (e.CancellationToken == cts.Token)
                Console.WriteLine("Canceled from UI thread throwing OCE.");
        }
        catch (AggregateException ae)
        {
            Console.WriteLine("AggregateException caught: " + ae.InnerException);
            foreach (var inner in ae.InnerExceptions)
            {
                Console.WriteLine(inner.Message + inner.Source);
            }
        }

        Console.WriteLine("Press any key to exit.");
        Console.ReadKey();
        cts.Dispose();
    }
}

class WorkerWithTimer
{
    CancellationTokenSource internalTokenSource = new CancellationTokenSource();
    CancellationToken internalToken;
    CancellationToken externalToken;
    Timer timer;

    public WorkerWithTimer()
    {
        // A toy cancellation trigger that times out after 3 seconds
        // if the user does not press 'c'.
        timer = new Timer(new TimerCallback(CancelAfterTimeout), null, 3000, 3000);
    }

    public void DoWork(CancellationToken externalToken)
    {
        // Create a new token that combines the internal and external tokens.
        this.internalToken = internalTokenSource.Token;
        this.externalToken = externalToken;

        using (CancellationTokenSource linkedCts =
                CancellationTokenSource.CreateLinkedTokenSource(internalToken, externalToken))
        {
            try
            {
                DoWorkInternal(linkedCts.Token);
            }
            catch (OperationCanceledException)
            {
                if (internalToken.IsCancellationRequested)
                {
                    Console.WriteLine("Operation timed out.");
                }
                else if (externalToken.IsCancellationRequested)
                {
                    Console.WriteLine("Cancelling per user request.");
                    externalToken.ThrowIfCancellationRequested();
                }
            }
        }
    }

    private void DoWorkInternal(CancellationToken token)
    {
        for (int i = 0; i < 1000; i++)
        {
            if (token.IsCancellationRequested)
            {
                // We need to dispose the timer if cancellation
                // was requested by the external token.
                timer.Dispose();

                // Throw the exception.
                token.ThrowIfCancellationRequested();
            }

            // Simulating work.
            Thread.SpinWait(7500000);
            Console.Write("working... ");
        }
    }

    public void CancelAfterTimeout(object? state)
    {
        Console.WriteLine("\r\nTimer fired.");
        internalTokenSource.Cancel();
        timer.Dispose();
    }
}

注意，当您完成对链接的令牌源的处理后，必须对它调用Dispose。

当linked token抛出一个操作消连时,传递给异常的token就是linked token,而不是前任token。为了确定token的哪个被取消,请直接检查前任token的状态。

在本例中，AggregateException不应该被抛出，但这里会捕获它，因为在实际场景中，除了从任务委托抛出的OperationCanceledException之外，任何其他异常都被包装在AggregateException中。

总结

以上为个人经验，希望能给大家一个参考，也希望大家多多支持编程客栈(www.devze.com)。