Hystrix源码分析

返回Observable

判断条件：circuitBreaker.allowRequest()

尝试从请求缓存中获取结果

@HystrixCommand

Future<R> delegate = toObservable().toBlocking().toFuture()

判断条件：executionSemaphore.tryAcquire()

getUserExecutionObservable(_cmd)

EnableCircuitBreaker

GenericCommand

HystrixInvokable invokable = HystrixCommandFactory.getInstance().create(metaHolder)

spring-cloud-netflix-hystrix-2.1.3.RELEASE.jar!/META-INF/spring.factories

@Import

拒绝处理：RejectedExecutionException

run()

handleSemaphoreRejectionViaFallback()

afterCache.doOnTerminate(terminateCommandCleanup)     .doOnUnsubscribe(unsubscribeCommandCleanup) .doOnCompleted(fireOnCompletedHook)

threadPool.getScheduler

根据执行隔离策略不同获取不同的信号量实现 ：Thread 不使用信号量，使用 TryableSemaphoreNoOp Semaphore 使用信号量，使用 TryableSemaphoreActual 

异常时降级逻辑

应用Hystrix语义，断路器，隔离核心方法

缓存Observable

HystrixCommand#execute

applyHystrixSemantics

创建被观察者Observable

Hystrix 执行命令的 Observable 声明关系

executeCommandAndObserve(_cmd)

HystrixCircuitBreakerConfiguration

execution.doOnNext(markEmits)                .doOnCompleted(markOnCompleted)                .onErrorResumeNext(handleFallback)                .doOnEach(setRequestContext)

unsubscribeCommandCleanup

EnableAutoConfiguration

否

Hystrix服务调用的内部逻辑

HystrixAutoConfiguration

线程池初始化

注意： HystrixBadRequestException不会执行降级逻辑，一般是由非法参数或者一些非系统异常引起的

handleFailureViaFallback(e)

terminateCommandCleanup

TryableSemaphore executionSemaphore = getExecutionSemaphore()

handleFallback

toBlocking().toFuture()

executeCommandWithSpecifiedIsolation(_cmd)

获得 执行命令Observable使用applyHystrixSemantics 来创建Observable

执行降级逻辑

((Observable<T>)that).single().subscribe(new Subscriber<T>(){}）

@EnableCircuitBreaker

其他非HystrixBadRequestException

订阅

getFallback()

subscribe

是，执行降级逻辑

注册

handleThreadPoolRejectionViaFallback(e)

HystrixCommand#getExecutionObservable

Observable<R> hystrixObservable =                        Observable.defer(applyHystrixSemantics)                                .map(wrapWithAllOnNextHooks)

线程池隔离（限流）

是否获得信号量 

执行正常逻辑

fireOnCompletedHook

HystrixTimeoutException

methodsAnnotatedWithHystrixCommand

 applyHystrixSemantics(_cmd)

取消订阅时处理者

new

HystrixCommandAspect

wrapWithAllOnNextHooks

rxjava响应式编程

EnableCircuitBreakerImportSelector

是否开启断路器

defer

@Bean

命令执行完成的Hook

是，执行正常逻辑

handleBadRequestByEmittingError(e)

Observable.defer

HystrixCommand#getFallbackObservable

toObservable()

queue().get()

handleShortCircuitViaFallback()

准备观察者

@Pointcut(\"@annotation(com.netflix.hystrix.contrib.javanica.annotation.HystrixCommand)\")AOP横切代理所有HystrixCommand注解标注的方法@Around(\"hystrixCommandAnnotationPointcut() || hystrixCollapserAnnotationPointcut()\")

发射数据(OnNext表示发射数据)时的Hook

命令模式实现了run和getFallback方法

ExecutionType.SYNCHRONOUS同步执行

命令执行结束后的清理者

执行命令Observable

BlockingObservable

超时处理：HystrixTimeoutException

处理拒绝的失败回退逻辑