位置：首页-资讯-后端开发

.Net使用Cancellation Framework取消并行任务

2024-04-02 19:55

短信预约 -IT技能 免费直播动态提醒

在.net 4.0中，引入了一个新的类CancellationToken，这个类基本上集成了我们各种常用的取消方式，在并发任务中非常有用。

同步模式下的取消：

一种比较常见的需要支持取消功能的的是一些比较耗时的分段操作：如视频转换，网络下载等，这种方式下的取消机制如下：

建立一个标记位，表示该操作是否已经取消
UI线程在获取到取消事件后，置标记位为true
耗时的操作线程里，没进行一小段操作之后查询该标记位，如果为true则主动退出。

使用方式如下：

    EventHandler externalEvent;
    void Example1()
    {
        CancellationTokenSource cts = new CancellationTokenSource();
        externalEvent +=
         (sender, obj) => { cts.Cancel(); }; //wire up an external requester 
        try
        {
            int val = LongRunningFunc(cts.Token);
        }
        catch (OperationCanceledException)
        {
            //cleanup after cancellation if required... 
        }
    }

    private static int LongRunningFunc(CancellationToken token)
    {
        int total = 0;
        for (int i = 0; i < 1000; i++)
        {
            for (int j = 0; j < 1000; j++)
            {
                total++;
            }
            if (token.IsCancellationRequested)
            { // observe cancellation 
                throw new OperationCanceledException(token); // acknowledge cancellation 
            }
        }
        return total;
    }

异步模式下的取消

另外一种常见的方式是在一些异步操作中，往往不能主动释放，只能等待异步操作回调的时候才能操作结果。此时一般取消方法如下：

任务线程注册异步操作完成的回调函数，开始异步操作。
UI线程接受取消指令，置取消标记位，并主动执行回调函数
回调函数中通过取消标记位判断该任务是已经完成还是被取消的，并执行相关析构操作。

使用方式如下：

    void BlockingOperation(CancellationToken token)
    {
        ManualResetEvent mre = new ManualResetEvent(false);
        //register a callback that will set the MRE 
        CancellationTokenRegistration registration =
         token.Register(() => mre.Set());
        using (registration)
        {
            mre.WaitOne();
            if (token.IsCancellationRequested) //did cancellation wake us? 
                throw new OperationCanceledException(token);
        } //dispose the registration, which performs the deregisteration. 
    }

这里我们通过CancellationToken注册了一个回调方法以通知任务等待线程，也可以以我们经常使用的WaitHandle的那样的方式使用。

    void Wait(WaitHandle wh, CancellationToken token)
    {
        WaitHandle.WaitAny(new[] { wh, token.WaitHandle });
        if (token.IsCancellationRequested) //did cancellation wake us? 
            throw new OperationCanceledException(token);
    }

高级应用

由于例子比较简单，这里就只列举一下代码，不多介绍了。

一个CancellationToken对应多个任务

    void Example4()
    {
        CancellationTokenSource cts = new CancellationTokenSource();
        Func1(cts.Token);
        Func2(cts.Token);
        Func3(cts.Token);
        //... 
        cts.Cancel(); // all listeners see the same cancellation request. 
    }

一个任务对应多个CancellationToken

    void LinkingExample(CancellationToken ct1, CancellationToken ct2)
    {
        CancellationTokenSource linkedCTS =
        CancellationTokenSource.CreateLinkedTokenSource(ct1, ct2);
        try
        {
            SlowFunc(linkedCTS.Token);
        }
        catch (OperationCanceledException oce)
        {
            if (ct1.IsCancellationRequested)
            {
                // ... 
            }
            else if (ct2.IsCancellationRequested)
            {
                // ... 
            }
        }
        linkedCTS.Dispose(); // clean up the linking. required. 
    }

最后我们再来一个并发查询时取消的例子：

    private void RunQuery()
    {
        int[] data = { 1, 2, 3 };
        CancellationTokenSource cts = new CancellationTokenSource();
        var query = data.AsParallel()
                     .WithCancellation(cts.Token) // token given to library code 
                     .Select((x) => SlowFunc(x, cts.Token)); // token passed to user code 
    }

    private int SlowFunc(int x, CancellationToken token) 
    { 
        int result 
        while(...) 
        { 
            if (token.IsCancellationRequested) 
            throw new OperationCanceledException(token); 
            ... 
        } 
        return result; 
    }