HttpClient要求を制限する簡単な方法

Question

私は、System.Net.HttpのHTTPClientを使用して、APIに対して要求を出しています。 APIは1秒あたり10リクエストに制限されています。

私のコードは大体そうのようなものです：

List<Task> tasks = new List<Task>(); 
    items..Select(i => tasks.Add(ProcessItem(i)); 

    try 
    { 
     await Task.WhenAll(taskList.ToArray()); 
    } 
    catch (Exception ex) 
    { 
    } 

ProcessItem方法は、いくつかのことを行いますが、常に次を使用してAPIを呼び出します。 await SendRequestAsync(..blah)。どのようになっています。もともとのコードがうまく働いたが、私はTask.WhenAllを使用し始めたとき、私はAPIからのメッセージ「をレート制限を超え」入門

private async Task<Response> SendRequestAsync(HttpRequestMessage request, CancellationToken token) 
{  
    token.ThrowIfCancellationRequested(); 
    var response = await HttpClient 
     .SendAsync(request: request, cancellationToken: token).ConfigureAwait(continueOnCapturedContext: false); 

    token.ThrowIfCancellationRequested(); 
    return await Response.BuildResponse(response); 
} 

。どのようにして要求が行われる割合を制限できますか？

ProcessItemは、アイテムに応じて1〜4つのAPI呼び出しを行うことができます。

Answer 1

答えはthis oneに似ています。

代わりのタスクとWhenAllのリストを使用して、Parallel.ForEachを使用して10に同時タスクの数を制限し、それぞれが少なくとも1秒を要することを確認するためにParallelOptionsを使用します。

Parallel.ForEach(
    items, 
    new ParallelOptions { MaxDegreeOfParallelism = 10 }, 
    async item => { 
     ProcessItems(item); 
     await Task.Delay(1000); 
    } 
); 

それともあれば

Parallel.ForEach(
    searches, 
    new ParallelOptions { MaxDegreeOfParallelism = 10 }, 
    async item => { 
     var watch = new Stopwatch(); 
     watch.Start(); 
     ProcessItems(item); 
     watch.Stop(); 
     if (watch.ElapsedMilliseconds < 1000) await Task.Delay((int)(1000 - watch.ElapsedMilliseconds)); 
    } 
); 

または：：

Parallel.ForEach(
    searches, 
    new ParallelOptions { MaxDegreeOfParallelism = 10 }, 
    async item => { 
     await Task.WhenAll(
       Task.Delay(1000), 
       Task.Run(() => { ProcessItems(item); }) 
      ); 
    } 
); 

必ず各項目は、できるだけ1に近い秒かかるようにしたいです

Answer 2

UPDATED ANSWER

マイProcessItems方法は、項目によって1-4 APIの呼び出しを行います。したがって、バッチサイズが10の場合、レート制限を超えています。

SendRequestAsyncでローリングウィンドウを実装する必要があります。各要求のタイムスタンプを含む待ち行列は、適切なデータ構造である。タイムスタンプが10秒より古いエントリは、デキューします。そうであるように、SOに関する同様の質問に対する答えとしてan implementationがあります。

ORIGINAL ANSWER

まだ他の人に有用である可能性が

これを処理する1つの簡単な方法は、10のグループで、あなたの要求は、同時にそれらを実行し、合計まで待つバッチにあります10秒間が経過していない場合は、これにより、要求のバッチが10秒で完了できる場合にはレート制限が適用されますが、要求のバッチが長くかかる場合は最適ではありません。 .Batch（）拡張メソッドをMoreLinqに見てください。コードは、APIは、毎秒10のリクエストに制限されている約

foreach (var taskList in tasks.Batch(10)) 
{ 
    Stopwatch sw = Stopwatch.StartNew(); // From System.Diagnostics 
    await Task.WhenAll(taskList.ToArray()); 
    if (sw.Elapsed.TotalSeconds < 10.0) 
    { 
     // Calculate how long you still have to wait and sleep that long 
     // You might want to wait 10.5 or 11 seconds just in case the rate 
     // limiting on the other side isn't perfectly implemented 
    } 
} 

Answer 3

ようになります。

それからちょうど、10のリクエストのバッチを実行するコードを持って、彼らは1秒以上を取る確保：

Items[] items = ...; 

int index = 0; 
while (index < items.Length) 
{ 
    var timer = Task.Delay(TimeSpan.FromSeconds(1.2)); // ".2" to make sure 
    var tasks = items.Skip(index).Take(10).Select(i => ProcessItemsAsync(i)); 
    var tasksAndTimer = tasks.Concat(new[] { timer }); 
    await Task.WhenAll(tasksAndTimer); 
    index += 10; 
} 

更新

マイProcessItems方法は、1-4 APIを作ります項目に応じてコール。

この場合、バッチ処理は適切な解決策ではありません。非同期メソッドを特定のの番号に限定する必要があります。これはSemaphoreSlimを意味します。面倒な部分は、の間にもっと多くのコールをに許可したいということです。

私はこのコードを試していませんが、セマフォーを解放する定期的な関数をに設定して、を10回実行することをお勧めします。したがって、このような何か：

private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(10); 

private async Task<Response> ThrottledSendRequestAsync(HttpRequestMessage request, CancellationToken token) 
{ 
    await _semaphore.WaitAsync(token); 
    return await SendRequestAsync(request, token); 
} 

private async Task PeriodicallyReleaseAsync(Task stop) 
{ 
    while (true) 
    { 
    var timer = Task.Delay(TimeSpan.FromSeconds(1.2)); 

    if (await Task.WhenAny(timer, stop) == stop) 
     return; 

    // Release the semaphore at most 10 times. 
    for (int i = 0; i != 10; ++i) 
    { 
     try 
     { 
     _semaphore.Release(); 
     } 
     catch (SemaphoreFullException) 
     { 
     break; 
     } 
    } 
    } 
} 

使用法：

// Start the periodic task, with a signal that we can use to stop it. 
var stop = new TaskCompletionSource<object>(); 
var periodicTask = PeriodicallyReleaseAsync(stop.Task); 

// Wait for all item processing. 
await Task.WhenAll(taskList); 

// Stop the periodic task. 
stop.SetResult(null); 
await periodicTask;