C#中CancellationToken与队列结合的单下载任务取消问题

你的核心问题在于全局共享了同一个CancellationTokenSource，所有下载任务都绑定这个全局取消令牌，调用CancelDownload自然会触发所有任务的取消。另外代码里还有几处关键问题导致单个任务取消逻辑失效：

• 全局CancellationTokenSource没有实现任务级别的隔离
• Add方法存在语法错误：_downloadQueues.Enqueue(Item); 应该传入小写的item变量
• Process方法中没有等待下载任务完成，await Task.Delay(1000)仅做延迟，无法追踪下载任务状态
• 没有建立Item与对应CancellationTokenSource的关联，无法精准定位要取消的任务

修正后的代码实现 #

首先确保Item类包含唯一标识（用于定位任务）：

public class Item
{
    public int Id { get; set; } // 任务唯一标识
    // 保留你的原有属性
    public CancellationTokenSource? Cts { get; set; } // 每个任务独立的取消源
}

然后修改Service类的核心逻辑：

public partial class Service : IService
{
    private Queue<Item> _downloadQueues = new Queue<Item>();
    // 追踪正在执行的任务与对应取消源的字典
    private Dictionary<int, CancellationTokenSource> _activeTaskCts = new Dictionary<int, CancellationTokenSource>();
    private readonly object _lockObj = new object(); // 线程安全锁（Queue非线程安全）

    private async Task Initialize(string traceID)
    {
        await Process(traceID);
    }

    public Service()
    {
        string traceID = EX.GetTraceID();
        _ = Initialize(traceID);
    }

    public async Task Add(string traceID, Item item)
    {
        if (item is null)
        {
            _logger.Error(traceID, $"AddToDownloadQueueAsync:: Item is null");
            return;
        }

        try
        {
            lock (_lockObj)
            {
                // 为每个新任务创建独立的取消源
                item.Cts = new CancellationTokenSource();
                _downloadQueues.Enqueue(item);
            }
        }
        catch (Exception ex)
        {
            _logger.Error(traceID, ex, $"AddDownloadToQueue:: ");
        }
    }

    // 修改为接收任务ID，精准取消单个任务
    public async Task CancelDownload(int itemId)
    {
        lock (_lockObj)
        {
            // 取消正在执行的任务
            if (_activeTaskCts.TryGetValue(itemId, out var cts))
            {
                cts.Cancel();
                _activeTaskCts.Remove(itemId);
            }
            // 移除并取消队列中未开始的任务
            var pendingItems = _downloadQueues.Where(i => i.Id != itemId).ToList();
            var targetItem = _downloadQueues.FirstOrDefault(i => i.Id == itemId);
            targetItem?.Cts?.Cancel();
            
            _downloadQueues.Clear();
            foreach (var item in pendingItems)
            {
                _downloadQueues.Enqueue(item);
            }
        }
    }

    public async Task Process(string traceId)
    {
        while (true)
        {
            Item? currentItem = null;
            lock (_lockObj)
            {
                if (_downloadQueues.Count > 0)
                {
                    currentItem = _downloadQueues.Dequeue();
                }
            }

            if (currentItem != null)
            {
                try
                {
                    lock (_lockObj)
                    {
                        // 将当前任务的取消源加入追踪字典
                        _activeTaskCts.Add(currentItem.Id, currentItem.Cts!);
                    }
                    // 传入当前任务的取消令牌，等待下载完成
                    await client.Download(traceId, currentItem, currentItem.Cts!.Token);
                }
                catch (OperationCanceledException)
                {
                    _logger.Info(traceId, $"下载任务[{currentItem.Id}]已被取消");
                }
                catch (Exception ex)
                {
                    _logger.Error(traceId, ex, $"下载任务[{currentItem.Id}]执行失败");
                }
                finally
                {
                    lock (_lockObj)
                    {
                        _activeTaskCts.Remove(currentItem.Id);
                        currentItem.Cts?.Dispose(); // 释放取消源资源
                    }
                }
            }
            else
            {
                // 队列空时短暂等待，避免空循环占用CPU
                await Task.Delay(1000);
            }
        }
    }
}

关键修改说明 #

1. 任务级取消隔离：给每个Item分配独立的CancellationTokenSource，不再依赖全局取消源
2. 任务追踪与精准取消：通过_activeTaskCts字典追踪正在执行的任务，CancelDownload方法根据任务ID精准取消目标任务，同时处理队列中未开始的任务
3. 线程安全处理：对队列和字典操作添加锁，避免多线程环境下的并发问题
4. 资源清理：在任务完成/取消后释放CancellationTokenSource资源，避免内存泄漏
5. 修正语法错误：修复Add方法中入队参数的大小写问题

内容的提问来源于stack exchange，提问作者Groot