C#实现带连续失败终止逻辑的并行循环最优方案

先回顾一下你的业务场景和核心约束：

• 单次API调用耗时约2分钟，需执行约25万次，必须并行提速
• API按请求付费，绝对不能浪费请求
• 停止规则：连续10次调用返回false则终止，否则继续递增ID

你提供的原串行代码逻辑是正确的，但效率极低；而自行尝试的批量Task代码存在一个关键问题：它没有严格遵循「连续10次false」的规则，可能会多执行不必要的API请求（比如原逻辑中累计7次false后再出现3次就会停止，但你的批量代码会把这3个和后面7个组成一批全部执行，浪费7次请求费用）。

下面是一个异步优先、严格遵循业务规则、高效利用并发且避免浪费请求的最优实现方案：

核心设计思路 #

1. 异步非阻塞：用async/await替代Task.Factory.StartNew+WaitAll，避免线程池线程被长时间阻塞（毕竟单次任务要跑2分钟，阻塞线程会严重拖慢并发效率）
2. 按序处理结果：必须按ID递增顺序处理API返回结果，才能准确累计连续false的次数，不会因为并行任务完成顺序打乱而误判
3. 可控并发度：用SemaphoreSlim限制同时运行的任务数量，既充分利用资源，又避免超出API的并发限制或触发限流
4. 及时终止：一旦连续false达到10次，立即停止启动新任务，并取消未完成的任务（如果API支持取消请求的话）

完整实现代码 #

using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;

class Program
{
    static async Task Main(string[] args)
    {
        var resultList = new ConcurrentBag<object>();
        int currentId = 0;
        int consecutiveFalseCount = 0;
        const int maxConsecutiveFalse = 10;
        const int maxConcurrentTasks = 10; // 可根据API并发限制灵活调整
        const int maxId = 250000;

        // 控制并发任务数量的信号量
        var semaphore = new SemaphoreSlim(maxConcurrentTasks);
        // 缓存已完成但未按顺序处理的任务结果
        var completedResults = new ConcurrentDictionary<int, bool>();
        // 用于终止后续任务的取消令牌
        var cancellationTokenSource = new CancellationTokenSource();

        try
        {
            while (consecutiveFalseCount < maxConsecutiveFalse && currentId < maxId)
            {
                // 等待获取并发执行的槽位
                await semaphore.WaitAsync(cancellationTokenSource.Token);

                // 启动当前ID的异步任务
                int id = currentId;
                _ = Task.Run(async () =>
                {
                    bool result = false;
                    try
                    {
                        // 调用异步API（如果API只有同步接口，这里可以用Task.Run包装，但尽量用原生异步）
                        result = await MyApiCallAsync(id, resultList, cancellationTokenSource.Token);
                    }
                    finally
                    {
                        // 缓存结果并释放信号量
                        completedResults.TryAdd(id, result);
                        semaphore.Release();
                    }
                }, cancellationTokenSource.Token);

                currentId++;

                // 处理已完成的、按顺序的结果
                while (completedResults.TryRemove(currentId - 1, out bool taskResult))
                {
                    if (taskResult)
                    {
                        // 返回true，重置连续false计数
                        consecutiveFalseCount = 0;
                    }
                    else
                    {
                        // 返回false，递增计数
                        consecutiveFalseCount++;
                        // 达到连续10次false，触发取消
                        if (consecutiveFalseCount >= maxConsecutiveFalse)
                        {
                            cancellationTokenSource.Cancel();
                            break;
                        }
                    }
                }
            }

            // 等待所有已启动的任务完成（可选，根据是否需要处理剩余结果决定）
            await semaphore.WaitAsync(maxConcurrentTasks);
            semaphore.Release(maxConcurrentTasks);
        }
        catch (OperationCanceledException)
        {
            // 任务被取消，正常退出
        }
        finally
        {
            // 清理资源
            cancellationTokenSource.Dispose();
            semaphore.Dispose();
        }
    }

    // 异步API调用方法（替换为你的实际逻辑）
    private static async Task<bool> MyApiCallAsync(int number, ConcurrentBag<object> resultList, CancellationToken token)
    {
        // 模拟2分钟的API调用耗时
        await Task.Delay(TimeSpan.FromMinutes(2), token);

        // 实际API逻辑：查询ID是否存在，返回true/false
        bool idExists = true; // 这里替换为你的判断逻辑
        if (idExists)
        {
            resultList.Add(new object());
        }
        return idExists;
    }
}

关键细节解释 #

• SemaphoreSlim：精准控制同时运行的任务数量，避免一次性启动25万个任务导致线程池崩溃或API限流
• ConcurrentDictionary：缓存已完成但未按顺序处理的结果，确保我们严格按ID递增顺序处理，准确累计连续false的次数
• CancellationTokenSource：当满足停止条件时，立即取消所有未完成的任务（如果你的API支持取消请求，一定要在MyApiCallAsync中检查令牌，避免浪费已发起的请求）
• 异步优先：尽量使用API提供的异步接口，这样线程池线程不会被长时间阻塞，能处理更多并发任务；如果只有同步接口，用Task.Run包装也比直接阻塞主线程好

对比你原有实现的优势 #

1. 严格遵循业务规则：按ID顺序处理结果，确保连续false的计数完全符合原逻辑，不会多执行不必要的API请求
2. 更高的并发效率：异步非阻塞的方式让线程池资源得到最大化利用，尤其是针对这种长时间运行的任务
3. 灵活的并发控制：可以根据API的并发限制随时调整maxConcurrentTasks，平衡提速效果和API成本
4. 最小化浪费：一旦满足停止条件，立即终止新任务的启动，并取消未完成的任务，最大限度减少无效请求

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