服务保障框架-Hystrix快速入门

用户1216676

发布于 2018-10-11 11:04:17

9720

发布于 2018-10-11 11:04:17

文章被收录于专栏：熊二哥

祝大家国庆快乐！对大部分电商和快递公司来说，每年年底（Q4季度）由于双11等大促活动的存在，将面对大量的用户流量，尤其是属于大促的那几天，无论是用户的商品订单还是物流订单，都将是平时的3倍以上。对于技术人员来说，提前落地相应的服务保障体系，并进行相应的压测和演习，是题中应有之意。整个保障体系的实现涉及的环节很多，本文将选取奈飞Netflix公司的Hystrix"豪猪"框架（其基于Java语言和最近比较流行RxJava流式框架），针对分布式应用的服务保障问题进行探讨，之后将按照基本知识、应用实践、配置知识和源码分析的顺序进行介绍，不足之处望不吝赐教。

首先通过一张思维导图来展示本文的思路，有标记部分的推荐程度高，图可以拖到浏览器新窗口放大。经过大半年的调整，手还是比较生，我熊二哥又回来了！:)

基本知识

为了便于理解本文的意图，首先提出并解答两个问题。

1.为什么需要在项目中引入Hystrix，其可以应用在什么场景中？

在分布式系统中，单个应用通常会有多个不同类型的外部依赖服务，内部通常依赖于各种RPC服务，外部则依赖于各种HTTP服务。这些依赖服务不可避免的会出现调用失败，比如超时、异常等情况，如何在外部依赖出问题的情况，仍然保证自身应用的稳定，就是Hystrix这类服务保障框架的工作了。常见的服务依赖如下图所示，应用X依赖于服务A、B和C，A和B正常提供服务，C服务出错，这是如何避免C服务对A、B服务产生影响，也引出了一个隔离的概念。

举个例子来说，某个应用中依赖了30个外部服务，实际应用中通常比这还要多，假设每个服务的可用性为99.9%，3个9的可用性，算是不错了，但99.9%的30次幂≈ 97.0%，这个可用性已经是无法容忍的了。

2.Hytrix的目标是什么，其采用了什么手段来达到该目标？

Hystrix的目标就是能够在1个或多个依赖出现问题时，系统依然可以稳定的运行，其手段包括隔离、限流和降级有等，接下来详细介绍这些手段。补充一点，张开涛老师曾对系统高可用手段进行过总结，除了以上的限流、隔离和降级，还有负载均衡、超时与重试、回滚、压测与预案，共7种手段。

隔离

隔离说到底还是分治思想的体现，在当前场景中，就是将不同的外部依赖进行分类，确定其边界，然后隔离开来分开进行管理。Hystrix支持的隔离策略isolationStrategy包括信号量和线程池两种，具体内容将在之后限流知识中介绍。

限流

在基于服务化（包括SOA和微服务）的系统架构中，对服务请求进行限流是保护服务稳定性的一个常见手段。此外，关于限流有两个比较重要的概念：限流算法，包括计数限流、令牌桶和漏桶等；限流粒度，包括方法级别、接口级别、应用级别、集群级别等。对于Hystrix来说，其采用了自己的一套限流方式，这里首先延续之前隔离知识中提到的信号量和线程池概念进行介绍。

信号量概念比较简单，常用于获取共享资源的场景中，比如计算机连接了两个打印机，那么初始的信号量就是2，被某个进程或线程获取后减1，信号量为0后，需要获取的线程或进程进入资源等待状态。Hystrix的处理有些不同，其不等待，直接返回失败。

线程池采用的就是jdk的线程池，其默认选用不使用阻塞队列的线程池，例如线程池大小为10，如果某时刻10个线程均被使用，那么新的请求将不会进入等待队列，而是直接返回失败，起到限流的作用。

此外，其还引入了一个断路器机制，当断路器处于打开状态时，直接返回失败或进入降级流程。断路器打开和关闭的触发流程为：当总的请求数达到可阈值HystrixCommandProperties.circuitBreakerRequestVolumeThreshold()，或总的请求失败百分比达到了阈值HystrixCommandProperties.circuitBreakerErrorThresholdPercentage()，这时将断路器的状态由关闭设置为打开。当断路器打开时，所有的请求均被短路，在经过指定休眠时间窗口后，让下一个请求通过（断路器被认为是半开状态）。如果请求失败，断路器进入打开状态，并进入新的休眠窗口；否则进入关闭状态。

断路器依赖的统计信息如下图所示，默认情况下10s为一个统计周期，10个滚动窗口，每个负责统计1s内的数据，包括请求成功、失败、超时和拒绝次数。

降级

这里的降级具体来说就是服务质量的降级，需要注意的是，只有方法所属的业务场景适合降级时才采用，一般为查询场景。Hystrix通过配置fallbackMethod指定降级时的处理方法，触发降级动作的4种情况如下所示。

run()方法抛出非HystrixBadRequestException异常。
run()方法调用超时
熔断器开启拦截调用
线程池/队列/信号量是否跑满

Hystrix整体的处理流程

主题流程如图所示，Hystrix框架通过命令模式来实现方法粒度上的服务保障，主要涉及HystrixCommand和HystrixObservableCommand类，前者提供同步的execute和异步的queue方法，后者提供立即执行observe和延迟执行toObservable的回调方法。此外，实际项目中通常不会使用Hystrix集成的本地缓存。

tip: 目前在服务保障方面，除了hystrix框架外，阿里巴巴公司开源的sentinel框架也是一个不错的可选方案。

应用实践

该节将从基础应用、项目应用、动态配置和监控等几个方面进行介绍。Hystrix基础应用比较简单，包括直接编码和使用注解等两种方式，一般选用注解方式，其基于javanica子包，hystrix-javanica官网，之后简要展示Hystrix如何在基于gradle依赖管理的Springboot应用中集成。

基础应用

1.Gradle配置和SpringBoot配置

//Gradle中添加Hystrix核心
compile('org.springframework.cloud:spring-cloud-starter-netflix-hystrix')

@Configuration
public class HystrixConfiguration {
  @Bean
  public HystrixCommandAspect hystrixAspect() {
    return new HystrixCommandAspect();
  }
}

2.同步使用方式（异步和响应式可以参考javanica的wiki页面）

@DefaultProperties(groupKey = "UserQueryGroup", threadPoolProperties = {
        @HystrixProperty(name = "coreSize", value = "20"),
        @HystrixProperty(name = "maxQueueSize", value = "20")
})
@Service
public class UserQueryManager {
    @HystrixCommand(commandKey = "GetUserCommand",fallbackMethod = "getUserFallback")
    public String getUser(long id) {
        return "test_user";
    }
    
    public String getUserFallback(long id) {
        return "test_user_fallback";
    }
}

项目应用

新建项目直接在涉及外部依赖服务的方法上加上相应注解即可，比较简单。而对于既有系统，为了降低相关风险，推荐采用引入开关变量，AB灰度的方式，具体方式如下图所示。

动态配置

在实际应用中，当发现在线应用的命令或线程池相关参数不合理时，如何进行参数的实时调优？目前，Hystrix提供了ConfigurationManager配置管理类来实时管理配置信息，是配置相关的核心类，既可以通过实现PolledConfigurationSource类，借助FixedDelayPollingScheduler类定时的PULL最新的配置信息，也可以通过自定义的方式监听相关配置项的修改以PUSH方式对配置进行修改。此外，每个Hystrix参数都有4个地方可以配置，优先级从低到高如下，如果每个地方都配置相同的属性，则优先级高的值会覆盖优先级低的值。

内置全局默认值：写死在代码里的值采用ConcurrentHashMap 有 HystrixCommandProperties HystrixThreadPoolProperties HystrixCollapserProperties
动态全局默认属性：通过属性文件配置全局的值
内置实例默认值：写死在代码里的实例的值
动态配置实例属性：通过属性文件配置特定实例的值

Tip: Hystrix默认的配置框架，archaius

应用监控

Hystrix源生提供了单机和集群的监控服务，单机借助Hystrix-Dashboard，集群借助Turbine，这里只介绍单机监控的实现，代码如下。

//Gradle中添加Hystrix面板
compile('org.springframework.cloud:spring-cloud-starter-netflix-hystrix-dashboard')

@Bean
public ServletRegistrationBean getServlet() {
    HystrixMetricsStreamServlet streamServlet = new HystrixMetricsStreamServlet();
    ServletRegistrationBean registrationBean = new ServletRegistrationBean(streamServlet);
    registrationBean.setLoadOnStartup(1);
    registrationBean.addUrlMappings("/hystrix.stream");
    registrationBean.setName("HystrixMetricsStreamServlet");
    return registrationBean;
}

之后直接访问localhost:port/hystrix即可进入面板管理页面，配置好hystrix.stream信息后就可以看到如下监控页面。可以看到getArticle方法的失败率达到59.0%，断路器已打开，大量请求被降级，请求峰值得到缓解，这部分可以使用Jmeter进行测试。

配置知识

在了解hystrix时，我最开始就曾被3个KEY给打败过，其分别是CommandGroupKey，CommandKey和ThreadPoolKey。其实可以通过一个很简单的划方式就可以将这3个KEY区分开，CommandGroupKey是一个纯逻辑的概念，其可以管理多个CommandKey，且在默认情况下ThreadPool和它同名，而后两者则带有实际意味，之后的配置信息可以看到，所有的配置都是基于Command和ThreadPool的。

关键配置

Hystrix的配置项比较多，大概有30个左右，但比较基础和关键的就是以下的10来个配置项，主要包括CommandProperties和ThreadPoolProperties两部分。

命令配置中，隔离策略包括线程池和信号量两种，默认和推荐使用前者，线程的超时时间一般设置为比依赖调用的99线平均时间略高即可。断路器部分，请求数量的熔断阈值和请求失败比例的熔断阈值推荐更加实际的测试请求进行设置，统计信息的滑动窗口大小和分桶数采用默认值通常就可以满足需求。

线程池配置中，主要就是线程大小的设置，默认为10个，推荐根据所管理服务的单机QPS和TP99线计算得出，这部分支持动态配置，可以在线实时调整。这部分配置很重要，虽然Hystrix推荐创建40个左右的线程池，每个10个线程左右，但实际项目中，一定要对当前应用的依赖服务进行合理分类，否则大量的线程池和线程会对应用带来一定不良影响。

源码分析

Hystrix由于引入了rxJava响应式编程，代码风格与过去习惯的结构化风格有一些差异，接下来从@HystrixCommand注解解析开始，简要展示命令执行的整个过程，解析在代码注释中。

HystrixCommandAspect

@Around("hystrixCommandAnnotationPointcut() || hystrixCollapserAnnotationPointcut()")
public Object methodsAnnotatedWithHystrixCommand(final ProceedingJoinPoint joinPoint) throws Throwable {
    Method method = getMethodFromTarget(joinPoint);
    ...
    MetaHolderFactory metaHolderFactory = META_HOLDER_FACTORY_MAP.get(HystrixPointcutType.of(method));
    MetaHolder metaHolder = metaHolderFactory.create(joinPoint);
    HystrixInvokable invokable = HystrixCommandFactory.getInstance().create(metaHolder);//1.创建metaHolder
    ExecutionType executionType = metaHolder.isCollapserAnnotationPresent() ?
            metaHolder.getCollapserExecutionType() : metaHolder.getExecutionType();
    Object result;
    try {//2.命令执行
        if (!metaHolder.isObservable()) {
            result = CommandExecutor.execute(invokable, executionType, metaHolder);
        } else {
            result = executeObservable(invokable, executionType, metaHolder);
        }
    } ...
}

CommandExecutor

public static Object execute(HystrixInvokable invokable, ExecutionType executionType, MetaHolder metaHolder) throws RuntimeException {
    ...
    switch (executionType) {
        case SYNCHRONOUS: {//1.同步执行，其实其内部也是用的异步执行queue().get()
            return castToExecutable(invokable, executionType).execute();
        }
        case ASYNCHRONOUS: {//2.异步执行
            HystrixExecutable executable = castToExecutable(invokable, executionType);
            if (metaHolder.hasFallbackMethodCommand()
                    && ExecutionType.ASYNCHRONOUS == metaHolder.getFallbackExecutionType()) {
                return new FutureDecorator(executable.queue());
            }
            return executable.queue();
        }
        case OBSERVABLE: {//3.响应式执行，observable.toObservable()是核心方法
            HystrixObservable observable = castToObservable(invokable);
            return ObservableExecutionMode.EAGER == metaHolder.getObservableExecutionMode() ? observable.observe() : observable.toObservable();
        }
      ...
}

AbstractCommand

public Observable<R> toObservable() {
   ...
    final Func0<Observable<R>> applyHystrixSemantics = new Func0<Observable<R>>() {
        @Override
        public Observable<R> call() {
            if (commandState.get().equals(CommandState.UNSUBSCRIBED)) {
                return Observable.never();
            }
            return applyHystrixSemantics(_cmd);//1.关键步骤，命令处理
        }
    };
  ...
  
private Observable<R> applyHystrixSemantics(final AbstractCommand<R> _cmd) {
    ...
    if (circuitBreaker.attemptExecution()) {//1.【断路器相关处理】，之后HystrixCircuitBreaker中展示
        ..
        if (executionSemaphore.tryAcquire()) {//2.获取信号量，如果是THREAD线程池策略，【直接返回true】，这里需要注意，不然流程将进行不下去
            try {
                executionResult = executionResult.setInvocationStartTime(System.currentTimeMillis());
                return executeCommandAndObserve(_cmd)//3.核心执行方法
                        .doOnError(markExceptionThrown)
                        .doOnTerminate(singleSemaphoreRelease)
                        .doOnUnsubscribe(singleSemaphoreRelease);
            } ...
}

private Observable<R> executeCommandAndObserve(final AbstractCommand<R> _cmd) {
    ...
    final Func1<Throwable, Observable<R>> handleFallback = new Func1<Throwable, Observable<R>>() {
        @Override
        public Observable<R> call(Throwable t) {
            circuitBreaker.markNonSuccess();
            Exception e = getExceptionFromThrowable(t);
            executionResult = executionResult.setExecutionException(e);
            if (e instanceof RejectedExecutionException) {
                return handleThreadPoolRejectionViaFallback(e);//1.处理线程池拒绝场景
            } else if (t instanceof HystrixTimeoutException) {
                return handleTimeoutViaFallback();//2.处理请求超时场景
            } else if (t instanceof HystrixBadRequestException) {
                return handleBadRequestByEmittingError(e);
            } else {
                /*
                 * Treat HystrixBadRequestException from ExecutionHook like a plain HystrixBadRequestException.
                 */
                if (e instanceof HystrixBadRequestException) {
                    eventNotifier.markEvent(HystrixEventType.BAD_REQUEST, commandKey);
                    return Observable.error(e);
                }
                return handleFailureViaFallback(e);//3.处理请求失败场景
            }
        }
    };

HystrixCircuitBreaker

@Override
public boolean attemptExecution() {
    if (properties.circuitBreakerForceOpen().get()) {//1.断路器是否强制打开
        return false;
    }
    if (properties.circuitBreakerForceClosed().get()) {//2.断路器是否强制关闭
        return true;
    }
    if (circuitOpened.get() == -1) {//3.短路器关闭
        return true;
    } else {
        if (isAfterSleepWindow()) {//4.是否经过了指定的窗口时间
            if (status.compareAndSet(Status.OPEN, Status.HALF_OPEN)) {//5.设置为半打开状态
                //only the first request after sleep window should execute
                return true;
            } else {
                return false;
            }
        } else {
            return false;
        }
    }
}

Tip: 老铁告知，目前阿里已经在部分业务项目中落地基于RxJava的全异步应用，大幅提高了资源的利用效率，大家有空可以关注下，【Java11】本周好像也已新鲜出炉。

附录

名词解释

TP=Top Percentile，Top百分数，是一个统计学里的术语，与平均数、中位数都是一类。 TP50、TP90和TP99等指标常用于系统性能监控场景，指高于50%、90%、99%等百分线的情况。 OSS = Open Source Software，开源软件（开放源代码软件）。 OSS= Operation Support System，运营支撑系统。