1、从比喻开始

场景：餐厅点餐后的等待

• 厨师做菜：长时间运行的任务

• 顾客：任务发起者

• 取消按钮：允许顾客在菜还没做好时取消订单

• 厨房通知系统：厨师能及时知道订单已取消

取消框架的本质：提供一种协作式、可传播的取消机制，让长时间运行的操作能够优雅地响应取消请求。

2、取消框架的核心本质

2.1 核心组件关系

// 取消框架的两个核心类型
class CancellationCoreComponents
{
    // 1. CancellationTokenSource - 取消信号的"发射器"
    //    创建取消令牌，并发出取消请求
    class CancellationTokenSource
    {
        public CancellationToken Token { get; }  // 产生令牌
        public void Cancel();                    // 发出取消信号
        public void CancelAfter(TimeSpan delay); // 延迟取消
    }
    
    // 2. CancellationToken - 取消信号的"接收器"  
    //    监听取消请求，并传递给需要取消的操作
    struct CancellationToken
    {
        public bool IsCancellationRequested { get; }  // 检查是否取消
        public void ThrowIfCancellationRequested();   // 如果取消则抛出异常
        public CancellationTokenRegistration Register(Action callback); // 注册取消回调
    }
}

2.2 取消令牌的状态机

class CancellationTokenStateMachine
{
    // 取消令牌的内部状态
    class TokenState
    {
        bool IsCancellationRequested;      // 是否已请求取消
        List<Action> Callbacks;           // 取消时的回调列表
        ManualResetEvent WaitHandle;      // 等待取消的信号
        int CallbackStackDepth;           // 回调执行深度（用于防止重入）
        
        public void RequestCancellation()
        {
            IsCancellationRequested = true;
            WaitHandle.Set(); // 通知所有等待者
            
            // 执行所有注册的回调
            foreach (var callback in Callbacks)
            {
                callback();
            }
        }
    }
}

3、底层原理深入解析

3.1 CancellationTokenSource 的内部实现

class CancellationTokenSourceInternals
{
    private volatile bool _isCancellationRequested;
    private List<Action> _registeredCallbacks;
    private readonly object _lockObject = new object();
    private ManualResetEvent _waitHandle;
    private Timer _timer; // 用于延迟取消
    
    public CancellationToken Token => new CancellationToken(this);
    
    public void Cancel()
    {
        List<Action> callbacksToExecute;
        
        lock (_lockObject)
        {
            if (_isCancellationRequested) return;
            
            _isCancellationRequested = true;
            _waitHandle?.Set();
            
            // 复制回调列表以便执行
            callbacksToExecute = new List<Action>(_registeredCallbacks);
            _registeredCallbacks.Clear();
        }
        
        // 在锁外执行回调，避免死锁
        foreach (var callback in callbacksToExecute)
        {
            try
            {
                callback();
            }
            catch (Exception ex)
            {
                // 记录异常，但不影响其他回调
                Trace.TraceError($"取消回调异常: {ex}");
            }
        }
    }
    
    public void CancelAfter(TimeSpan delay)
    {
        _timer = new Timer(_ => Cancel(), null, delay, Timeout.InfiniteTimeSpan);
    }
    
    internal void RegisterCallback(Action callback)
    {
        lock (_lockObject)
        {
            if (_isCancellationRequested)
            {
                // 如果已经取消，立即执行回调
                callback();
            }
            else
            {
                _registeredCallbacks.Add(callback);
            }
        }
    }
    
    internal bool IsCancellationRequested => _isCancellationRequested;
}

3.2 CancellationToken 的内部机制

struct CancellationTokenInternals
{
    private readonly CancellationTokenSource _source;
    
    public bool IsCancellationRequested => 
        _source?.IsCancellationRequested ?? false;
    
    public void ThrowIfCancellationRequested()
    {
        if (IsCancellationRequested)
        {
            // 抛出特定的取消异常
            throw new OperationCanceledException(this);
        }
    }
    
    public CancellationTokenRegistration Register(Action callback)
    {
        if (_source == null)
        {
            // 如果没有源（如 CancellationToken.None），返回空注册
            return new CancellationTokenRegistration();
        }
        
        _source.RegisterCallback(callback);
        return new CancellationTokenRegistration(_source, callback);
    }
    
    public WaitHandle WaitHandle
    {
        get
        {
            if (_source == null)
            {
                // 对于 None 令牌，返回一个永远不会触发的等待句柄
                return new ManualResetEvent(false);
            }
            return _source.GetWaitHandle();
        }
    }
}

3.3 链接取消令牌的实现

class LinkedCancellationTokenSourceInternals
{
    public static CancellationTokenSource CreateLinkedTokenSource(
        params CancellationToken[] tokens)
    {
        var linkedSource = new CancellationTokenSource();
        
        // 为每个输入令牌注册回调
        foreach (var token in tokens)
        {
            if (token.IsCancellationRequested)
            {
                // 如果任何令牌已经取消，立即取消链接源
                linkedSource.Cancel();
                break;
            }
            
            // 注册回调，当任何输入令牌取消时取消链接源
            token.Register(() => linkedSource.Cancel());
        }
        
        return linkedSource;
    }
    
    // 另一种实现：手动链接
    class ManualLinkedTokenSource : CancellationTokenSource
    {
        private List<CancellationTokenRegistration> _registrations;
        
        public ManualLinkedTokenSource(params CancellationToken[] tokens)
        {
            _registrations = new List<CancellationTokenRegistration>();
            
            foreach (var token in tokens)
            {
                if (token.IsCancellationRequested)
                {
                    Cancel();
                    break;
                }
                
                var registration = token.Register(() => Cancel());
                _registrations.Add(registration);
            }
        }
        
        protected override void Dispose(bool disposing)
        {
            if (disposing)
            {
                // 取消所有注册
                foreach (var registration in _registrations)
                {
                    registration.Dispose();
                }
                _registrations.Clear();
            }
            base.Dispose(disposing);
        }
    }
}

4、取消框架的完整使用

4.1 基础取消操作

class BasicCancellationExamples
{
    public async Task DemonstrateBasicCancellation()
    {
        Console.WriteLine("=== 基础取消操作演示 ===\n");
        
        // 1. 创建取消令牌源
        var cts = new CancellationTokenSource();
        
        // 2. 启动一个可取消的任务
        var task = Task.Run(() => DoWork(cts.Token), cts.Token);
        
        // 3. 模拟用户取消（2秒后）
        await Task.Delay(2000);
        Console.WriteLine("用户请求取消操作...");
        cts.Cancel();
        
        try
        {
            await task;
            Console.WriteLine("操作成功完成");
        }
        catch (OperationCanceledException)
        {
            Console.WriteLine("操作已被用户取消");
        }
        catch (Exception ex)
        {
            Console.WriteLine($"操作失败: {ex.Message}");
        }
    }
    
    private async Task DoWork(CancellationToken cancellationToken)
    {
        for (int i = 0; i < 100; i++)
        {
            // 方式1：检查取消请求并手动处理
            if (cancellationToken.IsCancellationRequested)
            {
                Console.WriteLine("检测到取消请求，正在清理资源...");
                // 执行清理操作
                await CleanupResourcesAsync();
                Console.WriteLine("资源清理完成");
                cancellationToken.ThrowIfCancellationRequested();
            }
            
            // 方式2：直接抛出异常（如果取消）
            cancellationToken.ThrowIfCancellationRequested();
            
            Console.WriteLine($"工作进度: {i}%");
            await Task.Delay(100, cancellationToken); // 支持取消的Delay
        }
    }
    
    private async Task CleanupResourcesAsync()
    {
        await Task.Delay(500); // 模拟资源清理
    }
    
    public void DemonstrateCancellationChecking()
    {
        Console.WriteLine("\n=== 取消检查方式演示 ===\n");
        
        var cts = new CancellationTokenSource();
        cts.CancelAfter(TimeSpan.FromSeconds(1));
        
        // 方式1：轮询检查
        Task.Run(() => {
            while (!cts.Token.IsCancellationRequested)
            {
                Console.WriteLine("轮询检查: 工作继续...");
                Thread.Sleep(200);
            }
            Console.WriteLine("轮询检查: 检测到取消");
        });
        
        // 方式2：使用 WaitHandle
        Task.Run(() => {
            while (true)
            {
                // 等待取消信号或超时
                if (cts.Token.WaitHandle.WaitOne(500))
                {
                    Console.WriteLine("WaitHandle: 检测到取消");
                    break;
                }
                Console.WriteLine("WaitHandle: 工作继续...");
            }
        });
        
        // 方式3：注册回调
        cts.Token.Register(() => {
            Console.WriteLine("回调: 取消请求收到，执行清理");
        });
        
        Thread.Sleep(3000);
    }
}

4.2 超时取消

class TimeoutCancellationExamples
{
    public async Task DemonstrateTimeoutCancellation()
    {
        Console.WriteLine("=== 超时取消演示 ===\n");
        
        // 1. 使用 CancelAfter 设置超时
        Console.WriteLine("1. 使用 CancelAfter:");
        var timeoutCts = new CancellationTokenSource();
        timeoutCts.CancelAfter(TimeSpan.FromSeconds(3));
        
        try
        {
            await LongRunningOperationAsync("操作1", timeoutCts.Token);
            Console.WriteLine("操作1完成");
        }
        catch (OperationCanceledException)
        {
            Console.WriteLine("操作1因超时被取消");
        }
        
        // 2. 使用 Task.Delay 实现超时
        Console.WriteLine("\n2. 使用 Task.Delay 超时:");
        var operationCts = new CancellationTokenSource();
        var operationTask = LongRunningOperationAsync("操作2", operationCts.Token);
        var timeoutTask = Task.Delay(TimeSpan.FromSeconds(2));
        
        var completedTask = await Task.WhenAny(operationTask, timeoutTask);
        
        if (completedTask == operationTask)
        {
            Console.WriteLine("操作2在超时前完成");
        }
        else
        {
            Console.WriteLine("操作2超时，请求取消");
            operationCts.Cancel();
            
            try
            {
                await operationTask; // 等待操作响应取消
            }
            catch (OperationCanceledException)
            {
                Console.WriteLine("操作2已取消");
            }
        }
        
        // 3. 带重试的超时机制
        Console.WriteLine("\n3. 带重试的超时机制:");
        await OperationWithRetryAndTimeout();
    }
    
    private async Task LongRunningOperationAsync(string name, CancellationToken cancellationToken)
    {
        Console.WriteLine($"{name} 开始");
        
        for (int i = 0; i < 10; i++)
        {
            cancellationToken.ThrowIfCancellationRequested();
            
            Console.WriteLine($"{name} 进度: {i * 10}%");
            await Task.Delay(1000, cancellationToken);
        }
        
        Console.WriteLine($"{name} 完成");
    }
    
    private async Task OperationWithRetryAndTimeout()
    {
        const int maxRetries = 3;
        const int timeoutSeconds = 2;
        
        for (int retry = 1; retry <= maxRetries; retry++)
        {
            var timeoutCts = new CancellationTokenSource();
            timeoutCts.CancelAfter(TimeSpan.FromSeconds(timeoutSeconds));
            
            try
            {
                Console.WriteLine($"尝试 {retry}/{maxRetries} (超时: {timeoutSeconds}秒)");
                await UnreliableOperationAsync(timeoutCts.Token);
                Console.WriteLine("操作成功");
                return;
            }
            catch (OperationCanceledException) when (timeoutCts.Token.IsCancellationRequested)
            {
                Console.WriteLine($"尝试 {retry} 超时");
                if (retry == maxRetries)
                    throw new TimeoutException("所有重试都超时");
            }
            catch (Exception ex)
            {
                Console.WriteLine($"尝试 {retry} 失败: {ex.Message}");
                if (retry == maxRetries)
                    throw;
            }
            
            await Task.Delay(1000); // 重试前等待
        }
    }
    
    private async Task UnreliableOperationAsync(CancellationToken cancellationToken)
    {
        // 模拟不可靠操作，有时快有时慢
        var random = new Random();
        int delay = random.Next(1000, 4000); // 1-4秒随机延迟
        
        await Task.Delay(delay, cancellationToken);
        
        // 模拟偶尔失败
        if (random.NextDouble() < 0.3) // 30% 失败率
            throw new InvalidOperationException("模拟操作失败");
    }
}

4.3 链接取消令牌

class LinkedTokenExamples
{
    public async Task DemonstrateLinkedTokens()
    {
        Console.WriteLine("=== 链接取消令牌演示 ===\n");
        
        // 1. 创建多个取消源
        var userCancellation = new CancellationTokenSource();
        var timeoutCancellation = new CancellationTokenSource(TimeSpan.FromSeconds(5));
        var systemCancellation = new CancellationTokenSource();
        
        // 2. 创建链接令牌
        var linkedTokenSource = CancellationTokenSource.CreateLinkedTokenSource(
            userCancellation.Token,
            timeoutCancellation.Token,
            systemCancellation.Token);
        
        try
        {
            // 3. 启动使用链接令牌的任务
            var task = ProcessDataAsync(linkedTokenSource.Token);
            
            // 4. 模拟用户取消（2秒后）
            await Task.Delay(2000);
            Console.WriteLine("用户请求取消...");
            userCancellation.Cancel();
            
            await task;
        }
        catch (OperationCanceledException ex)
        {
            // 确定取消来源
            if (userCancellation.Token.IsCancellationRequested)
            {
                Console.WriteLine("操作被用户取消");
            }
            else if (timeoutCancellation.Token.IsCancellationRequested)
            {
                Console.WriteLine("操作因超时被取消");
            }
            else if (systemCancellation.Token.IsCancellationRequested)
            {
                Console.WriteLine("操作被系统取消");
            }
            else if (linkedTokenSource.Token.IsCancellationRequested)
            {
                Console.WriteLine("操作被链接令牌取消");
            }
            
            Console.WriteLine($"取消异常: {ex.Message}");
        }
        finally
        {
            linkedTokenSource.Dispose();
        }
        
        // 5. 手动链接示例
        Console.WriteLine("\n5. 手动链接示例:");
        await ManualLinkingExample();
    }
    
    private async Task ProcessDataAsync(CancellationToken cancellationToken)
    {
        Console.WriteLine("开始处理数据...");
        
        for (int i = 0; i < 10; i++)
        {
            cancellationToken.ThrowIfCancellationRequested();
            
            Console.WriteLine($"处理进度: {i * 10}%");
            await Task.Delay(1000, cancellationToken);
        }
        
        Console.WriteLine("数据处理完成");
    }
    
    private async Task ManualLinkingExample()
    {
        var parentToken1 = new CancellationTokenSource();
        var parentToken2 = new CancellationTokenSource();
        
        // 手动创建链接
        using (var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(
            parentToken1.Token, parentToken2.Token))
        {
            var task = Task.Run(() => {
                try
                {
                    for (int i = 0; i < 5; i++)
                    {
                        linkedCts.Token.ThrowIfCancellationRequested();
                        Console.WriteLine($"工作进度: {i + 1}/5");
                        Thread.Sleep(500);
                    }
                }
                catch (OperationCanceledException)
                {
                    Console.WriteLine($"工作被取消，取消来源: " +
                        $"Token1: {parentToken1.Token.IsCancellationRequested}, " +
                        $"Token2: {parentToken2.Token.IsCancellationRequested}");
                    throw;
                }
            }, linkedCts.Token);
            
            // 第一个父令牌取消
            await Task.Delay(1500);
            parentToken1.Cancel();
            
            try
            {
                await task;
            }
            catch (OperationCanceledException)
            {
                Console.WriteLine("任务已取消");
            }
        }
    }
    
    public async Task HierarchicalCancellationExample()
    {
        Console.WriteLine("\n=== 分层取消示例 ===");
        
        // 全局取消（如应用程序关闭）
        var globalCts = new CancellationTokenSource();
        
        // 操作特定取消
        var operationCts = CancellationTokenSource.CreateLinkedTokenSource(globalCts.Token);
        operationCts.CancelAfter(TimeSpan.FromSeconds(10));
        
        // 子操作取消
        var subOperationCts = CancellationTokenSource.CreateLinkedTokenSource(operationCts.Token);
        
        try
        {
            await PerformHierarchicalOperation(subOperationCts.Token);
        }
        catch (OperationCanceledException)
        {
            if (globalCts.Token.IsCancellationRequested)
            {
                Console.WriteLine("操作因应用程序关闭而取消");
            }
            else if (operationCts.Token.IsCancellationRequested)
            {
                Console.WriteLine("操作因超时而取消");
            }
            else
            {
                Console.WriteLine("操作被子操作取消");
            }
        }
        finally
        {
            operationCts.Dispose();
            subOperationCts.Dispose();
        }
    }
    
    private async Task PerformHierarchicalOperation(CancellationToken cancellationToken)
    {
        Console.WriteLine("开始分层操作...");
        
        // 模拟一些工作
        for (int i = 0; i < 20; i++)
        {
            cancellationToken.ThrowIfCancellationRequested();
            Console.WriteLine($"分层操作进度: {i * 5}%");
            await Task.Delay(500, cancellationToken);
        }
    }
}

4.4 取消回调注册

class CancellationCallbackExamples
{
    public void DemonstrateCancellationCallbacks()
    {
        Console.WriteLine("=== 取消回调演示 ===\n");
        
        var cts = new CancellationTokenSource();
        
        // 1. 基本回调注册
        Console.WriteLine("1. 基本回调注册:");
        using (cts.Token.Register(() => Console.WriteLine("取消回调1被执行")))
        using (cts.Token.Register(() => Console.WriteLine("取消回调2被执行")))
        {
            // 启动一个任务
            var task = Task.Run(() => {
                for (int i = 0; i < 5; i++)
                {
                    if (cts.Token.IsCancellationRequested)
                    {
                        Console.WriteLine("任务检测到取消请求");
                        cts.Token.ThrowIfCancellationRequested();
                    }
                    
                    Console.WriteLine($"任务进度: {i + 1}/5");
                    Thread.Sleep(500);
                }
            }, cts.Token);
            
            // 取消操作
            Thread.Sleep(1500);
            Console.WriteLine("请求取消...");
            cts.Cancel();
            
            try
            {
                task.Wait();
            }
            catch (AggregateException ae)
            {
                if (ae.InnerExceptions[0] is OperationCanceledException)
                {
                    Console.WriteLine("任务被取消");
                }
            }
        }
        
        // 2. 回调执行顺序
        Console.WriteLine("\n2. 回调执行顺序:");
        var cts2 = new CancellationTokenSource();
        
        // 回调按注册的相反顺序执行（后进先出）
        cts2.Token.Register(() => Console.WriteLine("回调1"));
        cts2.Token.Register(() => Console.WriteLine("回调2"));
        cts2.Token.Register(() => Console.WriteLine("回调3"));
        
        cts2.Cancel();
        Console.WriteLine("取消完成");
        
        // 3. 资源清理回调
        Console.WriteLine("\n3. 资源清理回调:");
        DemonstrateResourceCleanup();
    }
    
    private void DemonstrateResourceCleanup()
    {
        var cts = new CancellationTokenSource();
        
        // 模拟需要清理的资源
        var expensiveResource = new ExpensiveResource();
        
        try
        {
            // 注册资源清理回调
            using (cts.Token.Register(() => {
                Console.WriteLine("执行资源清理...");
                expensiveResource.Dispose();
                Console.WriteLine("资源清理完成");
            }))
            {
                // 模拟工作
                for (int i = 0; i < 10; i++)
                {
                    cts.Token.ThrowIfCancellationRequested();
                    Console.WriteLine($"使用资源工作: {i + 1}/10");
                    Thread.Sleep(300);
                    
                    if (i == 3)
                    {
                        Console.WriteLine("模拟取消请求...");
                        cts.Cancel();
                    }
                }
            }
        }
        catch (OperationCanceledException)
        {
            Console.WriteLine("操作被取消，资源应该已经清理");
        }
    }
    
    public void DemonstrateCallbackManagement()
    {
        Console.WriteLine("\n=== 回调管理演示 ===");
        
        var cts = new CancellationTokenSource();
        var registrations = new List<CancellationTokenRegistration>();
        
        // 注册多个回调
        registrations.Add(cts.Token.Register(() => Console.WriteLine("回调A")));
        registrations.Add(cts.Token.Register(() => Console.WriteLine("回调B")));
        registrations.Add(cts.Token.Register(() => Console.WriteLine("回调C")));
        
        // 取消特定回调
        Console.WriteLine("取消回调B...");
        registrations[1].Dispose(); // 取消注册
        
        // 触发取消
        cts.Cancel();
        
        // 清理所有注册
        foreach (var registration in registrations)
        {
            registration.Dispose();
        }
    }
}

class ExpensiveResource : IDisposable
{
    public void Dispose()
    {
        Console.WriteLine("ExpensiveResource 被释放");
    }
}

4.5 在异步方法中使用取消

class AsyncCancellationExamples
{
    public async Task DemonstrateAsyncCancellation()
    {
        Console.WriteLine("=== 异步取消演示 ===\n");
        
        var cts = new CancellationTokenSource();
        
        // 1. 支持取消的异步方法
        try
        {
            cts.CancelAfter(TimeSpan.FromSeconds(3));
            var result = await ProcessDataAsync("示例数据", cts.Token);
            Console.WriteLine($"处理结果: {result}");
        }
        catch (OperationCanceledException)
        {
            Console.WriteLine("操作被取消");
        }
        
        // 2. 在异步流中使用取消
        Console.WriteLine("\n2. 异步流取消:");
        await DemonstrateAsyncStreamCancellation();
        
        // 3. 并行异步操作取消
        Console.WriteLine("\n3. 并行异步操作取消:");
        await DemonstrateParallelAsyncCancellation();
    }
    
    private async Task<string> ProcessDataAsync(string data, CancellationToken cancellationToken)
    {
        Console.WriteLine($"开始处理数据: {data}");
        
        // 模拟异步工作链
        var step1 = await Step1Async(data, cancellationToken);
        var step2 = await Step2Async(step1, cancellationToken);
        var step3 = await Step3Async(step2, cancellationToken);
        
        return step3;
    }
    
    private async Task<string> Step1Async(string data, CancellationToken cancellationToken)
    {
        await Task.Delay(1000, cancellationToken);
        return $"{data}_步骤1";
    }
    
    private async Task<string> Step2Async(string data, CancellationToken cancellationToken)
    {
        await Task.Delay(1000, cancellationToken);
        cancellationToken.ThrowIfCancellationRequested(); // 显式检查
        return $"{data}_步骤2";
    }
    
    private async Task<string> Step3Async(string data, CancellationToken cancellationToken)
    {
        await Task.Delay(1000, cancellationToken);
        return $"{data}_步骤3完成";
    }
    
    private async Task DemonstrateAsyncStreamCancellation()
    {
        var cts = new CancellationTokenSource();
        cts.CancelAfter(TimeSpan.FromSeconds(2));
        
        try
        {
            await foreach (var item in GenerateAsyncSequence(cts.Token))
            {
                Console.WriteLine($"收到: {item}");
            }
        }
        catch (OperationCanceledException)
        {
            Console.WriteLine("异步流被取消");
        }
    }
    
    private async IAsyncEnumerable<int> GenerateAsyncSequence(
        [EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        for (int i = 0; i < 100; i++)
        {
            await Task.Delay(200, cancellationToken);
            yield return i;
        }
    }
    
    private async Task DemonstrateParallelAsyncCancellation()
    {
        var cts = new CancellationTokenSource();
        cts.CancelAfter(TimeSpan.FromSeconds(3));
        
        var tasks = new List<Task<string>>();
        
        // 启动多个并行任务
        for (int i = 0; i < 5; i++)
        {
            tasks.Add(ProcessItemAsync(i, cts.Token));
        }
        
        try
        {
            var results = await Task.WhenAll(tasks);
            Console.WriteLine($"所有任务完成: {string.Join(", ", results)}");
        }
        catch (OperationCanceledException)
        {
            Console.WriteLine("部分或全部任务被取消");
            
            // 检查每个任务的状态
            foreach (var task in tasks)
            {
                if (task.IsCanceled)
                    Console.WriteLine($"任务被取消");
                else if (task.IsFaulted)
                    Console.WriteLine($"任务失败: {task.Exception?.Message}");
                else if (task.IsCompleted)
                    Console.WriteLine($"任务完成: {task.Result}");
            }
        }
    }
    
    private async Task<string> ProcessItemAsync(int itemId, CancellationToken cancellationToken)
    {
        var random = new Random();
        int processingTime = random.Next(1000, 5000);
        
        Console.WriteLine($"任务 {itemId} 需要 {processingTime}ms");
        
        try
        {
            await Task.Delay(processingTime, cancellationToken);
            return $"任务{itemId}_完成";
        }
        catch (OperationCanceledException)
        {
            Console.WriteLine($"任务 {itemId} 在 {processingTime}ms 时被取消");
            throw;
        }
    }
}

4.6 取消与并行操作

class ParallelCancellationExamples
{
    public void DemonstrateParallelCancellation()
    {
        Console.WriteLine("=== 并行操作取消演示 ===\n");
        
        var cts = new CancellationTokenSource();
        cts.CancelAfter(TimeSpan.FromSeconds(3));
        
        // 1. Parallel.For 取消
        Console.WriteLine("1. Parallel.For 取消:");
        try
        {
            Parallel.For(0, 100, new ParallelOptions { 
                    CancellationToken = cts.Token,
                    MaxDegreeOfParallelism = 4
                }, i => {
                cts.Token.ThrowIfCancellationRequested();
                
                Console.WriteLine($"并行处理项目 {i} 在线程 {Thread.CurrentThread.ManagedThreadId}");
                Thread.Sleep(200);
            });
        }
        catch (OperationCanceledException)
        {
            Console.WriteLine("Parallel.For 被取消");
        }
        
        // 2. PLINQ 取消
        Console.WriteLine("\n2. PLINQ 取消:");
        try
        {
            var data = Enumerable.Range(0, 1000);
            var query = data.AsParallel()
                .WithCancellation(cts.Token)
                .WithDegreeOfParallelism(4)
                .Where(x => {
                    Thread.Sleep(10); // 模拟工作
                    return x % 2 == 0;
                })
                .Select(x => x * x);
            
            var results = query.ToList();
            Console.WriteLine($"PLINQ 处理了 {results.Count} 个项目");
        }
        catch (OperationCanceledException)
        {
            Console.WriteLine("PLINQ 查询被取消");
        }
        
        // 3. 数据流取消
        Console.WriteLine("\n3. 数据流取消:");
        DemonstrateDataflowCancellation();
    }
    
    private async void DemonstrateDataflowCancellation()
    {
        var cts = new CancellationTokenSource();
        cts.CancelAfter(TimeSpan.FromSeconds(2));
        
        var bufferBlock = new BufferBlock<int>(
            new DataflowBlockOptions { CancellationToken = cts.Token });
        
        var actionBlock = new ActionBlock<int>(
            async i => {
                Console.WriteLine($"处理数据: {i}");
                await Task.Delay(500);
            },
            new ExecutionDataflowBlockOptions { 
                CancellationToken = cts.Token,
                MaxDegreeOfParallelism = 2
            });
        
        bufferBlock.LinkTo(actionBlock);
        
        // 发送数据
        for (int i = 0; i < 10; i++)
        {
            if (!await bufferBlock.SendAsync(i))
            {
                Console.WriteLine($"无法发送数据 {i} - 数据流已取消");
                break;
            }
        }
        
        bufferBlock.Complete();
        
        try
        {
            await actionBlock.Completion;
            Console.WriteLine("数据流处理完成");
        }
        catch (OperationCanceledException)
        {
            Console.WriteLine("数据流被取消");
        }
    }
    
    public void DemonstrateCooperativeCancellation()
    {
        Console.WriteLine("\n=== 协作式取消演示 ===");
        
        var cts = new CancellationTokenSource();
        cts.CancelAfter(TimeSpan.FromSeconds(4));
        
        var tasks = new List<Task>();
        
        for (int i = 0; i < 3; i++)
        {
            int taskId = i;
            tasks.Add(Task.Run(() => CooperativeWorker(taskId, cts.Token)));
        }
        
        try
        {
            Task.WaitAll(tasks.ToArray());
            Console.WriteLine("所有工作线程正常完成");
        }
        catch (AggregateException ae)
        {
            foreach (var ex in ae.InnerExceptions)
            {
                if (ex is OperationCanceledException)
                {
                    Console.WriteLine("工作线程被取消");
                }
                else
                {
                    Console.WriteLine($"工作线程异常: {ex.Message}");
                }
            }
        }
    }
    
    private void CooperativeWorker(int workerId, CancellationToken cancellationToken)
    {
        Console.WriteLine($"工作线程 {workerId} 启动");
        
        try
        {
            for (int i = 0; i < 10; i++)
            {
                // 协作式取消检查点
                if (cancellationToken.IsCancellationRequested)
                {
                    Console.WriteLine($"工作线程 {workerId} 检测到取消请求");
                    
                    // 执行清理操作
                    Cleanup(workerId);
                    
                    // 优雅退出
                    Console.WriteLine($"工作线程 {workerId} 优雅退出");
                    return;
                }
                
                // 模拟工作
                Console.WriteLine($"工作线程 {workerId} 进度: {i * 10}%");
                Thread.Sleep(1000);
                
                // 另一个检查点
                cancellationToken.ThrowIfCancellationRequested();
            }
            
            Console.WriteLine($"工作线程 {workerId} 完成");
        }
        catch (OperationCanceledException)
        {
            Console.WriteLine($"工作线程 {workerId} 被强制取消");
            Cleanup(workerId);
            throw;
        }
    }
    
    private void Cleanup(int workerId)
    {
        Console.WriteLine($"工作线程 {workerId} 清理资源...");
        Thread.Sleep(500); // 模拟清理时间
        Console.WriteLine($"工作线程 {workerId} 清理完成");
    }
}

5、取消框架的内部工作机制

5.1 取消请求的传播机制

class CancellationPropagationMechanism
{
    // 取消请求通过以下方式传播：
    public void DemonstratePropagationMethods()
    {
        var cts = new CancellationTokenSource();
        
        // 1. 轮询检查 (Polling)
        Task.Run(() => {
            while (!cts.Token.IsCancellationRequested)
            {
                // 继续工作
                Thread.Sleep(100);
            }
            // 检测到取消，执行清理
            Console.WriteLine("轮询检查: 检测到取消");
        });
        
        // 2. 回调执行 (Callbacks)
        cts.Token.Register(() => {
            Console.WriteLine("回调: 取消请求收到");
        });
        
        // 3. 等待句柄 (WaitHandle)
        Task.Run(() => {
            while (true)
            {
                // 等待取消信号或超时
                if (cts.Token.WaitHandle.WaitOne(500))
                {
                    Console.WriteLine("WaitHandle: 检测到取消");
                    break;
                }
                // 继续工作
                Console.WriteLine("WaitHandle: 工作继续...");
            }
        });
        
        // 4. 异常抛出 (Exception)
        Task.Run(() => {
            try
            {
                for (int i = 0; i < 100; i++)
                {
                    cts.Token.ThrowIfCancellationRequested();
                    // 工作
                    Thread.Sleep(100);
                }
            }
            catch (OperationCanceledException)
            {
                Console.WriteLine("异常抛出: 操作被取消");
            }
        });
        
        cts.CancelAfter(1000);
        Thread.Sleep(2000);
    }
}

5.2 取消令牌的性能优化

class CancellationTokenPerformance
{
    // CancellationToken 是结构体，轻量且线程安全
    public void DemonstrateTokenCharacteristics()
    {
        // 1. 默认令牌（永不取消）
        CancellationToken noneToken = CancellationToken.None;
        Console.WriteLine($"None Token: {noneToken.IsCancellationRequested}");
        
        // 2. 令牌比较
        var cts1 = new CancellationTokenSource();
        var cts2 = new CancellationTokenSource();
        
        CancellationToken token1 = cts1.Token;
        CancellationToken token2 = cts1.Token; // 相同源
        CancellationToken token3 = cts2.Token; // 不同源
        
        Console.WriteLine($"token1 == token2: {token1.Equals(token2)}");
        Console.WriteLine($"token1 == token3: {token1.Equals(token3)}");
        
        // 3. 性能测试
        TestCancellationPerformance();
    }
    
    private void TestCancellationPerformance()
    {
        Console.WriteLine("\n=== 取消性能测试 ===");
        
        int iterations = 1000000;
        var cts = new CancellationTokenSource();
        var token = cts.Token;
        
        // 测试 IsCancellationRequested 性能
        var stopwatch = Stopwatch.StartNew();
        for (int i = 0; i < iterations; i++)
        {
            if (token.IsCancellationRequested)
            {
                break;
            }
        }
        stopwatch.Stop();
        Console.WriteLine($"IsCancellationRequested: {stopwatch.ElapsedTicks} ticks");
        
        // 测试 ThrowIfCancellationRequested 性能（未取消时）
        stopwatch.Restart();
        for (int i = 0; i < iterations; i++)
        {
            token.ThrowIfCancellationRequested();
        }
        stopwatch.Stop();
        Console.WriteLine($"ThrowIfCancellationRequested: {stopwatch.ElapsedTicks} ticks");
        
        // 测试取消后的性能
        cts.Cancel();
        
        stopwatch.Restart();
        for (int i = 0; i < iterations; i++)
        {
            if (token.IsCancellationRequested)
            {
                break;
            }
        }
        stopwatch.Stop();
        Console.WriteLine($"已取消的 IsCancellationRequested: {stopwatch.ElapsedTicks} ticks");
    }
}

6、最佳实践和注意事项

6.1 正确模式

class CancellationBestPractices
{
    // 1. 总是提供 CancellationToken 参数
    public async Task<string> DownloadAsync(string url, CancellationToken cancellationToken = default)
    {
        using (var client = new HttpClient())
        {
            // 2. 将令牌传递给支持取消的 API
            var response = await client.GetAsync(url, cancellationToken);
            return await response.Content.ReadAsStringAsync();
        }
    }
    
    // 3. 定期检查取消请求
    public void ProcessData(CancellationToken cancellationToken)
    {
        for (int i = 0; i < 1000; i++)
        {
            // 在耗时的循环中定期检查
            if (i % 100 == 0) // 每100次迭代检查一次
            {
                cancellationToken.ThrowIfCancellationRequested();
            }
            
            // 工作...
        }
    }
    
    // 4. 合理使用回调进行资源清理
    public void UseResourceWithCancellation(CancellationToken cancellationToken)
    {
        var resource = new ExpensiveResource();
        
        // 注册取消回调来清理资源
        using (cancellationToken.Register(() => {
            Console.WriteLine("取消时清理资源");
            resource.Dispose();
        }))
        {
            resource.DoWork();
            
            // 正常工作...
            cancellationToken.ThrowIfCancellationRequested();
        }
    }
    
    // 5. 处理多个取消源
    public async Task<string> ProcessWithMultipleTokensAsync(
        CancellationToken userToken, 
        CancellationToken systemToken)
    {
        // 创建链接令牌
        using (var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(userToken, systemToken))
        {
            return await ProcessAsync(linkedCts.Token);
        }
    }
    
    // 6. 适当的取消异常处理
    public async Task<string> RobustOperationAsync(CancellationToken cancellationToken)
    {
        try
        {
            return await ProcessAsync(cancellationToken);
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            // 预期的取消
            Console.WriteLine("操作被正常取消");
            return "取消";
        }
        catch (OperationCanceledException)
        {
            // 非预期的取消
            Console.WriteLine("操作被意外取消");
            throw;
        }
    }
    
    private async Task<string> ProcessAsync(CancellationToken cancellationToken)
    {
        await Task.Delay(1000, cancellationToken);
        return "完成";
    }
}

6.2 常见陷阱

class CancellationPitfalls
{
    // 1. 忘记传递取消令牌
    public async Task<string> BadDownloadAsync(string url, CancellationToken cancellationToken)
    {
        using (var client = new HttpClient())
        {
            // 错误：没有传递 cancellationToken
            var response = await client.GetAsync(url);
            return await response.Content.ReadAsStringAsync();
        }
    }
    
    // 2. 在 finally 块中抛出异常
    public void BadCleanupExample(CancellationToken cancellationToken)
    {
        try
        {
            // 工作...
            cancellationToken.ThrowIfCancellationRequested();
        }
        finally
        {
            // 错误：在 finally 中可能抛出 OperationCanceledException
            cancellationToken.ThrowIfCancellationRequested();
        }
    }
    
    // 3. 忽略取消异常
    public async Task<string> IgnoreCancellationAsync(CancellationToken cancellationToken)
    {
        try
        {
            return await ProcessAsync(cancellationToken);
        }
        catch (OperationCanceledException)
        {
            // 错误：静默忽略取消异常
            return string.Empty;
        }
    }
    
    // 4. 不处理回调异常
    public void BadCallbackExample(CancellationToken cancellationToken)
    {
        cancellationToken.Register(() => {
            throw new InvalidOperationException("回调异常"); // 这个异常可能被忽略
        });
    }
    
    // 5. 忘记处理链接的 CancellationTokenSource
    public async Task<string> MemoryLeakExample(CancellationToken token1, CancellationToken token2)
    {
        // 错误：没有使用 using 或手动 Dispose
        var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(token1, token2);
        
        try
        {
            return await ProcessAsync(linkedCts.Token);
        }
        catch (OperationCanceledException)
        {
            // 忘记：linkedCts.Dispose();
            throw;
        }
    }
    
    // 6. 过度频繁的取消检查
    public void OverCheckingExample(CancellationToken cancellationToken)
    {
        for (int i = 0; i < 1000000; i++)
        {
            // 错误：每次迭代都检查，性能开销大
            cancellationToken.ThrowIfCancellationRequested();
            
            // 轻量级工作...
        }
    }
    
    // 正确的做法：定期检查
    public void ProperCheckingExample(CancellationToken cancellationToken)
    {
        for (int i = 0; i < 1000000; i++)
        {
            if (i % 1000 == 0) // 每1000次迭代检查一次
            {
                cancellationToken.ThrowIfCancellationRequested();
            }
            
            // 轻量级工作...
        }
    }
    
    private async Task<string> ProcessAsync(CancellationToken cancellationToken)
    {
        await Task.Delay(1000, cancellationToken);
        return "完成";
    }
}

7、总结

取消框架的本质：

• 是协作式取消的标准实现

• 基于观察者模式的信号传递机制

• 提供轻量级、线程安全的取消原语

核心优势：

1. 标准化：统一的取消模式 across .NET

2. 轻量级：CancellationToken 是结构体，性能好

3. 可组合：支持链接多个取消源

4. 灵活：支持轮询、回调、异常多种响应方式

关键技术点：

• CancellationTokenSource：取消信号的生产者

• CancellationToken：取消信号的消费者

• 链接令牌：组合多个取消源

• 取消回调：在取消时执行清理操作

• 超时取消：自动超时机制

使用要点：

• 总是在异步和长时间运行的操作中支持取消

• 定期检查取消令牌，及时响应取消请求

• 使用链接令牌组合多个取消源

• 在取消回调中进行资源清理

• 妥善处理 OperationCanceledException