处理并发 http 请求时的 IHttpClient Polly Timeout 和 WaitAndRetry 策略说明

Question

只是对 Pollys 超时/重试策略以及它在处理并发 http 请求时的工作原理有疑问。

根据阅读/我自己的理解，超时和重试策略将应用于每个单独的 http 请求，所以如果我们有 5 个 http 请求，每个请求都会有自己的超时和重试策略，所以从下面的代码中每个 http请求将在 5 秒后超时并重试 4 次。

    public override void Configure(IFunctionsHostBuilder builder)
    {
        
        var timeout = Policy.TimeoutAsync<HttpResponseMessage>(TimeSpan.FromSeconds(5));

        builder.Services
            .AddHttpClient("PointsbetClient")
            .AddPolicyHandler(GetRetryPolicy())
            .AddPolicyHandler(timeout);
    }

    private static IAsyncPolicy<HttpResponseMessage> GetRetryPolicy()
    {
        return HttpPolicyExtensions
            .HandleTransientHttpError()
            .Or<TimeoutRejectedException>()
            .WaitAndRetryAsync(Backoff.DecorrelatedJitterBackoffV2(
                medianFirstRetryDelay: TimeSpan.FromMilliseconds(500),
                retryCount: 4));
    }

现在假设我有 1000 个 http 请求，我需要对其进行 GetAsync() 调用，以便我可以抓取它们的数据并出于性能目的，我使用 await Task.WhenAll(tasks); 同时进行这些调用。由于 1000 次请求太多，因此我使用 SemaphoreSlim 类并将 MaxParallelRequests 限制为 100。

Polly 重试和超时政策现在将如何应用？它是否仍然适用于每个单独的 1000 个请求，还是会将包含 100 个请求的 1 个任务视为单个超时/重试策略？据我了解，它仍然会处理每个单独的 http 请求并将其应用策略，我一直在搜索，但找不到对此的确认。

Answer 1

简短的回答是肯定的，它们是分开处理的。

---

为了了解系统的工作原理，我们必须深入了解一下。让我们在AddPolicyHandler 开始我们的旅程。

免责声明：为简洁起见，我稍微编辑了代码 sn-ps。

public static IHttpClientBuilder AddPolicyHandler(this IHttpClientBuilder builder, IAsyncPolicy<HttpResponseMessage> policy)
{
    if (builder == null) throw new ArgumentNullException(nameof(builder));
    if (policy == null) throw new ArgumentNullException(nameof(policy));
    builder.AddHttpMessageHandler(() => new PolicyHttpMessageHandler(policy));
    return builder;
}

此方法在PollyHttpClientBuilderExtensions 类中定义，它为IHttpClientBuilder 提供扩展方法。

如您所见，它只是将另一个HttpMessageHandler 注册到链中。

现在，让我们看看这个特殊的处理程序是什么样子的

protected override async Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
{
    if (request == null) throw new ArgumentNullException(nameof(request));

    // Guarantee the existence of a context for every policy execution, 
    // but only create a new one if needed.
    // This allows later handlers to flow state if desired.
    var cleanUpContext = false;
    var context = request.GetPolicyExecutionContext();
    if (context == null)
    {
        context = new Context();
        request.SetPolicyExecutionContext(context);
        cleanUpContext = true;
    }

    HttpResponseMessage response;
    try
    {
        var policy = _policy ?? SelectPolicy(request);
        response = await policy.ExecuteAsync((c, ct) => SendCoreAsync(request, c, ct), context, cancellationToken).ConfigureAwait(false);
    }
    finally
    {
        if (cleanUpContext)
            request.SetPolicyExecutionContext(null);
    }

    return response;
}

这个方法是在PolicyHttpMessageHandler里面定义的。

如您所见，这里没有发生任何异常情况

• 我们要么检索要么创建一个新的上下文
• 我们要么从注册表中检索策略，要么使用提供的策略
• 我们执行装饰SendCoreAsync的政策

那么，魔法发生在哪里？让我们跳转到这个类的documentation comment

Polly 提供的所有政策都旨在长期有效地使用。某些政策，例如 Bulkhead 和 Circuit-Breaker 保持状态，并且应该在您希望共享 Bulkhead 或 Circuit-Breaker 状态的调用中进行范围划分。 在自定义场景中同时使用策略和消息处理程序时，请注意确保正确的生命周期。扩展名 PollyHttpClientBuilderExtensions 提供的方法旨在为策略分配较长的生命周期 并确保它们可以在处理程序轮换功能处于活动状态时使用。

要了解 Retry 与 Circuit Breaker 有何不同，请查看他们的 Engines' Implementation 签名

RetryEngine

internal static class RetryEngine
{
    internal static TResult Implementation<TResult>(
        Func<Context, CancellationToken, TResult> action,
        Context context,
        CancellationToken cancellationToken,
        ExceptionPredicates shouldRetryExceptionPredicates,
        ResultPredicates<TResult> shouldRetryResultPredicates,
        Action<DelegateResult<TResult>, TimeSpan, int, Context> onRetry,
        int permittedRetryCount = Int32.MaxValue,
        IEnumerable<TimeSpan> sleepDurationsEnumerable = null,
        Func<int, DelegateResult<TResult>, Context, TimeSpan> sleepDurationProvider = null)
    {
    ...
    }
}

CircuitBreakerEngine

internal class CircuitBreakerEngine
{
    internal static TResult Implementation<TResult>(
        Func<Context, CancellationToken, TResult> action,
        Context context,
        CancellationToken cancellationToken,
        ExceptionPredicates shouldHandleExceptionPredicates, 
        ResultPredicates<TResult> shouldHandleResultPredicates, 
        ICircuitController<TResult> breakerController)
    {
    ...
    }
}

您必须在这里发现ICircuitController。 CircuitStateController 基类存储状态信息。其派生类之一是不同策略执行之间的shared

internal readonly ICircuitController<EmptyStruct> _breakerController;
...
CircuitBreakerEngine.Implementation(
    action,
    context,
    cancellationToken,
    ExceptionPredicates,
    ResultPredicates,
    _breakerController);

---

我希望这能澄清一些事情。