一、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<HealthCounts>() {                @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】Observable<R>applyHystrixSemantics(finalAbstractCommand<R> _cmd):尝试获取信号量,若能获取到,则持续调用服务提供者;若不能获取到,则进入降级策略。

private Observable<R> applyHystrixSemantics(final AbstractCommand<R> _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<Throwable> markExceptionThrown = new Action1<Throwable>() {            @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】Observable<R>executeCommandWithSpecifiedIsolation(final AbstractCommand<R> _cmd):从线程池中获取线程,并执行,过程中记录线程状态。

private Observable<R> executeCommandWithSpecifiedIsolation(final AbstractCommand<R> _cmd) {      //判断是否为线程池隔离模式      if (properties.executionIsolationStrategy().get() == ExecutionIsolationStrategy.THREAD) {          return Observable.defer(new Func0<Observable<R>>() {              @Override              public Observable<R> 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() && _cmd.isCommandTimedOut.get() == TimedOutStatus.TIMED_OUT;              }          }));      } else {          //信号量模式          //省略      }  }

【HystrixThreadPool】Subscription schedule(final Action0 action):HystrixContextScheduler是Hystrix对rx中Scheduler调度器的重写,次要为了实现在Observable未被订阅时,不执行命令,以及反对在命令执行过程中可能打断运行。在rx中,Scheduler将生成对应的Worker给Observable用于执行命令,由Worker具体负责相干执行线程的调度,ThreadPoolWorker是Hystrix自行实现的Worker,执行调度的外围办法。

public Subscription schedule(final Action0 action) {    //若无订阅,则不执行间接返回    if (subscription.isUnsubscribed()) {        return Subscriptions.unsubscribed();    }    ScheduledAction sa = new ScheduledAction(action);    subscription.add(sa);    sa.addParent(subscription);    //获取线程池    ThreadPoolExecutor executor = (ThreadPoolExecutor) threadPool.getExecutor();    //提交执行工作    FutureTask<?> f = (FutureTask<?>) executor.submit(sa);    sa.add(new FutureCompleterWithConfigurableInterrupt(f, shouldInterruptThread, executor));    return sa;}

五、超时检测

Hystrix超时机制升高了第三方依赖项提早过高对调用方的影响,使申请疾速失败。次要通过提早工作机制实现,包含注册延时工作过程和执行延时工作过程。

当隔离策略为线程池时,主线程订阅执行后果,线程池中工作线程调用提供者服务端,同时会有定时器线程在肯定工夫后检测工作是否实现,若未实现则示意工作超时,抛出超时异样,并且后续工作线程的执行后果也会跳过不再公布;若已实现则示意工作在超时工夫内实现执行实现,定时器检测工作完结。

当隔离策略为信号量时,主线程订阅执行后果并理论调用提供者服务端(没有工作线程),当超出指定工夫,主线程依然会执行完业务调用,而后抛出超时异样。信号量模式下超时配置有肯定缺点,不能取消在执行的调用,并不能限度主线程返回工夫。

【AbstractCommand】Observable<R>executeCommandAndObserve(finalAbstractCommand<R> \_cmd):超时检测入口,执行lift(new HystrixObservableTimeoutOperator<R>(\_cmd))关联超时检测工作。

private Observable<R> executeCommandAndObserve(final AbstractCommand<R> _cmd) {    //省略    Observable<R> execution;    //判断是否开启超时检测    if (properties.executionTimeoutEnabled().get()) {        execution = executeCommandWithSpecifiedIsolation(_cmd)                //减少超时检测操作                .lift(new HystrixObservableTimeoutOperator<R>(_cmd));    } else {        //失常执行        execution = executeCommandWithSpecifiedIsolation(_cmd);    }    return execution.doOnNext(markEmits)            .doOnCompleted(markOnCompleted)            .onErrorResumeNext(handleFallback)            .doOnEach(setRequestContext);}

【HystrixObservableTimeoutOperator】Subscriber<? super R> call(final Subscriber<? super R> child):创立检测工作,并关联提早工作;若检测工作执行时仍未执行实现,则抛出超时异样;若已执行实现或异样,则革除检测工作。

public Subscriber<? super R> call(final Subscriber<? super R> child) {        final CompositeSubscription s = new CompositeSubscription();        child.add(s);        final HystrixRequestContext hystrixRequestContext = HystrixRequestContext.getContextForCurrentThread();        //实列化监听器        TimerListener listener = new TimerListener() {            @Override            public void tick() {                //若工作未执行实现,则更新为超时                if (originalCommand.isCommandTimedOut.compareAndSet(TimedOutStatus.NOT_EXECUTED, TimedOutStatus.TIMED_OUT)) {                    // 上报超时失败                    originalCommand.eventNotifier.markEvent(HystrixEventType.TIMEOUT, originalCommand.commandKey);                    // 勾销订阅                    s.unsubscribe();                    final HystrixContextRunnable timeoutRunnable = new HystrixContextRunnable(originalCommand.concurrencyStrategy, hystrixRequestContext, new Runnable() {                         @Override                        public void run() {                            child.onError(new HystrixTimeoutException());                        }                    });                    //抛出超时异样                    timeoutRunnable.run();                }            }            //超时工夫配置            @Override            public int getIntervalTimeInMilliseconds() {                return originalCommand.properties.executionTimeoutInMilliseconds().get();            }        };        //注册监听器,关联检测工作        final Reference<TimerListener> tl = HystrixTimer.getInstance().addTimerListener(listener);        originalCommand.timeoutTimer.set(tl);        Subscriber<R> parent = new Subscriber<R>() {            @Override            public void onCompleted() {                if (isNotTimedOut()) {                    // 未超时状况下,工作执行实现,革除超时检测工作                    tl.clear();                    child.onCompleted();                }            }            @Override            public void onError(Throwable e) {                if (isNotTimedOut()) {                    // 未超时状况下,工作执行异样,革除超时检测工作                    tl.clear();                    child.onError(e);                }            }            @Override            public void onNext(R v) {                    //未超时状况下,公布执行后果;超时时则间接跳过公布执行后果                if (isNotTimedOut()) {                    child.onNext(v);                }            }            //判断是否超时            private boolean isNotTimedOut() {                return originalCommand.isCommandTimedOut.get() == TimedOutStatus.COMPLETED ||                        originalCommand.isCommandTimedOut.compareAndSet(TimedOutStatus.NOT_EXECUTED, TimedOutStatus.COMPLETED);            }        };        s.add(parent);        return parent;    }}

【HystrixTimer】Reference<TimerListener>addTimerListener(finalTimerListener listener):addTimerListener通过java的定时工作服务scheduleAtFixedRate在提早超时工夫后执行。

public Reference<TimerListener> addTimerListener(final TimerListener listener) {//初始化xianstartThreadIfNeeded();//结构检测工作Runnable r = new Runnable() {

public Reference<TimerListener> addTimerListener(final TimerListener listener) {    //初始化xian    startThreadIfNeeded();    //结构检测工作    Runnable r = new Runnable() {         @Override        public void run() {            try {                listener.tick();            } catch (Exception e) {                logger.error("Failed while ticking TimerListener", e);            }        }    };    //提早执行检测工作    ScheduledFuture<?> f = executor.get().getThreadPool().scheduleAtFixedRate(r, listener.getIntervalTimeInMilliseconds(), listener.getIntervalTimeInMilliseconds(), TimeUnit.MILLISECONDS);    return new TimerReference(listener, f);}

六、降级

Hystrix降级逻辑作为兜底的策略,当呈现业务执行异样、线程池或信号量已满、执行超时等状况时,会进入降级逻辑。降级逻辑中应从内存或动态逻辑获取通用返回,尽量不依赖依赖网络调用,如果未实现降级办法或降级办法中也出现异常,则业务线程中会引发异样。

【AbstractCommand】Observable<R> getFallbackOrThrowException(finalAbstractCommand<R> _cmd, final HystrixEventType eventType, final FailureType failureType, final String message, final Exception originalException):首先判断是否为不可复原异样,若是则不走降级逻辑,间接异样返回;其次判断是否能获取到降级信号量,而后走降级逻辑;当降级逻辑中也产生异样或者没有降级办法实现时,则异样返回。

private Observable<R> getFallbackOrThrowException(final AbstractCommand<R> _cmd, final HystrixEventType eventType, final FailureType failureType, final String message, final Exception originalException) {    final HystrixRequestContext requestContext = HystrixRequestContext.getContextForCurrentThread();    long latency = System.currentTimeMillis() - executionResult.getStartTimestamp();    executionResult = executionResult.addEvent((int) latency, eventType);    //判断是否为不可复原异样,如栈溢出、OOM等    if (isUnrecoverable(originalException)) {        logger.error("Unrecoverable Error for HystrixCommand so will throw HystrixRuntimeException and not apply fallback. ", originalException);        Exception e = wrapWithOnErrorHook(failureType, originalException);        //间接返回异样        return Observable.error(new HystrixRuntimeException(failureType, this.getClass(), getLogMessagePrefix() + " " + message + " and encountered unrecoverable error.", e, null));    } else {        //判断为是否可复原谬误        if (isRecoverableError(originalException)) {            logger.warn("Recovered from java.lang.Error by serving Hystrix fallback", originalException);        }        //判断降级配置是否关上        if (properties.fallbackEnabled().get()) {          /**            * 省略            */            final Func1<Throwable, Observable<R>> handleFallbackError = new Func1<Throwable, Observable<R>>() {                @Override                public Observable<R> call(Throwable t) {                    Exception e = wrapWithOnErrorHook(failureType, originalException);                    Exception fe = getExceptionFromThrowable(t);                     long latency = System.currentTimeMillis() - executionResult.getStartTimestamp();                    Exception toEmit;                    //是否是不反对操作异样,当业务中没有覆写getFallBack办法时,会抛出此异样                    if (fe instanceof UnsupportedOperationException) {                        logger.debug("No fallback for HystrixCommand. ", fe);                        eventNotifier.markEvent(HystrixEventType.FALLBACK_MISSING, commandKey);                        executionResult = executionResult.addEvent((int) latency, HystrixEventType.FALLBACK_MISSING);                        toEmit = new HystrixRuntimeException(failureType, _cmd.getClass(), getLogMessagePrefix() + " " + message + " and no fallback available.", e, fe);                    } else {                        //执行降级逻辑时产生异样                        logger.debug("HystrixCommand execution " + failureType.name() + " and fallback failed.", fe);                        eventNotifier.markEvent(HystrixEventType.FALLBACK_FAILURE, commandKey);                        executionResult = executionResult.addEvent((int) latency, HystrixEventType.FALLBACK_FAILURE);                        toEmit = new HystrixRuntimeException(failureType, _cmd.getClass(), getLogMessagePrefix() + " " + message + " and fallback failed.", e, fe);                    }                    //判断异样是否包装                    if (shouldNotBeWrapped(originalException)) {                        //抛出异样                        return Observable.error(e);                    }                    //抛出异样                    return Observable.error(toEmit);                }            };            //获取降级信号量            final TryableSemaphore fallbackSemaphore = getFallbackSemaphore();            final AtomicBoolean semaphoreHasBeenReleased = new AtomicBoolean(false);            final Action0 singleSemaphoreRelease = new Action0() {                @Override                public void call() {                    if (semaphoreHasBeenReleased.compareAndSet(false, true)) {                        fallbackSemaphore.release();                    }                }            };            Observable<R> fallbackExecutionChain;            // 尝试获取降级信号量            if (fallbackSemaphore.tryAcquire()) {                try {                    //判断是否定义了fallback办法                    if (isFallbackUserDefined()) {                        executionHook.onFallbackStart(this);                        //执行降级逻辑                        fallbackExecutionChain = getFallbackObservable();                    } else {                        //执行降级逻辑                        fallbackExecutionChain = getFallbackObservable();                    }                } catch (Throwable ex) {                    fallbackExecutionChain = Observable.error(ex);                }                return fallbackExecutionChain                        .doOnEach(setRequestContext)                        .lift(new FallbackHookApplication(_cmd))                        .lift(new DeprecatedOnFallbackHookApplication(_cmd))                        .doOnNext(markFallbackEmit)                        .doOnCompleted(markFallbackCompleted)                        .onErrorResumeNext(handleFallbackError)                        .doOnTerminate(singleSemaphoreRelease)                        .doOnUnsubscribe(singleSemaphoreRelease);            } else {                //解决降级信号量回绝异样               return handleFallbackRejectionByEmittingError();            }        } else {            //解决降级配置敞开时异样            return handleFallbackDisabledByEmittingError(originalException, failureType, message);        }    }}

【HystrixCommand】R getFallback():HystrixCommand默认抛出操作不反对异样,须要子类覆写getFalBack办法实现降级逻辑。

protected R getFallback() {    throw new UnsupportedOperationException("No fallback available.");}

七、衰弱统计

Hystrix基于通过滑动窗口的数据统计断定服务失败占比选择性熔断,可能实现疾速失败并走降级逻辑。步骤如下:

  • AbstractCommand执行实现后调⽤ handleCommandEnd⽅法将执行后果HystrixCommandCompletion事件公布到事件流中;
  • 事件流通过 Observable.window()⽅法将事件按工夫分组,并通过 flatMap()⽅法将事件按类型(胜利、失败等)聚合成桶,造成桶流;
  • 再将各个桶使⽤Observable.window()按窗口内桶数量聚合成滑动窗⼝数据;
  • 将滑动窗口数据聚合成数据对象(如衰弱数据流、累计数据等);
  • 熔断器CircuitBreaker初始化时订阅衰弱数据流,依据衰弱状况批改熔断器的开关。

【AbstractCommand】void handleCommandEnd(boolean commandExecutionStarted):在业务执行结束后,会调用handleCommandEnd办法,在此办法中,上报执行后果executionResult,这也是衰弱统计的入口。

private void handleCommandEnd(boolean commandExecutionStarted) {    Reference<TimerListener> tl = timeoutTimer.get();    if (tl != null) {        tl.clear();    }    long userThreadLatency = System.currentTimeMillis() - commandStartTimestamp;    executionResult = executionResult.markUserThreadCompletion((int) userThreadLatency);    //执行后果上报衰弱统计    if (executionResultAtTimeOfCancellation == null) {        metrics.markCommandDone(executionResult, commandKey, threadPoolKey, commandExecutionStarted);    } else {        metrics.markCommandDone(executionResultAtTimeOfCancellation, commandKey, threadPoolKey, commandExecutionStarted);    }    if (endCurrentThreadExecutingCommand != null) {        endCurrentThreadExecutingCommand.call();    }}

【BucketedRollingCounterStream】BucketedRollingCounterStream(HystrixEventStream<Event> stream, final int numBuckets, int bucketSizeInMs,final Func2<Bucket, Event, Bucket> appendRawEventToBucket,final Func2<Output, Bucket, Output> re-duceBucket)

衰弱统计类HealthCountsStream的滑动窗口实现次要是在父类BucketedRollingCounterStream,首先父类BucketedCounterStream将事件流解决成桶流,BucketedRollingCounterStream解决成滑动窗口,而后由HealthCountsStream传入的reduceBucket函数解决成衰弱统计信息.

protected BucketedRollingCounterStream(HystrixEventStream<Event> stream, final int numBuckets, int bucketSizeInMs,                                       final Func2<Bucket, Event, Bucket> appendRawEventToBucket,                                       final Func2<Output, Bucket, Output> reduceBucket) {    //调用父类,数据处理成桶流    super(stream, numBuckets, bucketSizeInMs, appendRawEventToBucket);    //依据传入的reduceBucket函数,解决滑动窗口内数据    Func1<Observable<Bucket>, Observable<Output>> reduceWindowToSummary = new Func1<Observable<Bucket>, Observable<Output>>() {        @Override        public Observable<Output> call(Observable<Bucket> window) {            return window.scan(getEmptyOutputValue(), reduceBucket).skip(numBuckets);        }    };    //对父类桶流数据进行操作    this.sourceStream = bucketedStream    //窗口内桶数量为numBuckets,每次挪动1个桶            .window(numBuckets, 1)            //滑动窗口内数据处理            .flatMap(reduceWindowToSummary)            .doOnSubscribe(new Action0() {                @Override                public void call() {                    isSourceCurrentlySubscribed.set(true);                }            })            .doOnUnsubscribe(new Action0() {                @Override                public void call() {                    isSourceCurrentlySubscribed.set(false);                }            })            .share()            .onBackpressureDrop();}

【HealthCounts】HealthCounts plus(long[] eventTypeCounts):对桶内数据按事件类型累计,生成统计数据HealthCounts;

public HealthCounts plus(long[] eventTypeCounts) {    long updatedTotalCount = totalCount;    long updatedErrorCount = errorCount;    long successCount = eventTypeCounts[HystrixEventType.SUCCESS.ordinal()];    long failureCount = eventTypeCounts[HystrixEventType.FAILURE.ordinal()];    long timeoutCount = eventTypeCounts[HystrixEventType.TIMEOUT.ordinal()];    long threadPoolRejectedCount = eventTypeCounts[HystrixEventType.THREAD_POOL_REJECTED.ordinal()];    long semaphoreRejectedCount = eventTypeCounts[HystrixEventType.SEMAPHORE_REJECTED.ordinal()];    //总数    updatedTotalCount += (successCount + failureCount + timeoutCount + threadPoolRejectedCount + semaphoreRejectedCount);    //失败数    updatedErrorCount += (failureCount + timeoutCount + threadPoolRejectedCount + semaphoreRejectedCount);    return new HealthCounts(updatedTotalCount, updatedErrorCount);}

八、总结

在分布式环境中,不可避免地会有许多服务的依赖项中有的失败。Hystrix作为一个库,可通过增加熔断、隔离、降级等逻辑来帮忙用户管制分布式服务之间的交互,以进步零碎的整体弹性。次要性能如下:

  • 爱护零碎,管制来自拜访第三方依赖项(通常是通过网络)的提早和失败
  • 阻止简单分布式系统中的级联故障
  • 疾速失败并疾速复原
  • 平滑降级
  • 近乎实时的监控,警报和管制

Hystrix应用过程中,有一些要留神的点:

  • 覆写的getFallback()办法,尽量不要有网络依赖。如果有网络依赖,倡议采纳屡次降级,即在getFallback()内实例化 HystrixCommand,并执行Command。getFallback()尽量保障高性能返回,疾速降级。
  • HystrixCommand 倡议采纳的是线程隔离策略。
  • hystrix.threadpool.default.allowMaximumSizeToDivergeFromCoreSize设置为true时,hystrix.threadpool.default.maximumSize才会失效。最大线程数须要依据业务本身状况和性能测试后果来考量,尽量初始时设置小一些,反对动静调整大小,因为它是缩小负载并避免资源在提早产生时被阻塞的次要工具。
  • 信号隔离策略下,执行业务逻辑时,应用的是应用服务的父级线程(如Tomcat容器线程)。所以,肯定要设置好并发量,有网络开销的调用,不倡议应用该策略,容易导致容器线程排队梗塞,从而影响整个应用服务。
  • 另外Hystrix高度依赖RxJava这个响应式函数编程框架,简略理解RxJava的应用形式,有利于了解源码逻辑。

参考文档

Hystrix Github仓库:https://github.com/Netflix/Hystrix