每主机使用HttpClient和Polly发送并行请求，以正常处理429个响应

Question

Intro：

我正在构建一个单节点Web搜寻器，以简单地验证URL在.NET Core控制台应用程序中是否为200 OK。我在使用HttpClient向其发送请求的不同主机上有一组URL。我对使用Polly和TPL数据流还很陌生。

需求：

1. 我想支持同时发送多个HTTP请求和可配置的MaxDegreeOfParallelism。
2. 我想将对任何给定主机的并行请求数限制为1（或可配置）。这是为了用Polly策略妥善处理每主机429 TooManyRequests响应。或者，我可能会使用断路器来取消收到同一429响应的同一主机的并发请求，然后一次处理该特定主机？
3. 我完全不使用TPL Dataflow很好，而赞成对节流的并行请求使用Polly Bulkhead或其他机制，但是我不确定为了实现要求＃2该配置是什么样。] >

当前实施：

我当前的实现工作正常，只是我经常看到对同一主机的x并行请求大约在同一时间返回429 ...然后，它们全部暂停以执行重试策略...然后，他们全都在同一时间再次猛击同一台主机，但仍然经常收到429。即使我在整个队列中平均分配了同一主机的多个实例，我的URL集合也会被一些特定的主机超载，这些主机最终仍会开始生成429。

[收到429之后，我想我只想向该主机发送一个并发请求，以尊重远程主机并追求200。

验证器方法：

public async Task<int> GetValidCount(IEnumerable<Uri> urls, CancellationToken cancellationToken)
{
    var validator = new TransformBlock<Uri, bool>(
        async u => (await _httpClient.GetAsync(u, HttpCompletionOption.ResponseHeadersRead, cancellationToken)).IsSuccessStatusCode,
        new ExecutionDataflowBlockOptions {MaxDegreeOfParallelism = MaxDegreeOfParallelism}
    );
    foreach (var url in urls)
        await validator.SendAsync(url, cancellationToken);
    validator.Complete();
    var validUrlCount = 0;
    while (await validator.OutputAvailableAsync(cancellationToken))
    {
        if(await validator.ReceiveAsync(cancellationToken))
            validUrlCount++;
    }
    await validator.Completion;
    return validUrlCount;
}
应用于上面GetValidCount()中使用的HttpClient实例的Polly策略。

IAsyncPolicy<HttpResponseMessage> waitAndRetryTooManyRequests = Policy
    .HandleResult<HttpResponseMessage>(r => r.StatusCode == HttpStatusCode.TooManyRequests)
    .WaitAndRetryAsync(3,
        (retryCount, response, context) =>
            response.Result?.Headers.RetryAfter.Delta ?? TimeSpan.FromMilliseconds(120),
        async (response, timespan, retryCount, context) =>
        {
            // log stuff
        });
问题：

我如何修改或替换此解决方案以增加对要求2的满足？

简介：我正在构建一个单节点Web搜寻器，以仅验证.NET Core控制台应用程序中的URL是否正确200。我在向其发送请求的不同主机上有一组URL ...

Answer 1

我将尝试引入某种标志LimitedMode以检测到此特定客户端以受限模式进入。在下面，我声明了两种策略-一种简单的重试策略，仅用于捕获TooManyRequests并设置标志。第二个策略是现成的BulkHead策略。

Answer 2

这里是为具有相同密钥的消息创建并发级别受限的TransformBlock的方法。通过调用提供的TransformBlock功能可获得每个消息的键。具有相同密钥的消息将彼此顺序处理（而不是并行处理）。该键也作为参数传递给keySelector函数，因为在某些情况下它可能很有用。