c# - Polly - 调用异步 ExecuteAsync (和类似)方法时请使用异步定义的策略

Question

我在执行包装策略时遇到上述异常，包括:重试、断路器和隔板。

我有以下政策:

var sharedBulkhead = Policy.BulkheadAsync(
            maxParallelization: maxParallelizations, 
            maxQueuingActions: maxQueuingActions,
            onBulkheadRejectedAsync: (context) =>
            {
                Log.Info($"Bulk head rejected => Policy Wrap: {context.PolicyWrapKey}, Policy: {context.PolicyKey}, Endpoint: {context.OperationKey}");
                return TaskHelper.EmptyTask;
            }
        );

var retryPolicy = Policy.Handle<Exception>(e => (e is HttpRequestException)).WaitAndRetryAsync(
            retryCount: maxRetryCount,
            sleepDurationProvider: attempt => TimeSpan.FromSeconds(Math.Pow(2, attempt)),
            onRetryAsync: (exception, calculatedWaitDuration, retryCount, context) =>
            {
                Log.Error($"Retry => Count: {retryCount}, Wait duration: {calculatedWaitDuration}, Policy Wrap: {context.PolicyWrapKey}, Policy: {context.PolicyKey}, Endpoint: {context.OperationKey}, Exception: {exception}.");
                return TaskHelper.EmptyTask;
            });

            var circuitBreaker = Policy.Handle<Exception>(e => (e is HttpRequestException)).CircuitBreakerAsync(
            exceptionsAllowedBeforeBreaking: maxExceptionsBeforeBreaking, 
            durationOfBreak: TimeSpan.FromSeconds(circuitBreakDurationSeconds), 
            onBreak: (exception, timespan, context) =>
            {
                Log.Error($"Circuit broken => Policy Wrap: {context.PolicyWrapKey}, Policy: {context.PolicyKey}, Endpoint: {context.OperationKey}, Exception: {exception}");
            },
            onReset: (context) =>
            {
                Log.Info($"Circuit reset => Policy Wrap: {context.PolicyWrapKey}, Policy: {context.PolicyKey}, Endpoint: {context.OperationKey}");
            }
        );

var fallbackForCircuitBreaker = Policy<bool>
         .Handle<BrokenCircuitException>()
         .FallbackAsync(
             fallbackValue: false,
             onFallbackAsync: (b, context) =>
             {
                 Log.Error($"Operation attempted on broken circuit => Policy Wrap: {context.PolicyWrapKey}, Policy: {context.PolicyKey}, Endpoint: {context.OperationKey}");
                 return TaskHelper.EmptyTask;
             }
         );

var fallbackForAnyException = Policy<bool>
            .Handle<Exception>()
            .FallbackAsync(
                fallbackAction: (ct, context) => { return Task.FromResult(false); },
                onFallbackAsync: (e, context) =>
                {
                    Log.Error($"An unexpected error occured => Policy Wrap: {context.PolicyWrapKey}, Policy: {context.PolicyKey}, Endpoint: {context.OperationKey}");
                    return TaskHelper.EmptyTask;
                }
            );


var resilienceStrategy = Policy.WrapAsync(retryPolicy, circuitBreaker, sharedBulkhead);
        var policyWrap = fallbackForAnyException.WrapAsync(fallbackForCircuitBreaker.Wrap(resilienceStrategy));

我像这样执行策略:

Task.Run(() =>
        {
            foreach (var changeMessage in changeMessages)
            {
                policyWrap.ExecuteAsync((context) => CallApi(changeMessage), new Context(endPoint));
            }
        });

这会产生上述异常:“在调用异步 ExecuteAsync(和类似)方法时，请使用异步定义的策略。”在 fallbackForAnyException 内。我做错了什么？

Answer 1

您在这里结合了同步和异步执行，因此出现了错误消息。这是因为最后一行代码:

var policyWrap = fallbackForAnyException.WrapAsync(
    fallbackForCircuitBreaker.Wrap(resilienceStrategy));
                            //^^^^

请注意我突出显示的部分。那应该是:

var policyWrap = fallbackForAnyException.WrapAsync(
    fallbackForCircuitBreaker.WrapAsync(resilienceStrategy));