并行Select扩展中Disposable对象提前释放的问题排查

你的问题核心是对SemaphoreSlim生命周期和LINQ延迟执行特性的理解偏差，并非单纯代码错误，是Disposable对象使用时机与执行域不匹配导致的问题。

为什么using会引发失败或死循环？ #

LINQ的Select是延迟执行的——调用Select时并不会立即执行内部逻辑，只有当你枚举返回的IEnumerable<Task<TOutput>>（比如用foreach、ToList()）时，才会逐个创建并执行任务。

而using块的作用是：当代码块执行完毕（也就是Select语句执行完成、返回枚举器的瞬间），自动调用SemaphoreSlim.Dispose()。此时你的任务还未开始获取信号量，甚至可能还没被创建，直接导致：

• 信号量被提前销毁，后续任务调用WaitAsync()时抛出异常
• 若信号量已被Dispose，等待操作会直接失效，引发任务卡住的死循环

正确的实现思路 #

你需要保证SemaphoreSlim的生命周期覆盖所有任务的执行周期，而非Select的代码块周期。具体要注意三点：

1. 不在Select内部用using包裹SemaphoreSlim，而是将其作为外部对象，在所有任务完成后手动释放（或用using包裹整个枚举+任务等待的流程）
2. 每个任务内部通过try/finally确保信号量被Release()，无论任务成功、失败还是被取消
3. 用CancellationTokenSource实现失败取消时，要在任务出错时触发Cancel()，且每个任务都监听取消信号

示例实现代码 #

public static IEnumerable<Task<TOutput>> ParallelSelect<TSource, TOutput>(
    this IEnumerable<TSource> source,
    Func<TSource, CancellationToken, Task<TOutput>> selector,
    int maxParallelism,
    CancellationToken externalToken = default)
{
    if (source == null) throw new ArgumentNullException(nameof(source));
    if (selector == null) throw new ArgumentNullException(nameof(selector));
    if (maxParallelism < 1) throw new ArgumentOutOfRangeException(nameof(maxParallelism));

    var semaphore = new SemaphoreSlim(maxParallelism, maxParallelism);
    var cts = CancellationTokenSource.CreateLinkedTokenSource(externalToken);
    bool hasFailed = false;

    foreach (var item in source)
    {
        if (hasFailed || cts.Token.IsCancellationRequested)
        {
            // 已取消或失败，停止创建新任务
            yield break;
        }

        var task = ExecuteWithSemaphore(item);
        _ = task.ContinueWith(t =>
        {
            if (t.IsFaulted || t.IsCanceled)
            {
                // 任一任务失败，触发全局取消
                Interlocked.Exchange(ref hasFailed, 1);
                cts.Cancel();
            }
        }, TaskContinuationOptions.ExecuteSynchronously);

        yield return task;
    }

    // 内部执行逻辑：封装信号量申请、任务执行、释放流程
    async Task<TOutput> ExecuteWithSemaphore(TSource item)
    {
        await semaphore.WaitAsync(cts.Token);
        try
        {
            cts.Token.ThrowIfCancellationRequested();
            return await selector(item, cts.Token);
        }
        finally
        {
            semaphore.Release();
        }
    }

    // 注意：若需自动清理资源，可返回一个清理任务让调用方等待，或让调用方手动管理信号量生命周期
    // 示例清理逻辑（需调用方主动await）：
    // yield return Task.Run(async () =>
    // {
    //     await Task.WhenAll(source.Select(ExecuteWithSemaphore));
    //     semaphore.Dispose();
    //     cts.Dispose();
    // });
}

关键注意事项 #

• 延迟执行陷阱：LINQ延迟执行意味着Select内部代码仅在枚举时触发，using块的生命周期远短于任务执行周期，绝对不能用它包裹信号量
• 信号量安全性：必须在finally块中调用Release()，避免任务出错时信号量被永久占用
• 取消逻辑原子性：用Interlocked.Exchange确保只有一个任务触发全局取消，避免重复调用Cancel()

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