从源码分析Hystrix工作机制

作者：vivo互联网服务器团队-Pu Shuai

一、Hystrix解决了什么问题？

在复杂的分布式应用中有着许多的依赖，各个依赖都难免会在某个时刻失败，如果应用不隔离各个依赖，降低外部的风险，那容易拖垮整个应用。

举个电商场景中常见的例子，比如订单服务调用了库存服务、商品服务、积分服务、支付服务，系统均正常情况下，订单模块正常运行。

但是当积分服务发生异常时且会阻塞30s时，订单服务就会有部分请求失败，且工作线程阻塞在调用积分服务上。

流量高峰时，问题会更加严重，订单服务的所有请求都会阻塞在调用积分服务上，工作线程全部挂起，导致机器资源耗尽，订单服务也不可用，造成级联影响，整个集群宕机，这种称为雪崩效应。

所以需要一种机制，使得单个服务出现故障时，整个集群可用性不受到影响。Hystrix就是实现这种机制的框架，下面我们分析一下Hystrix整体的工作机制。

二、整体机制

• 【入口】Hystrix的执行入口是HystrixCommand或HystrixObservableCommand对象，通常在Spring应用中会通过注解和AOP来实现对象的构造，以降低对业务代码的侵入性；

• 【缓存】HystrixCommand对象实际开始执行后，首先是否开启缓存，若开启缓存且命中，则直接返回；

• 【熔断】若熔断器打开，则执行短路，直接走降级逻辑；若熔断器关闭，继续下一步，进入隔离逻辑。熔断器的状态主要基于窗口期内执行失败率，若失败率过高，则熔断器自动打开；

• 【隔离】用户可配置走线程池隔离或信号量隔离，判断线程池任务已满（或信号量），则进入降级逻辑；否则继续下一步，实际由线程池任务线程执行业务调用；

• 【执行】实际开始执行业务调用，若执行失败或异常，则进入降级逻辑；若执行成功，则正常返回；

• 【超时】通过定时器延时任务检测业务调用执行是否超时，若超时则取消业务执行的线程，进入降级逻辑；若未超时，则正常返回。线程池、信号量两种策略均隔离方式支持超时配置（信号量策略存在缺陷）；

• 【降级】进入降级逻辑后，当业务实现了HystrixCommand.getFallback() 方法，则返回降级处理的数据；当未实现时，则返回异常；

• 【统计】业务调用执行结果成功、失败、超时等均会进入统计模块，通过健康统计结果来决定熔断器打开或关闭。

都说源码里没有秘密，下面我们来分析下核心功能源码，看看Hystrix如何实现整体的工作机制。

三、熔断

家用电路中都有保险丝，保险丝的作用场景是，当电路发生故障或异常时，伴随着电流不断升高，并且升高的电流有可能损坏电路中的某些重要器件或贵重器件，也有可能烧毁电路甚至造成火灾。

若电路中正确地安置了保险丝，那么保险丝就会在电流异常升高到一定程度的时候，自身熔断切断电流，从而起到保护电路安全运行的作用。Hystrix提供的熔断器就有类似功能，应用调用某个服务提供者，当一定时间内请求总数超过配置的阈值，且窗口期内错误率过高，那Hystrix就会对调用请求熔断，后续的请求直接短路，直接进入降级逻辑，执行本地的降级策略。

Hystrix具有自我调节的能力，熔断器打开在一定时间后，会尝试通过一个请求，并根据执行结果调整熔断器状态，让熔断器在closed,open,half-open三种状态之间自动切换。

【HystrixCircuitBreaker】

boolean attemptExecution()：

每次HystrixCommand执行，都要调用这个方法，判断是否可以继续执行，若熔断器状态为打开且超过休眠窗口，更新熔断器状态为half-open；通过CAS原子变更熔断器状态来保证只放过一条业务请求实际调用提供方，并根据执行结果调整状态。

public boolean attemptExecution() { //判断配置是否强制打开熔断器 if (properties.circuitBreakerForceOpen().get()) { return false; } //判断配置是否强制关闭熔断器 if (properties.circuitBreakerForceClosed().get()) { return true; } //判断熔断器开关是否关闭 if (circuitOpened.get() == -1) { return true; } else { //判断请求是否在休眠窗口后 if (isAfterSleepWindow()) { //更新开关为半开，并允许本次请求通过 if (status.compareAndSet(Status.OPEN, Status.HALF_OPEN)) { return true; } else { return false; } } else { //拒绝请求 return false; } }}

【HystrixCircuitBreaker】

void markSuccess()：HystrixCommand执行成功后调用，当熔断器状态为half-open，更新熔断器状态为closed。此种情况为熔断器原本为open，放过单条请求实际调用服务提供者，并且后续执行成功，Hystrix自动调节熔断器为closed。

public void markSuccess() { //更新熔断器开关为关闭 if (status.compareAndSet(Status.HALF_OPEN, Status.CLOSED)) { //重置订阅健康统计 metrics.resetStream(); Subscription previousSubscription = activeSubscription.get(); if (previousSubscription != null) { previousSubscription.unsubscribe(); } Subscription newSubscription = subscribeToStream(); activeSubscription.set(newSubscription); //更新熔断器开关为关闭 circuitOpened.set(-1L); }}

【HystrixCircuitBreaker】

void markNonSuccess()：HystrixCommand执行成功后调用，若熔断器状态为half-open，更新熔断器状态为open。此种情况为熔断器原本为open，放过单条请求实际调用服务提供者，并且后续执行失败，Hystrix继续保持熔断器打开，并把此次请求作为休眠窗口期开始时间。

public void markNonSuccess() { //更新熔断器开关，从半开变为打开 if (status.compareAndSet(Status.HALF_OPEN, Status.OPEN)) { //记录失败时间，作为休眠窗口开始时间 circuitOpened.set(System.currentTimeMillis()); } }

【HystrixCircuitBreaker】

void subscribeToStream()：熔断器订阅健康统计结果，若当前请求数据大于一定值且错误率大于阈值，自动更新熔断器状态为opened，后续请求短路，不再实际调用服务提供者，直接进入降级逻辑。

private Subscription subscribeToStream() { //订阅监控统计信息 return metrics.getHealthCountsStream() .observe() .subscribe(new Subscriber() { @Override public void onCompleted() {} @Override public void onError(Throwable e) {} @Override public void onNext(HealthCounts hc) { // 判断总请求数量是否超过配置阈值，若未超过，则不改变熔断器状态 if (hc.getTotalRequests() < properties.circuitBreakerRequestVolumeThreshold().get()) { } else { //判断请求错误率是否超过配置错误率阈值，若未超过，则不改变熔断器状态；若超过，则错误率过高，更新熔断器状态未打开，拒绝后续请求 if (hc.getErrorPercentage() < properties.circuitBreakerErrorThresholdPercentage().get()) { } else { if (status.compareAndSet(Status.CLOSED, Status.OPEN)) { circuitOpened.set(System.currentTimeMillis()); } } } } });}

四、资源隔离

在货船中，为了防止漏水和火灾的扩散，一般会将货仓进行分割，避免了一个货仓出事导致整艘船沉没的悲剧。同样的，在Hystrix中，也采用了这样的舱壁模式，将系统中的服务提供者隔离起来，一个服务提供者延迟升高或者失败，并不会导致整个系统的失败，同时也能够控制调用这些服务的并发度。如下图，订单服务调用下游积分、库存等服务使用不同的线程池，当积分服务故障时，只会把对应线程池打满，而不会影响到其他服务的调用。Hystrix隔离模式支持线程池和信号量两种方式。

4.1 信号量模式

信号量模式控制单个服务提供者执行并发度，比如单个CommondKey下正在请求数为N，若N小于maxConcurrentRequests，则继续执行；若大于等于maxConcurrentRequests，则直接拒绝，进入降级逻辑。信号量模式使用请求线程本身执行，没有线程上下文切换，开销较小，但超时机制失效。

【AbstractCommand】

ObservableapplyHystrixSemantics(finalAbstractCommand _cmd)：尝试获取信号量，若能获取到，则继续调用服务提供者；若不能获取到，则进入降级策略。

private Observable applyHystrixSemantics(final AbstractCommand _cmd) { executionHook.onStart(_cmd); //判断熔断器是否通过 if (circuitBreaker.attemptExecution()) { //获取信号量 final TryableSemaphore executionSemaphore = getExecutionSemaphore(); final AtomicBoolean semaphoreHasBeenReleased = new AtomicBoolean(false); final Action0 singleSemaphoreRelease = new Action0() { @Override public void call() { if (semaphoreHasBeenReleased.compareAndSet(false, true)) { executionSemaphore.release(); } } }; final Action1 markExceptionThrown = new Action1() { @Override public void call(Throwable t) { eventNotifier.markEvent(HystrixEventType.EXCEPTION_THROWN, commandKey); } }; //尝试获取信号量 if (executionSemaphore.tryAcquire()) { try { //记录业务执行开始时间 executionResult = executionResult.setInvocationStartTime(System.currentTimeMillis()); //继续执行业务 return executeCommandAndObserve(_cmd) .doOnError(markExceptionThrown) .doOnTerminate(singleSemaphoreRelease) .doOnUnsubscribe(singleSemaphoreRelease); } catch (RuntimeException e) { return Observable.error(e); } } else { //信号量拒绝，进入降级逻辑 return handleSemaphoreRejectionViaFallback(); } } else { //熔断器拒绝，直接短路，进入降级逻辑 return handleShortCircuitViaFallback(); }}

【AbstractCommand】

TryableSemaphore getExecutionSemaphore()：

获取信号量实例，若当前隔离模式为信号量，则根据commandKey获取信号量，不存在时初始化并缓存；若当前隔离模式为线程池，则使用默认信号量TryableSemaphoreNoOp.DEFAULT，全部请求可通过。

protected TryableSemaphore getExecutionSemaphore() { //判断隔离模式是否为信号量 if (properties.executionIsolationStrategy().get() == ExecutionIsolationStrategy.SEMAPHORE) { if (executionSemaphoreOverride == null) { //获取信号量 TryableSemaphore _s = executionSemaphorePerCircuit.get(commandKey.name()); if (_s == null) { //初始化信号量并缓存 executionSemaphorePerCircuit.putIfAbsent(commandKey.name(), new TryableSemaphoreActual(properties.executionIsolationSemaphoreMaxConcurrentRequests())); //返回信号量 return executionSemaphorePerCircuit.get(commandKey.name()); } else { return _s; } } else { return executionSemaphoreOverride; } } else { //返回默认信号量，任何请求均可通过 return TryableSemaphoreNoOp.DEFAULT; }}

4.2 线程池模式

线程池模式控制单个服务提供者执行并发度，代码上都会先走获取信号量，只是使用默认信号量，全部请求可通过，然后实际调用线程池逻辑。线程池模式下，比如单个CommondKey下正在请求数为N，若N小于maximumPoolSize，会先从 Hystrix 管理的线程池里面获得一个线程，然后将参数传递给任务线程去执行真正调用，如果并发请求数多于线程池线程个数，就有任务需要进入队列排队，但排队队列也有上限，如果排队队列也满，则进去降级逻辑。线程池模式可以支持异步调用，支持超时调用，存在线程切换，开销大。

【AbstractCommand】ObservableexecuteCommandWithSpecifiedIsolation(final AbstractCommand _cmd)：从线程池中获取线程，并执行，过程中记录线程状态。

private Observable executeCommandWithSpecifiedIsolation(final AbstractCommand _cmd) { //判断是否为线程池隔离模式 if (properties.executionIsolationStrategy().get() == ExecutionIsolationStrategy.THREAD) { return Observable.defer(new Func0>() { @Override public Observable call() { executionResult = executionResult.setExecutionOccurred(); if (!commandState.compareAndSet(CommandState.OBSERVABLE_CHAIN_CREATED, CommandState.USER_CODE_EXECUTED)) { return Observable.error(new IllegalStateException("execution attempted while in state : " commandState.get().name())); } //统计信息 metrics.markCommandStart(commandKey, threadPoolKey, ExecutionIsolationStrategy.THREAD); //判断是否超时，若超时，直接抛出异常 if (isCommandTimedOut.get() == TimedOutStatus.TIMED_OUT) { return Observable.error(new RuntimeException("timed out before executing run()")); } //更新线程状态为已开始 if (threadState.compareAndSet(ThreadState.NOT_USING_THREAD, ThreadState.STARTED)) { HystrixCounters.incrementGlobalConcurrentThreads(); threadPool.markThreadExecution(); endCurrentThreadExecutingCommand = Hystrix.startCurrentThreadExecutingCommand(getCommandKey()); executionResult = executionResult.setExecutedInThread(); //执行hook，若异常，则直接抛出异常 try { executionHook.onThreadStart(_cmd); executionHook.onRunStart(_cmd); executionHook.onExecutionStart(_cmd); return getUserExecutionObservable(_cmd); } catch (Throwable ex) { return Observable.error(ex); } } else { //空返回 return Observable.empty(); } } }).doOnTerminate(new Action0() { @Override public void call() { //结束逻辑，省略 } }).doOnUnsubscribe(new Action0() { @Override public void call() { //取消订阅逻辑，省略 } //从线程池中获取业务执行线程 }).subscribeOn(threadPool.getScheduler(new Func0<Boolean>() { @Override public Boolean call() { //判断是否超时 return properties.executionIsolationThreadInterruptOnTimeout().get()