API并发调用仅保留成功结果却返回空集合问题排查

我编写了一个可并发处理数十次API调用的方法（代码如下）。该API服务器最多支持约20个并发调用，超出此数量后部分调用会返回500错误，但此时所有任务似乎被“取消”或“忽略”，最终生成的metadatas集合为空。我尝试在Task.WaitAll外层添加try/catch块，但未捕获到任何异常（这并不意外，因为TryGetMetadataAsync方法内部已包含try/catch处理）。

我仅希望保留成功调用的结果，忽略错误调用。

请问我遗漏了什么？是否是CancellationToken存在问题？

注：api.GetMetadataWithHttpInfoAsync是由Swagger文件生成的Open API Generator客户端。

public async Task<IEnumerable<Metadata>> GetMetadataAsync(IEnumerable<string> dataRefs,
    string? userJwt, CancellationToken cancellationToken)
{
    var userId = GetUserIdFromJwt(userJwt);

    var tasks = dataRefs
        .Where(dataRef => !string.IsNullOrEmpty(dataRef))
        .Select(dataRef => TryGetMetadataAsync(dataRef, userId, cancellationToken))
        .ToArray();

    // Load data concurrently and wait for all results:
    Task.WaitAll(tasks, cancellationToken);

    var metadatas = new List<Metadata>();
    foreach (var task in tasks)
    {
        var metadata = await task;
        if (metadata != null) { metadatas.Add(metadata); }
    }

    return metadatas;
}

private async Task<Metadata?> TryGetMetadataAsync(string dataRef, string userId,
    CancellationToken cancellationToken)
{
    try
    {
        if (cancellationToken.IsCancellationRequested) { return null; }

        // Add a custom timeout to 30 seconds
        // (don't wait for potential TaskCanceledException that may be due
        // to 500 errors):
        using var timeoutCancellationTokenSource = new CancellationTokenSource(
            TimeSpan.FromSeconds(30));

        var id = GetIdFromDataRef(dataRef);
        if (id == null) { return null; }

        var metadataResponse = await api.GetMetadataWithHttpInfoAsync(id, userId,
            timeoutCancellationTokenSource.Token);

        if (metadataResponse.StatusCode == HttpStatusCode.OK
            && metadataResponse.Data != null)
        {
            // Just a simple object conversion here...
            return await ConvertToMetadataAsync(dataRef, metadataResponse.Data);
        }
        else
        {
            // Error management removed for brevity...
        }
    }
    catch (ApiException apiEx)
    {
        // Exception specific management removed for brevity...
    }
    catch (Exception ex)
    {
        // Exception management removed for brevity...
    }

    return null;
}

核心问题 #

1. Task.WaitAll的同步阻塞风险：在异步方法中使用Task.WaitAll会导致线程阻塞，若传入的cancellationToken被触发（比如某个任务异常间接引发取消），WaitAll会直接抛出OperationCanceledException，而你外层未处理该异常，导致后续收集结果的代码完全不执行，最终返回空集合。
2. 无并发数控制：API仅支持20并发，你直接发起数十个任务，超出的请求触发500错误是预期的，但大量错误可能引发客户端内部的取消逻辑，结合WaitAll的特性，会导致整个等待流程提前终止。
3. CancellationToken未合并：内部超时Token和外部传入的cancellationToken未关联，导致外部取消信号无法传递到API调用，同时内部超时也无法反馈到外部上下文。

修改方案 #

public async Task<IEnumerable<Metadata>> GetMetadataAsync(IEnumerable<string> dataRefs,
    string? userJwt, CancellationToken cancellationToken)
{
    var userId = GetUserIdFromJwt(userJwt);
    var validDataRefs = dataRefs.Where(dataRef => !string.IsNullOrEmpty(dataRef)).ToList();
    
    // 限制并发数为20，匹配API最大支持量
    using var semaphore = new SemaphoreSlim(20);
    var tasks = validDataRefs.Select(async dataRef =>
    {
        await semaphore.WaitAsync(cancellationToken);
        try
        {
            return await TryGetMetadataAsync(dataRef, userId, cancellationToken);
        }
        finally
        {
            semaphore.Release();
        }
    }).ToArray();

    // 异步等待所有任务完成，避免线程阻塞
    var results = await Task.WhenAll(tasks);
    
    // 过滤null结果，保留成功数据
    return results.Where(metadata => metadata != null)!;
}

private async Task<Metadata?> TryGetMetadataAsync(string dataRef, string userId,
    CancellationToken cancellationToken)
{
    try
    {
        // 合并外部取消Token与内部超时Token，确保双向响应
        using var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
        timeoutCts.CancelAfter(TimeSpan.FromSeconds(30));
        var combinedToken = timeoutCts.Token;

        var id = GetIdFromDataRef(dataRef);
        if (id == null) { return null; }

        var metadataResponse = await api.GetMetadataWithHttpInfoAsync(id, userId, combinedToken);

        if (metadataResponse.StatusCode == HttpStatusCode.OK
            && metadataResponse.Data != null)
        {
            return await ConvertToMetadataAsync(dataRef, metadataResponse.Data);
        }
    }
    catch (ApiException apiEx)
    {
        // 保留原有异常处理逻辑
    }
    catch (OperationCanceledException)
    {
        // 处理取消/超时，直接返回null
    }
    catch (Exception ex)
    {
        // 保留原有异常处理逻辑
    }

    return null;
}

关键修改说明 #

• 并发数控制：用SemaphoreSlim限制同时执行的API调用数量为20，从根源减少服务器返回500错误的情况。
• 异步等待替代同步阻塞：用await Task.WhenAll替代Task.WaitAll，避免线程阻塞，确保所有任务完成后再收集结果，不会因单个任务异常导致流程中断。
• Token合并：将外部取消Token与内部超时Token关联，确保外部取消信号和内部超时都能正确终止API调用，同时避免无效的任务等待。
• 简化结果收集：直接从Task.WhenAll的返回结果中过滤非null值，代码更简洁高效。

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