C#中使用分区器并行查询分页REST API的方案优化咨询

问题背景 #

当前使用分区器并行查询带分页限制的REST API：API单次最多返回1000条结果，为提升查询速度，用分区器创建10个范围（总查询上限10000条），每个范围并行调用API，通过Task.WhenAll等待所有任务完成后扁平化结果。现有实现代码如下：

public async Task<string[]> QueryApiAsParallel() {
    int maximum = 10000; // I don't want to query more than 10000 results,
                         // even I know that are a lot more results
    int rangeSize = 1000; // maximum number that can be received via API

    Task<string[]>[] tasks = Partitioner.Create(0, maximum, rangeSize).AsParallel()
        .Select(async (range, index) => {
        int skip = range.Item1;
        int first = range.Item2 - range.Item1;

        string[] names = await apiClient.GetNames(skip, first);

        return names;
    }).ToArray();

    string[][] tasksCompleted = await Task.WhenAll(tasks);

    string[] flattened = tasksCompleted.SelectMany(x => x).ToArray();

    return flattened;
}

请问该方案是否合理？现有实现是否存在低效问题？有没有更优解决方案？

现有方案的合理性与潜在问题 #

合理性 #

• 核心思路可行：通过并行请求拆分后的分页范围，确实能比串行请求大幅提升总查询速度，尤其在API响应延迟较高的场景下效果明显。
• 结果上限控制到位：通过maximum限制最多获取10000条数据，符合需求。

潜在低效/风险点 #

1. AsParallel()冗余
   用Partitioner.Create生成范围后再套AsParallel()完全没必要——我们最终是要生成异步任务数组，直接遍历分区器的范围创建Task<string[]>即可，AsParallel()会额外引入并行遍历的开销，反而拖慢任务创建过程。

2. 无并发控制风险
   直接发起10个并行请求，若API服务端对单IP并发请求数有限制，可能触发限流、429错误，甚至被临时封禁。

3. 结果顺序可能混乱
   当前实现扁平化后的结果顺序是任务完成的顺序，而非原数据的逻辑顺序（比如第一个范围的结果可能最后返回，导致最终数组顺序不符合预期）。

更优解决方案 #

优化点1：移除冗余的AsParallel() #

直接遍历分区器生成的范围创建异步任务，避免不必要的并行遍历开销：

public async Task<string[]> QueryApiAsParallel() {
    int maximum = 10000;
    int rangeSize = 1000;

    var ranges = Partitioner.Create(0, maximum, rangeSize);
    var tasks = new List<Task<string[]>>();
    
    foreach (var range in ranges)
    {
        int skip = range.Item1;
        int first = range.Item2 - range.Item1;
        tasks.Add(apiClient.GetNames(skip, first));
    }

    var tasksCompleted = await Task.WhenAll(tasks);
    return tasksCompleted.SelectMany(x => x).ToArray();
}

优化点2：添加并发控制 #

用SemaphoreSlim限制同时发起的请求数，避免触发服务端限流：

public async Task<string[]> QueryApiAsParallel() {
    int maximum = 10000;
    int rangeSize = 1000;
    int maxConcurrentRequests = 5; // 根据API服务端限制调整
    var semaphore = new SemaphoreSlim(maxConcurrentRequests);

    var ranges = Partitioner.Create(0, maximum, rangeSize);
    var tasks = new List<Task<string[]>>();
    
    foreach (var range in ranges)
    {
        tasks.Add(ExecuteWithSemaphore(range));
    }

    async Task<string[]> ExecuteWithSemaphore(Tuple<int, int> range)
    {
        await semaphore.WaitAsync();
        try
        {
            int skip = range.Item1;
            int first = range.Item2 - range.Item1;
            return await apiClient.GetNames(skip, first);
        }
        finally
        {
            semaphore.Release();
        }
    }

    var tasksCompleted = await Task.WhenAll(tasks);
    return tasksCompleted.SelectMany(x => x).ToArray();
}

优化点3：保证结果顺序 #

如果需要最终结果和原数据的逻辑顺序一致（按skip从小到大排列），可以给每个任务绑定对应的范围索引，完成后按索引排序再扁平化：

public async Task<string[]> QueryApiAsParallel() {
    int maximum = 10000;
    int rangeSize = 1000;
    int maxConcurrentRequests = 5;
    var semaphore = new SemaphoreSlim(maxConcurrentRequests);

    var ranges = Partitioner.Create(0, maximum, rangeSize).ToList();
    var tasks = new List<Task<(int Index, string[] Data)>>();
    
    for (int i = 0; i < ranges.Count; i++)
    {
        int index = i;
        var range = ranges[i];
        tasks.Add(ExecuteWithSemaphore(index, range));
    }

    async Task<(int Index, string[] Data)> ExecuteWithSemaphore(int index, Tuple<int, int> range)
    {
        await semaphore.WaitAsync();
        try
        {
            int skip = range.Item1;
            int first = range.Item2 - range.Item1;
            return (index, await apiClient.GetNames(skip, first));
        }
        finally
        {
            semaphore.Release();
        }
    }

    var tasksCompleted = await Task.WhenAll(tasks);
    return tasksCompleted.OrderBy(t => t.Index)
                         .SelectMany(t => t.Data)
                         .ToArray();
}

额外建议 #

• 增加异常处理：给每个API请求添加try-catch，避免单个请求失败导致整个任务全部失败（可根据需求选择忽略失败请求或重试）。
• 添加重试机制：针对API的临时错误（如5xx、429），添加重试逻辑提升稳定性。

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