如何更快地进行多个 API 调用？

Question

我正在从某种 Products API 请求数据，但问题是我得到了 20 x 20 的数据。所以端点看起来像这样：

https://www.someDummyAPI.com/Api/Products?offset=0&count=20

注意：我无法更改计数，它将始终为 20。

即来自该端点的数据将包含 20 条记录，从 0 到 20，之后我必须将偏移量增加 20 才能获得下 20 条记录，依此类推（总共大约 1500 条记录，所以我必须发出大约 700 条请求）。

获取所有数据后，我使用存储过程将其插入 SQL 数据库（这是不同的过程）。

所以我的问题是，如何加快获取过程，我考虑过并行运行任务，但我需要从响应中获取结果。

现在这个过程看起来像这样：

    protected async void FSL_Sync_btn_Click(object sender, EventArgs e)
    {
        int offset = 0;
        int total= 0;
        bool isFirst = true;
        DataTable resTbl = CreateDt();
        while (offset < total || offset == 0)
        {
            try
            {
                var data = await GetFSLData(offset.ToString(),"Products");

                JObject Jresult = JObject.Parse(data);

                if (isFirst)
                {
                    Int32.TryParse(Jresult.SelectToken("total").ToString(),out total);
                    isFirst = false;
                }
                // Function to chain up data in DataTable
                resTbl = WriteInDataTable(resTbl, Jresult);

                offset += 20;
            }
            catch(Exception ex)
            {
                var msg = ex.Message;
            }
        }
    }

所以我采取的流程是：

1. 从 API 获取数据（比如说前 20 条记录）。
2. 使用WriteInDataTable 函数添加两个现有的DataTable。
3. 从resTblDatatable 将数据插入 SQL 数据库（完全不同的过程，此屏幕截图中未显示）。

我还没有使用并行任务（甚至不知道它是否是一个正确的解决方案），所以非常感谢任何帮助。

Answer 1

如果您已升级到 .NET 6 平台，您可以考虑使用 Parallel.ForEachAsync 方法来并行化 GetFSLData 调用。此方法需要IEnumerable<T> 序列作为源。您可以使用 LINQ（Enumerable.Range 方法）创建此序列。为避免与DataTable 类的线程安全相关的任何问题，您可以将JObject 结果存储在中间ConcurrentQueue<JObject> 集合中，并推迟DataTable 的创建，直到所有数据都已获取并本地可用。您可能还需要存储与每个JObject 关联的offset，以便可以按原始顺序插入结果。把所有东西放在一起：

protected async void FSL_Sync_btn_Click(object sender, EventArgs e)
{
    int total = Int32.MaxValue;
    IEnumerable<int> offsets = Enumerable
        .Range(0, Int32.MaxValue)
        .Select(n => checked(n * 20))
        .TakeWhile(offset => offset < Volatile.Read(ref total));

    var options = new ParallelOptions() { MaxDegreeOfParallelism = 10 };
    var results = new ConcurrentQueue<(int Offset, JObject JResult)>();
    await Parallel.ForEachAsync(offsets, options, async (offset, ct) =>
    {
        string data = await GetFSLData(offset.ToString(), "Products");
        JObject Jresult = JObject.Parse(data);
        if (offset == 0)
        {
            Volatile.Write(ref total,
                Int32.Parse(Jresult.SelectToken("total").ToString()));
        }
        results.Enqueue((offset, Jresult));
    });

    DataTable resTbl = CreateDt();
    foreach (var (offset, Jresult) in results.OrderBy(e => e.Offset))
    {
        resTbl = WriteInDataTable(resTbl, Jresult);
    }
}

Volatile.Read/Volatile.Write 是必需的，因为total 变量可能被多个线程并行访问。

为了获得最佳性能，您可能需要根据远程服务器的能力和您的互联网连接来调整MaxDegreeOfParallelism 配置。

注意：此解决方案在内存方面效率不高，因为它要求所有数据同时以两种不同的格式存储在内存中。

Answer 2

您可以使用Task.WhenAll 并行运行您的请求。

    public async Task<IEnumerable<string>> GetDataInParallel()
    {
        var tasks = new List<Task<string>>();
        
        while(...)
        {
           var dataTask = GetFastLaneData(offset.ToString(), "Products"); // does not launch request, only add task to a list
           tasks.Add(dataTask);
           offset += 20
        }

        var datas = await Task.WhenAll(tasks); // launch all tasks

        return datas;
    }

此方法将尝试创建或使用 1000 个线程并对其进行管理，这可能会损害性能，但会比按顺序启动它们快得多。您可以考虑对它们进行批处理以实现更好的性能并一次启动 100 个任务。

Answer 3

由于代码中的高抽象级别（恕我直言，这很好，但很难在像 SO 这样的页面上发现错误），因此很难知道您真正使用和获得了什么。

所以这里只是一个关于如何将所有请求并行化到 API 以缩短获取时间并将结果写入数据库的草图。也许 API 上有一些配额，您可能必须分块运行这些东西，但这可以通过 LINQ 轻松采用。

var httpClient = new HttpClient();
var requests = Enumerable.Range(0, 1500)
    .Where(i => i % 20 == 0)
    // Create all needed requests
    .Select(offset => $"https://www.someDummyAPI.com/Api/Products?offset={offset}&count=20")
    .Select(url => new HttpRequestMessage(HttpMethod.Get, url))
    // Create tasks to call these requests
    .Select(request => httpClient.SendAsync(request));

// Run all of these requests in parallel.
var responses = await Task.WhenAll(requests);
// Create all tasks to get the content out of the requests
var allContentStreams = responses
    .Select(response => response.Content.ReadAsStringAsync());

// Retrieve all content bodies as strings
var allRawContents = await Task.WhenAll(allContentStreams);
// Serialize strings into some usable object
var allData = allRawContents
    .Select(JsonConvert.DeserializeObject<MyDataDTO>);

// Add all objects to the database context.
foreach (var data in allData)
{
    WriteIntoDatabase(data);
}

// Let context persist data into database.
SaveDatabase();

Answer 4

获取您的第一条记录并在循环之前先设置总数：

var data = await GetFSLData(offset.ToString(),"Products");

JObject Jresult = JObject.Parse(data);

Int32.TryParse(Jresult.SelectToken("total").ToString(),out total);         
                

在下一步中，您可以并行化您的任务：

DataTable resTbl = CreateDt();
var downloadTasks = new List<Task>();
while (offset < total)
{
    downloadTasks.Add(GetFSLData(offset.ToString(),"Products"));
    offset += 20;
}

然后就可以使用Task.WhenAll获取数据了

var httpResults = await Task.WhenAll(downloadTasks);
foreach (var jObjectResult in httpResults.Select(JObject.Parse)) 
{
  resTbl = WriteInDataTable(resTbl, Jresult);
}

需要注意一些事项：您将同时使用大量请求访问该 api，这可能不是一个好主意。如果遇到此问题，您可以在 TPL 数据流库中使用 TransformBlock 和 ActionBlock。您可以在此处找到更多信息：

https://docs.microsoft.com/en-us/dotnet/standard/parallel-programming/dataflow-task-parallel-library