.Net Core with 微服务

在我们实施微服务之后，服务间的调用变的异常频繁。多个服务之间可能是互相依赖的关系。某个服务出现故障或者是服务间的网络出现故障都会造成服务调用的失败，进而影响到某个业务服务处理失败。某一个服务调用失败轻则造成当前相关业务无法处理；重则可能耗尽资源而拉垮整个应用。为了尽可能的保证我们生产环境的可用性，至少是部分可用性我们就需要一些策略来保护我们的服务。

服务降级

比如我们的订单详情服务里面会调用会员信息服务接口。如果会员信息服务接口故障会造成订单详情服务也同样故障。这时候我们可以对会员信息服务接口进行降级，在发生故障的时候直接返回固定的信息从而保证订单详情主服务是可用的。
另外一种情况是服务器的资源总是有限的，在面对突发的高并发，高流量情况下我们也可以对部分服务进行降级处理，从而释放更多的资源给核心服务，从而保证核心业务正常工作。

熔断

我们的服务很可能是一个链式的调用的过程。期间如果某个服务出现故障，特别是出现超时故障的时候很有可能耗尽服务器的资源从而影响整个服务。比如订单详情服务依赖会员信息服务，如果会员信息服务因为某些原因出现处理过慢、异常等情况，会阻塞整个订单详情服务的链路。而可能其它服务同样依赖订单详情服务，这样其它服务同样也会被阻塞。资源被越来越多的消耗而不释放，造成所有服务处理越来越慢，积压的请求越来越多， 犹如死循环一般，直到所有资源都被耗尽，整个生成环境奔溃。
所以面对这种情况当我们某个服务持续出现故障的时候我们可以直接断开对它的调用依赖，从而保证不会因为请求积压造成资源耗尽的情况发生。

Polly

Polly 是一个开源的弹性跟瞬态故障处理类库。它可以在你的程序出现故障，超时，或者返回值达成某种条件的时候进行多种策略处理，比如重试、降级、熔断等等。它是 .NET Foundation 的成员项目。

Policy.Handle< T >

Policy.Handle< T > 用来定义异常的类型，表示当执行的方法发生某种异常的时候定义为故障。
当故障发生的时候 Polly 会为我们自动执行某种恢复策略，比如重试。
我们演示项目中，订单接口需要获取会员的详细信息。
http 有一定几率失败，下面我们演示下如果使用 Polly 在出现当请求网络失败的时候进行3次重试。

var memberJson = await Policy.Handle<HttpRequestException>().RetryAsync(3).ExecuteAsync(async () =>
                {
                    using (var httpClient = new HttpClient())
                    {
                        httpClient.BaseAddress = new Uri($"http://{memberServiceAddress.Address}:{memberServiceAddress.Port}");
                        var memberResult = await httpClient.GetAsync("/member/" + order.MemberId);
                        memberResult.EnsureSuccessStatusCode();
                        var json = await memberResult.Content.ReadAsStringAsync();
                        return json;
                    }
                });

使用 Policy.Handle< HttpRequestException > 来捕获网络异常。当发生 HttpRequestException 的时候触发 RetryAsync 重试，并且最多重试3次。
以下我们接着演示下当 http 的返回值是500的时候进行3次重试：

Policy.HandleResult< T>

Policy.HandleResult< T > 用来定义返回值的类型，表示当执行的方法返回值达成某种条件的时候定义为故障。
当故障发生的时候 Polly 会为我们自动执行某种恢复策略，比如重试。
下面我们演示下如何使用 Polly 在出现当请求结果为 http status_code 500 的时候进行3次重试。

  var memberResult = await Policy.HandleResult<HttpResponseMessage>(x => (int)x.StatusCode == 500).RetryAsync(3).ExecuteAsync(async () =>
                  {
                      using (var httpClient = new HttpClient())
                      {
                          httpClient.BaseAddress =
                              new Uri($"http://{memberServiceAddress.Address}:{memberServiceAddress.Port}");
                          var result = await httpClient.GetAsync("/member/" + order.MemberId);
                          return result;
                      }
                  });

Policy.TimeoutAsync

Policy.TimeoutAsync 表示当一个操作超过设定时间时会引发一个 TimeoutRejectedException 。
这也是一个很常用的故障处理策略。

var memberJson = await Policy.TimeoutAsync(10).ExecuteAsync(async () =>
                {
                    using (var httpClient = new HttpClient())
                    {
                        httpClient.BaseAddress =
                            new Uri($"http://{memberServiceAddress.Address}:{memberServiceAddress.Port}");
                        var memberResult = await httpClient.GetAsync("/member/" + order.MemberId);
                        memberResult.EnsureSuccessStatusCode();
                        var json = await memberResult.Content.ReadAsStringAsync();
                        return json;
                    }
                });
            

以上代码表示当获取会员详情的接口超过10秒还未返回结果的时候直接抛出一个 TimeoutRejectedException 异常终止执行。

服务降级

以上我们演示了出现故障的时候如何进行重试，但是所有重试都失败我们的程序还是会故障。
因为期间某个服务持续的故障导致更多的服务出现故障，一系列连锁反应后很可能导致整个应用瘫痪。
面对这种情况我们可以把相关服务进行降级。
当相关服务调用失败的时候我们可以给出一个统一标准的失败返回值，而不是直接抛出异常。让我们的程序依然能够继续执行下去。
下面我们演示下如何使用 Polly 进行服务调用的降级处理。

 var fallback = Policy<string>.Handle<HttpRequestException>().FallbackAsync("FALLBACK")
                    .WrapAsync(Policy.Handle<HttpRequestException>().RetryAsync(3));
                var memberJson = await fallback.ExecuteAsync(async () =>
                {
                    using (var httpClient = new HttpClient())
                    {
                        httpClient.BaseAddress =
                            new Uri($"http://{memberServiceAddress.Address}:{memberServiceAddress.Port}");
                        var result = await httpClient.GetAsync("/member/" + order.MemberId);
                        result.EnsureSuccessStatusCode();
                        var json = await result.Content.ReadAsStringAsync();
                        return json;
                    }

                });
                if (memberJson != "FALLBACK")
                {
                    var member = JsonConvert.DeserializeObject<MemberVM>(memberJson);
                    vm.Member = member;
                }

首先我们使用 Policy 的 FallbackAsync("FALLBACK") 方法设置降级的返回值。当我们服务需要降级的时候会返回 "FALLBACK" 的固定值。
同时使用 WrapAsync 方法把重试策略包裹起来。这样我们就可以达到当服务调用失败的时候重试3次，如果重试依然失败那么返回值降级为固定的 "FALLBACK" 值。

熔断

通过以上演示，我们的服务当发生故障的时候可以自动重试，自动降级了。虽然现在看起来挺健壮，但是还是会有不小的问题。
当我们引入重试策略后，如果服务调用一直失败，每次调用都会反复进行重试，虽然最后会进行降级处理，但是这势必会影响服务的处理速度。
当流量很大的时候，某个接口服务调用很慢有可能会阻塞整个服务，请求不断积压，资源不断耗尽，速度越来越慢，这是一种恶性循环。最终同样可能导致整个应用全部瘫痪的严重后果。
面对这种情况我们就需要引入熔断机制。当一个服务的调用频繁出现故障的时候我们可以认为它当前是不稳定的，在一段时间内我们不应该再去调用这个服务。

        static AsyncCircuitBreakerPolicy circuitBreaker =  Policy.Handle<HttpRequestException>().CircuitBreakerAsync(
            exceptionsAllowedBeforeBreaking: 10,
            durationOfBreak: TimeSpan.FromSeconds(30),
        onBreak: (ex, ts) =>
        {
            Console.WriteLine("circuitBreaker onBreak .");
        },
        onReset: () =>
        {
            Console.WriteLine("circuitBreaker onReset ");
        },
        onHalfOpen: () =>
        {
            Console.WriteLine("circuitBreaker onHalfOpen");
        }
        );

         var retry = Policy.Handle<HttpRequestException>().RetryAsync(3);
                var fallback = Policy<string>.Handle<HttpRequestException>().Or<BrokenCircuitException>().FallbackAsync("FALLBACK")
                    .WrapAsync(circuitBreaker.WrapAsync(retry));
                var memberJson = await fallback.ExecuteAsync(async () =>
                {
                    using (var httpClient = new HttpClient())
                    {
                        httpClient.BaseAddress =
                            new Uri($"http://{memberServiceAddress.Address}:{memberServiceAddress.Port}");
                        var result = await httpClient.GetAsync("/member/" + order.MemberId);
                        result.EnsureSuccessStatusCode();
                        var json = await result.Content.ReadAsStringAsync();
                        return json;
                    }
                });
                if (memberJson != "FALLBACK")
                {
                    var member = JsonConvert.DeserializeObject<MemberVM>(memberJson);
                    vm.Member = member;
                }

首先定义 circuitBreaker 熔断器策略。这个策略注意最好定义成静态变量。这样能够以整个完整服务的错误为基础来判断是否开启断路器。
然后在业务代码内定义重试策略，降级策略。我们使这些策略一一嵌套。fallback => circuitBreaker => retry ，表示当发生异常的时候首先开始重试，
重试失败后尝试熔断，如果达到熔断的条件就抛出 BrokenCircuitException 异常，降级策略捕获到 HttpRequestException 或者 BrokenCircuitException 进行降级操作。
Polly 还有很多用法比如“缓存”、“隔离” 等策略，这里不在一一演示了。更多请查看文档：https://github.com/App-vNext/Polly/wiki

使用AOP思想改进体验

通过以上对于 Polly 的演示，虽然我们完成了简单的重试、服务降级、熔断等功能。但是显然对于每个方法都去使用 Polly 编写一堆策略的话实在是太麻烦了。那么有什么办法能改进一下 Polly 的使用体验吗？答案是使用 AOP 的思想，通过在执行的方法上打上 Attribute 的方式来指定 Polly 的策略。
下面我们使用 lemon 大佬的 AspectCore AOP 组件结合 Polly 来演示下如何通过 AOP 的思想来处理重试、降级、熔断等策略。

Install-Package AspectCore.Core

通过 nuget 安装 AspectCore 核心类库。

 public class PollyHandleAttribute : AbstractInterceptorAttribute
    {
        /// <summary>
        /// 重试次数
        /// </summary>
        public int RetryTimes { get; set; } 

        /// <summary>
        /// 是否熔断
        /// </summary>
        public bool IsCircuitBreaker { get; set; }

        /// <summary>
        /// 熔断前的异常次数
        /// </summary>
        public int ExceptionsAllowedBeforeBreaking { get; set; }

        /// <summary>
        /// 熔断时间
        /// </summary>
        public int SecondsOfBreak { get; set; }

        /// <summary>
        /// 降级方法
        /// </summary>
        public string FallbackMethod { get; set; }

        /// <summary>
        /// 一些方法级别统一计数的策略，比如熔断
        /// </summary>
        static ConcurrentDictionary<string, AsyncCircuitBreakerPolicy> policyCaches = new ConcurrentDictionary<string, AsyncCircuitBreakerPolicy>();

        public PollyHandleAttribute()
        {

        }

        public override async Task Invoke(AspectContext context, AspectDelegate next)
        {
            Context pollyCtx = new Context();
            pollyCtx["aspectContext"] = context;

            Polly.Wrap.AsyncPolicyWrap policyWarp = null;

            var retry = Policy.Handle<HttpRequestException>().RetryAsync(RetryTimes);
            var fallback = Policy.Handle<Exception>().FallbackAsync(async (fallbackContent, token) =>
            {
                AspectContext aspectContext = (AspectContext)fallbackContent["aspectContext"];
                var fallBackMethod = context.ServiceMethod.DeclaringType.GetMethod(this.FallbackMethod);
                var fallBackResult = fallBackMethod.Invoke(context.Implementation, context.Parameters);
                aspectContext.ReturnValue = fallBackResult;
            }, async (ex, t) => { });
            AsyncCircuitBreakerPolicy circuitBreaker = null;
            if (IsCircuitBreaker)
            {
                var cacheKey = $"{context.ServiceMethod.DeclaringType.ToString()}_{context.ServiceMethod.Name}";
                if (policyCaches.TryGetValue(cacheKey, out circuitBreaker))
                {
                    //从缓存内获取该方法的全局熔断策略
                }
                else
                {
                    circuitBreaker = Policy.Handle<Exception>().CircuitBreakerAsync(
                      exceptionsAllowedBeforeBreaking: this.ExceptionsAllowedBeforeBreaking,
                      durationOfBreak: TimeSpan.FromSeconds(this.SecondsOfBreak));

                    policyCaches.TryAdd(cacheKey, circuitBreaker);
                }
            }

            if (circuitBreaker == null)
            {
                policyWarp = fallback.WrapAsync(retry);
            }
            else
            {
                policyWarp = fallback.WrapAsync(circuitBreaker.WrapAsync(retry));
            }


            await policyWarp.ExecuteAsync(ctx => next(context), pollyCtx);
        }
    }

定义一个 PollyHandleAttribute 类，它继承自 AbstractInterceptorAttribute 类，然后实现 Invoke 方法。我们需要在 Invoke 方法内动态构造出 Polly 的相关策略，然后通过 Polly 去执行真正的方法。这里主要需要注意的是熔断策略不能每次新建，因为对于熔断来说是需要全局统计该方法的异常数量来判断是否熔断的，所以需要把熔断策略缓存起来。
这个类参考了 Edison Zhou 大佬的部分代码，原文：Polly+AspectCore实现熔断与降级机制

   public interface IMemberService
    {
        Task<MemberVM> GetMemberInfo(string id);
        MemberVM GetMemberInfoFallback(string id);
    }

public class MemberService : IMemberService
    {
        private IConsulService _consulservice;

        public MemberService(IConsulService consulService)
        {
            _consulservice = consulService;
        }

        [PollyHandle(IsCircuitBreaker = true, FallbackMethod = "GetMemberInfoFallback", ExceptionsAllowedBeforeBreaking = 5, SecondsOfBreak = 30, RetryTimes = 3)]
        public async Task<MemberVM> GetMemberInfo(string id)
        {
            var memberServiceAddresses = await _consulservice.GetServicesAsync("member_center");
            var memberServiceAddress = memberServiceAddresses.FirstOrDefault();

            using (var httpClient = new HttpClient())
            {
                httpClient.BaseAddress =
                    new Uri($"http://{memberServiceAddress.Address}:{memberServiceAddress.Port}");
                var result = await httpClient.GetAsync("/member/" + id);
                result.EnsureSuccessStatusCode();
                var json = await result.Content.ReadAsStringAsync();

                if (string.IsNullOrEmpty(json))
                {
                    return JsonConvert.DeserializeObject<MemberVM>(json);
                }
            }

            return null;
        }

        public MemberVM GetMemberInfoFallback(string id)
        {
            return null;
        }
    }

因为我们需要在方法上标记 PollyHandleAttribute ，所以把获取会员相关的逻辑封住进 MemberService 的 GetMemberInfo 方法内。并且在方法上打上Attribute ：
[PollyHandle(IsCircuitBreaker = true, FallbackMethod = "GetMemberInfoFallback", ExceptionsAllowedBeforeBreaking = 5, SecondsOfBreak = 30, RetryTimes = 3)] 直接通过 AOP 的方式来配置 Polly 的策略，这样就方便了很多。
上面这些配置好之后，下面开始就是如何使 aspectcore 接管 asp.net core 的依赖注入了。根据文档也很简单：

Install-Package AspectCore.Extensions.DependencyInjection

通过 nuget 安装 AspectCore.Extensions.DependencyInjection 包。

  public static IHostBuilder CreateHostBuilder(string[] args) =>
            Host.CreateDefaultBuilder(args)
                .ConfigureWebHostDefaults(webBuilder =>
                {
                    webBuilder.ConfigureKestrel(options =>
                    {
                        options.ListenAnyIP(6001);
                    });
                    webBuilder.UseStartup<Startup>();
                })
                .UseServiceProviderFactory(new DynamicProxyServiceProviderFactory());

在 CreateHostBuilder 内使用 UseServiceProviderFactory 替换 ServiceProviderFactory 为 aspectcore 的实现。

     public void ConfigureServices(IServiceCollection services)
        {
            services.AddSingleton<IMemberService, MemberService>();

            ...

            services.ConfigureDynamicProxy();
        }

在 ConfigureServices 方法内配置 IMemberService 的依赖关系以及配置 aspectcore 的动态代理。

总结

通过以上文字我们大致了解了什么是服务降级、什么是熔断。并且通过 Polly 演示了如何处理这些情况。最后使用 lemon 大佬的 AspectCore 封装成一个 Attribute 来演示如何通过 AOP 的思想来简化 Polly 的使用。

谢谢阅读。

演示项目地址

https://github.com/kklldog/myhotel_microservice

相关文章

NET Core with 微服务 - 什么是微服务
.Net Core with 微服务 - 架构图
.Net Core with 微服务 - Ocelot 网关
.Net Core with 微服务 - Consul 注册中心
.Net Core with 微服务 - Seq 日志聚合
.Net Core with 微服务 - Elastic APM
.Net Core with 微服务 - Consul 配置中心

关注我的公众号一起玩转技术