微服务组件--限流框架Spring Cloud Hystrix分析

1
2
3
4
5

 org.springframework.cloud
 spring-cloud-starter-netflix-hystrix

1
2
3
4
5
6
7
@SpringBootApplication
//注册到eureka
@EnableEurekaClient
//开启断路器功能
@EnableCircuitBreaker
public class WebApplication {

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
public @interface HystrixCommand {
 // HystrixCommand 命令所属的组的名称：默认注解方法类的名称
    String groupKey() default "";
 // HystrixCommand 命令的key值，默认值为注解方法的名称
    String commandKey() default "";
 // 线程池名称，默认定义为groupKey
    String threadPoolKey() default "";
 // 定义回退方法的名称, 此方法必须和hystrix的执行方法在相同类中
    String fallbackMethod() default "";
 // 配置hystrix命令的参数
    HystrixProperty[] commandProperties() default {};
 // 配置hystrix依赖的线程池的参数
    HystrixProperty[] threadPoolProperties() default {};
 // 如果hystrix方法抛出的异常包括RUNTIME_EXCEPTION，则会被封装HystrixRuntimeException异常。我们也可以通过此方法定义哪些需要忽略的异常
    Class <span style="color: rgba(0, 0, 255, 1)"extends Throwable>[] ignoreExceptions() default {};
 // 定义执行hystrix observable的命令的模式，类型详细见ObservableExecutionMode
    ObservableExecutionMode observableExecutionMode() default ObservableExecutionMode.EAGER;
 // 如果hystrix方法抛出的异常包括RUNTIME_EXCEPTION，则会被封装HystrixRuntimeException异常。此方法定义需要抛出的异常
    HystrixException[] raiseHystrixExceptions() default {};
 // 定义回调方法：但是defaultFallback不能传入参数，返回参数和hystrix的命令兼容
    String defaultFallback() default "";
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
//线程池隔离的设置，线程池隔离与信号量隔离的最大区别在于发送请求的线程，信号量是采用调用方法的线程，而线程池则是用池内的线程去发送请求
@HystrixCommand(
 groupKey="test-provider",
 threadPoolKey="test-provider",
 threadPoolProperties = {
 @HystrixProperty(name = "coreSize", value = "20"),//线程池大小
                @HystrixProperty(name = "maximumSize", value = "30"),//最大线程池大小
                @HystrixProperty(name = "maxQueueSize", value = "20"),//最大队列长度
                @HystrixProperty(name =  "keepAliveTimeMinutes", value = "2")//线程存活时间
        },commandProperties = {
 @HystrixProperty(name = "execution.isolation.strategy",value = "THREAD")
}
//信号量隔离的设置
@HystrixCommand(
 //用来设置降级方法
        fallbackMethod = "myTestFallbackMethod",
 commandProperties = {
 //进行熔断配置
 //条件1，设置在滚动时间窗口中，断路器的最小请求数（没有达到不会熔断）。默认20。
                @HystrixProperty(name = "circuitBreaker.requestVolumeThreshold" ,value = "10"),
 //条件2，设置断路器打开的错误百分比。在滚动时间内，在请求数量超过requestVolumeThreshold的值,且错误请求数的百分比超过这个比例，断路器就为打开状态。
                @HystrixProperty(name = "circuitBreaker.errorThresholdPercentage" ,value = "30"),
 //条件3，设置滚动时间窗的长度，单位毫秒。这个时间窗口就是断路器收集信息的持续时间。断路器在收集指标信息的时会根据这个时间窗口把这个窗口拆分成多个桶，每个桶代表一段时间的指标，默认10000.
                @HystrixProperty(name = "metrics.rollingStats.timeInMilliseconds" ,value = "10000"),
 //设置当断路器打开之后的休眠时间，休眠时间结束后断路器为半开状态，断路器能接受请求，如果请求失败又重新回到打开状态，如果请求成功又回到关闭状态
 //单位是毫秒
                @HystrixProperty(name = "circuitBreaker.sleepWindowInMilliseconds" ,value = "3000"),

 //配置信号量隔离
 //配置信号量的数值
                @HystrixProperty(name = "execution.isolation.semaphore.maxConcurrentRequests",value = "100"),
 //选择策略为信号量隔离
                @HystrixProperty(name = "execution.isolation.strategy", value = "SEMAPHORE"),
 //设置HystrixCommand执行的超时时间，单位毫秒
                @HystrixProperty(name = "execution.isolation.thread.timeoutInMilliseconds", value = "1000000000")
 }
)
public String Test(){
 ....
}

public String myTestFallbackMethod() {
 log.info("========myTestFallbackMethod=========");
 return "myTestFallbackMethod";
}

1
2
3
4
5
6
7
8
9
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
@Import(EnableCircuitBreakerImportSelector.class)
public @interface EnableCircuitBreaker {}
//注解说明：注释以启用断路器实现
//但实际上只是导入了EnableCircuitBreakerImportSelector类

1
2
3
4
5
6
7
8
9
10
//会发现什么都没做，只是将环境变量中的某个值设置为true
@Order(Ordered.LOWEST_PRECEDENCE - 100)
public class EnableCircuitBreakerImportSelector extends SpringFactoryImportSelector {
 @Override
 protected boolean isEnabled() {
 return getEnvironment().getProperty("spring.cloud.circuit.breaker.enabled",
 Boolean.class, Boolean.TRUE);
 }
}

1
2
3
4
5
6
7
8
org.springframework.boot.autoconfigure.EnableAutoConfiguration=
org.springframework.cloud.netflix.hystrix.HystrixAutoConfiguration,
org.springframework.cloud.netflix.hystrix.HystrixCircuitBreakerAutoConfiguration,
org.springframework.cloud.netflix.hystrix.security.HystrixSecurityAutoConfiguration

org.springframework.cloud.client.circuitbreaker.EnableCircuitBreaker=
org.springframework.cloud.netflix.hystrix.HystrixCircuitBreakerConfiguration //该类会比较重要

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
@Configuration(proxyBeanMethods = false)
public class HystrixCircuitBreakerConfiguration {

 //这个看名字就很重要，初始化AOP的拦截
 @Bean
 public HystrixCommandAspect hystrixCommandAspect() {
 return new HystrixCommandAspect();
 }

 @Bean
 public HystrixShutdownHook hystrixShutdownHook() {
 return new HystrixShutdownHook();
 }

 @Bean
 public HasFeatures hystrixFeature() {
 return HasFeatures.namedFeatures(new NamedFeature("Hystrix", HystrixCommandAspect.class));
 }

 private class HystrixShutdownHook implements DisposableBean {

 @Override
 public void destroy() throws Exception {
 // Just call Hystrix to reset thread pool etc.
 Hystrix.reset();
 }

 }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
//先是定义了两个切入点
@Pointcut("@annotation(com.netflix.hystrix.contrib.javanica.annotation.HystrixCommand)")
public void hystrixCommandAnnotationPointcut() {}

@Pointcut("@annotation(com.netflix.hystrix.contrib.javanica.annotation.HystrixCollapser)")
public void hystrixCollapserAnnotationPointcut() {}

//定义切面
@Around("hystrixCommandAnnotationPointcut() || hystrixCollapserAnnotationPointcut()")
public Object methodsAnnotatedWithHystrixCommand(final ProceedingJoinPoint joinPoint) throws Throwable {
 //通过切点获取被拦截的方法
    Method method = getMethodFromTarget(joinPoint);
 Validate.notNull(method, "failed to get method from joinPoint: %s", joinPoint);
 if (method.isAnnotationPresent(HystrixCommand.class) && method.isAnnotationPresent(HystrixCollapser.class)) {
 throw new IllegalStateException(...);
 }
 MetaHolderFactory metaHolderFactory = META_HOLDER_FACTORY_MAP.get(HystrixPointcutType.of(method));
 //metaholder中保存了很多和切点相关的信息，说白了就是解析注解获得上面的信息
    MetaHolder metaHolder = metaHolderFactory.create(joinPoint);
 HystrixInvokable invokable = HystrixCommandFactory.getInstance().create(metaHolder);
 ExecutionType executionType = metaHolder.isCollapserAnnotationPresent() ?
 metaHolder.getCollapserExecutionType() : metaHolder.getExecutionType();

 Object result;
 try {
 if (!metaHolder.isObservable()) {
 result = CommandExecutor.execute(invokable, executionType, metaHolder);
 } else {
 result = executeObservable(invokable, executionType, metaHolder);
 }
 } catch (HystrixBadRequestException e) {...} catch (HystrixRuntimeException e) {...}
 return result;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
public MetaHolder create(final ProceedingJoinPoint joinPoint) {
 Method method = getMethodFromTarget(joinPoint);
 Object obj = joinPoint.getTarget();
 Object[] args = joinPoint.getArgs();
 Object proxy = joinPoint.getThis();
 return create(proxy, method, obj, args, joinPoint);
}

private static class CommandMetaHolderFactory extends MetaHolderFactory {
 @Override
 public MetaHolder create(Object proxy, Method method, Object obj, Object[] args, final ProceedingJoinPoint joinPoint) {
 HystrixCommand hystrixCommand = method.getAnnotation(HystrixCommand.class);
 ExecutionType executionType = ExecutionType.getExecutionType(method.getReturnType());
 MetaHolder.Builder builder = metaHolderBuilder(proxy, method, obj, args, joinPoint);
 if (isCompileWeaving()) {
 builder.ajcMethod(getAjcMethodFromTarget(joinPoint));
 }
 return builder.defaultCommandKey(method.getName())
 .hystrixCommand(hystrixCommand)
 .observableExecutionMode(hystrixCommand.observableExecutionMode())
 .executionType(executionType)
 .observable(ExecutionType.OBSERVABLE == executionType)
 .build();
 }
}

public enum ExecutionType {

 ASYNCHRONOUS,
 SYNCHRONOUS,
 OBSERVABLE;

 //所以根据我们的基本使用可以判断是SYNCHRONOUS，同步模式
    public static ExecutionType getExecutionType(Class type) {
 if (Future.class.isAssignableFrom(type)) {
 return ExecutionType.ASYNCHRONOUS;
 } else if (Observable.class.isAssignableFrom(type)) {
 return ExecutionType.OBSERVABLE;
 } else {
 return ExecutionType.SYNCHRONOUS;
 }
 }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
//分析HystrixInvokable的创建
public HystrixInvokable create(MetaHolder metaHolder) {
 HystrixInvokable executable;　　//判断是不是HystrixCollapser注解
 if (metaHolder.isCollapserAnnotationPresent()) {
 executable = new CommandCollapser(metaHolder);
 } else if (metaHolder.isObservable()) {
 executable = new GenericObservableCommand(HystrixCommandBuilderFactory.getInstance().create(metaHolder));
 } else {
 //主要是这里
        executable = new GenericCommand(HystrixCommandBuilderFactory.getInstance().create(metaHolder));
 }
 return executable;
}

public GenericCommand(HystrixCommandBuilder builder) {
 super(builder);
}

protected AbstractHystrixCommand(HystrixCommandBuilder builder) {
 super(builder.getSetterBuilder().build());
 this.commandActions = builder.getCommandActions();
 this.collapsedRequests = builder.getCollapsedRequests();
 this.cacheResultInvocationContext = builder.getCacheResultInvocationContext();
 this.cacheRemoveInvocationContext = builder.getCacheRemoveInvocationContext();
 this.ignoreExceptions = builder.getIgnoreExceptions();
 this.executionType = builder.getExecutionType();
}

HystrixCommand(HystrixCommandGroupKey group, HystrixCommandKey key, HystrixThreadPoolKey threadPoolKey, HystrixCircuitBreaker circuitBreaker, HystrixThreadPool threadPool,
 HystrixCommandProperties.Setter commandPropertiesDefaults, HystrixThreadPoolProperties.Setter threadPoolPropertiesDefaults,
 HystrixCommandMetrics metrics, TryableSemaphore fallbackSemaphore, TryableSemaphore executionSemaphore,
 HystrixPropertiesStrategy propertiesStrategy, HystrixCommandExecutionHook executionHook) {
 super(group, key, threadPoolKey, circuitBreaker, threadPool, commandPropertiesDefaults, threadPoolPropertiesDefaults, metrics, fallbackSemaphore, executionSemaphore, propertiesStrategy, executionHook);
}

protected AbstractCommand(HystrixCommandGroupKey group, HystrixCommandKey key, HystrixThreadPoolKey threadPoolKey, HystrixCircuitBreaker circuitBreaker, HystrixThreadPool threadPool,
 HystrixCommandProperties.Setter commandPropertiesDefaults, HystrixThreadPoolProperties.Setter threadPoolPropertiesDefaults,
 HystrixCommandMetrics metrics, TryableSemaphore fallbackSemaphore, TryableSemaphore executionSemaphore,
 HystrixPropertiesStrategy propertiesStrategy, HystrixCommandExecutionHook executionHook) {

 this.commandGroup = initGroupKey(group);
 this.commandKey = initCommandKey(key, getClass());
 this.properties = initCommandProperties(this.commandKey, propertiesStrategy, commandPropertiesDefaults);
 this.threadPoolKey = initThreadPoolKey(threadPoolKey, this.commandGroup, this.properties.executionIsolationThreadPoolKeyOverride().get());
 this.metrics = initMetrics(metrics, this.commandGroup, this.threadPoolKey, this.commandKey, this.properties);
 this.circuitBreaker = **initCircuitBreaker**(this.properties.circuitBreakerEnabled().get(), circuitBreaker, this.commandGroup, this.commandKey, this.properties, this.metrics);
 this.threadPool = **initThreadPool**(threadPool, this.threadPoolKey, threadPoolPropertiesDefaults);

 //Strategies from plugins
    this.eventNotifier = HystrixPlugins.getInstance().getEventNotifier();
 this.concurrencyStrategy = HystrixPlugins.getInstance().getConcurrencyStrategy();
 HystrixMetricsPublisherFactory.createOrRetrievePublisherForCommand(this.commandKey, this.commandGroup, this.metrics, this.circuitBreaker, this.properties);
 this.executionHook = initExecutionHook(executionHook);

 this.requestCache = HystrixRequestCache.getInstance(this.commandKey, this.concurrencyStrategy);
 this.currentRequestLog = initRequestLog(this.properties.requestLogEnabled().get(), this.concurrencyStrategy);

 /* fallback semaphore override if applicable */
    this.fallbackSemaphoreOverride = fallbackSemaphore;

 /* execution semaphore override if applicable */
    this.executionSemaphoreOverride = executionSemaphore;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
public static Object execute(HystrixInvokable invokable, ExecutionType executionType, MetaHolder metaHolder) throws RuntimeException {
 Validate.notNull(invokable);
 Validate.notNull(metaHolder);

 switch (executionType) {
 //基于上面构成先分析同步方法
        case SYNCHRONOUS: {
 //调用HystrixCommand
            return castToExecutable(invokable, executionType).execute();
 }
 case ASYNCHRONOUS: {
 HystrixExecutable executable = castToExecutable(invokable, executionType);
 if (metaHolder.hasFallbackMethodCommand()
 && ExecutionType.ASYNCHRONOUS == metaHolder.getFallbackExecutionType()) {
 return new FutureDecorator(executable.queue());
 }
 return executable.queue();
 }
 case OBSERVABLE: {
 HystrixObservable observable = castToObservable(invokable);
 return ObservableExecutionMode.EAGER == metaHolder.getObservableExecutionMode() ? observable.observe() : observable.toObservable();
 }
 default:
 throw new RuntimeException("unsupported execution type: " + executionType);
 }
}

//HystrixCommand类#execute方法
public R execute() {
 try {
 //利用了Future模式
        return queue().get();
 } catch (Exception e) {
 throw Exceptions.sneakyThrow(decomposeException(e));
 }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
//分析queue()方法怎么使用Future模式的
//熟悉线程池的，应该知道线程池有个FutureTask的任务
//通过持有FutureTask句柄可以异步获取返回结果
//本质上就是FutureTask持有
//一个结果存放地址
//线程执行的run方法（执行完后将结果放入固定的存放地址）
//那么现在看下面的逻辑就会十分清晰
public Future queue() {
 
 final Future delegate = toObservable().toBlocking().toFuture();
 
 final Future f = new Future() {

 @Override
 public boolean cancel(boolean mayInterruptIfRunning) {
 if (delegate.isCancelled()) {
 return false;
 }

 if (HystrixCommand.this.getProperties().executionIsolationThreadInterruptOnFutureCancel().get()) {
 interruptOnFutureCancel.compareAndSet(false, mayInterruptIfRunning);
 }

 final boolean res = delegate.cancel(interruptOnFutureCancel.get());

 if (!isExecutionComplete() && interruptOnFutureCancel.get()) {
 final Thread t = executionThread.get();
 if (t != null && !t.equals(Thread.currentThread())) {
 t.interrupt();
 }
 }

 return res;
 }

 @Override
 public boolean isCancelled() {
 return delegate.isCancelled();
 }

 @Override
 public boolean isDone() {
 return delegate.isDone();
 }

 @Override
 public R get() throws InterruptedException, ExecutionException {
 return delegate.get();
 }

 @Override
 public R get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
 return delegate.get(timeout, unit);
 }
 
 };

 /* special handling of error states that throw immediately */
 if (f.isDone()) {
 try {
 f.get();
 return f;
 } catch (Exception e) {
 Throwable t = decomposeException(e);
 if (t instanceof HystrixBadRequestException) {
 return f;
 } else if (t instanceof HystrixRuntimeException) {
 HystrixRuntimeException hre = (HystrixRuntimeException) t;
 switch (hre.getFailureType()) {
 case COMMAND_EXCEPTION:
 case TIMEOUT:
 // we don't throw these types from queue() only from queue().get() as they are execution errors
 return f;
 default:
 // these are errors we throw from queue() as they as rejection type errors
 throw hre;
 }
 } else {
 throw Exceptions.sneakyThrow(t);
 }
 }
 }
 //也就是将产生的Future对象返回
 return f;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
//定义观察者
public Observable toObservable() {
 final AbstractCommand _cmd = this;

 //doOnCompleted handler already did all of the SUCCESS work
 //doOnError handler already did all of the FAILURE/TIMEOUT/REJECTION/BAD_REQUEST work
 //第一个观察者，命令执行结束后的清理者
 final Action0 terminateCommandCleanup = new Action0() {
 @Override
 public void call() {
 if (_cmd.commandState.compareAndSet(CommandState.OBSERVABLE_CHAIN_CREATED, CommandState.TERMINAL)) {
 handleCommandEnd(false); //user code never ran
 } else if (_cmd.commandState.compareAndSet(CommandState.USER_CODE_EXECUTED, CommandState.TERMINAL)) {
 handleCommandEnd(true); //user code did run
 }
 }
 };

 //mark the command as CANCELLED and store the latency (in addition to standard cleanup)
 //第二个观察者，取消订阅时处理者
 final Action0 unsubscribeCommandCleanup = new Action0() {
 @Override
 public void call() {
 if (_cmd.commandState.compareAndSet(CommandState.OBSERVABLE_CHAIN_CREATED, CommandState.UNSUBSCRIBED)) {
 if (!_cmd.executionResult.containsTerminalEvent()) {
 _cmd.eventNotifier.markEvent(HystrixEventType.CANCELLED, _cmd.commandKey);
 try {
 executionHook.onUnsubscribe(_cmd);
 } catch (Throwable hookEx) {...}
 _cmd.executionResultAtTimeOfCancellation = _cmd.executionResult
 .addEvent((int) (System.currentTimeMillis() - _cmd.commandStartTimestamp), HystrixEventType.CANCELLED);
 }
 handleCommandEnd(false); //user code never ran
 } else if (_cmd.commandState.compareAndSet(CommandState.USER_CODE_EXECUTED, CommandState.UNSUBSCRIBED)) {
 if (!_cmd.executionResult.containsTerminalEvent()) {
 _cmd.eventNotifier.markEvent(HystrixEventType.CANCELLED, _cmd.commandKey);
 try {
 executionHook.onUnsubscribe(_cmd);
 } catch (Throwable hookEx) {...}
 _cmd.executionResultAtTimeOfCancellation = _cmd.executionResult
 .addEvent((int) (System.currentTimeMillis() - _cmd.commandStartTimestamp), HystrixEventType.CANCELLED);
 }
 handleCommandEnd(true); //user code did run
 }
 }
 };
 //第三个观察者，重点：Hystrix 核心逻辑: 断路器、隔离
 final Func0> applyHystrixSemantics = new Func0>() {
 @Override
 public Observable call() {
 if (commandState.get().equals(CommandState.UNSUBSCRIBED)) {
 return Observable.never();
 }
 return applyHystrixSemantics(_cmd);
 }
 };
 //第四个观察者，发射数据(OnNext表示发射数据)时的Hook
 final Func1 wrapWithAllOnNextHooks = new Func1() {
 @Override
 public R call(R r) {
 R afterFirstApplication = r;

 try {
 afterFirstApplication = executionHook.onComplete(_cmd, r);
 } catch (Throwable hookEx) {
 logger.warn("Error calling HystrixCommandExecutionHook.onComplete", hookEx);
 }

 try {
 return executionHook.onEmit(_cmd, afterFirstApplication);
 } catch (Throwable hookEx) {
 logger.warn("Error calling HystrixCommandExecutionHook.onEmit", hookEx);
 return afterFirstApplication;
 }
 }
 };
 //第五个观察者，命令执行完成的Hook
 final Action0 fireOnCompletedHook = new Action0() {
 @Override
 public void call() {
 try {
 executionHook.onSuccess(_cmd);
 } catch (Throwable hookEx) {
 logger.warn("Error calling HystrixCommandExecutionHook.onSuccess", hookEx);
 }
 }
 };
　　 //进行包装
 return Observable.defer(new Func0>() {
 @Override
 public Observable call() {
 /* this is a stateful object so can only be used once */
 if (!commandState.compareAndSet(CommandState.NOT_STARTED, CommandState.OBSERVABLE_CHAIN_CREATED)) {
 ..省略抛出异常..
 }

 commandStartTimestamp = System.currentTimeMillis();

 if (properties.requestLogEnabled().get()) {
 // log this command execution regardless of what happened
 if (currentRequestLog != null) {
 currentRequestLog.addExecutedCommand(_cmd);
 }
 }

 final boolean requestCacheEnabled = isRequestCachingEnabled();
 final String cacheKey = getCacheKey();

 /* try from cache first */
 if (requestCacheEnabled) {
 HystrixCommandResponseFromCache fromCache = (HystrixCommandResponseFromCache) requestCache.get(cacheKey);
 if (fromCache != null) {
 isResponseFromCache = true;
 return handleRequestCacheHitAndEmitValues(fromCache, _cmd);
 }
 }
　　　　　　　// 使用上面的Func0：applyHystrixSemantics 来创建Observable
 Observable hystrixObservable =
 Observable.defer(applyHystrixSemantics)
 .map(wrapWithAllOnNextHooks);

 Observable afterCache;

 // 设置缓存逻辑，不太重要
 if (requestCacheEnabled && cacheKey != null) {
 // wrap it for caching
 HystrixCachedObservable toCache = HystrixCachedObservable.from(hystrixObservable, _cmd);
 HystrixCommandResponseFromCache fromCache = (HystrixCommandResponseFromCache) requestCache.putIfAbsent(cacheKey, toCache);
 if (fromCache != null) {
 // another thread beat us so we'll use the cached value instead
 toCache.unsubscribe();
 isResponseFromCache = true;
 return handleRequestCacheHitAndEmitValues(fromCache, _cmd);
 } else {
 // we just created an ObservableCommand so we cast and return it
 afterCache = toCache.toObservable();
 }
 } else {
 afterCache = hystrixObservable;
 }
　　　　　　　//链式
 return afterCache
 .doOnTerminate(terminateCommandCleanup) // perform cleanup once (either on normal terminal state (this line), or unsubscribe (next line))
 .doOnUnsubscribe(unsubscribeCommandCleanup) // perform cleanup once
 .doOnCompleted(fireOnCompletedHook);
 }
 });
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
private Observable applyHystrixSemantics(final AbstractCommand _cmd) {
 //执行命令开始执行的钩子方法 可能有人会问 前面绑定了那么多的钩子方法 这里怎么才开始
 //start 因为前面绑定但是并没有执行。 当有订阅者订阅 这里才是开始执行的代码逻辑
 executionHook.onStart(_cmd);

 //判断断路器是否开启
 if (circuitBreaker.allowRequest()) {
 //如果是信号量隔离 返回TryableSemaphoreActual 根据设置的并发量来判断是否能执行，如果不能执行，进入fallback。
 //如果是线程池隔离 返回TryableSemaphoreNoOp 直接返回true没有任何操作
 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 {
 /* used to track userThreadExecutionTime */
 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();
 }
}

//TryableSemaphoreActual类#tryAcquire方法
@Override
public boolean tryAcquire() {
 int currentCount = count.incrementAndGet();
 if (currentCount > numberOfPermits.get()) {
 count.decrementAndGet();
 return false;
 } else {
 return true;
 }
}

//TryableSemaphoreNoOp类#tryAcquire方法
@Override
public boolean tryAcquire() {
 return true;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
@Override
public boolean allowRequest() {
 if (properties.circuitBreakerForceOpen().get()) {
 // 属性要求我们强制打开电路，这样我们将允许NO请求
        return false;
 }
 if (properties.circuitBreakerForceClosed().get()) {
 // 我们仍然希望允许isOpen()执行它的计算，因此我们模拟正常的行为
 isOpen();
 // 属性要求我们忽略错误，所以我们将忽略isOpen的结果，只允许所有的流量通过
        return true;
 }
 return !isOpen() || allowSingleTest();
}

@Override
public boolean isOpen() {
 //如果断路器打开立即返回true
    if (circuitOpen.get()) {
 return true;
 }

 // we're closed, so let's see if errors have made us so we should trip the circuit open
    HealthCounts health = metrics.getHealthCounts();

 // check if we are past the statisticalWindowVolumeThreshold
    if (health.getTotalRequests() < properties.circuitBreakerRequestVolumeThreshold().get()) {
 // we are not past the minimum volume threshold for the statisticalWindow so we'll return false immediately and not calculate anything
        return false;
 }

 if (health.getErrorPercentage() < properties.circuitBreakerErrorThresholdPercentage().get()) {
 return false;
 } else {
 // our failure rate is too high, trip the circuit
        if (circuitOpen.compareAndSet(false, true)) {
 // if the previousValue was false then we want to set the currentTime
 circuitOpenedOrLastTestedTime.set(System.currentTimeMillis());
 return true;
 } else {
 // How could previousValue be true? If another thread was going through this code at the same time a race-condition could have
 // caused another thread to set it to true already even though we were in the process of doing the same
 // In this case, we know the circuit is open, so let the other thread set the currentTime and report back that the circuit is open
            return true;
 }
 }
}

public boolean allowSingleTest() {
 long timeCircuitOpenedOrWasLastTested = circuitOpenedOrLastTestedTime.get();
 // 1) 如果断路器是打开的
 // 2) 且已经过了休眠时间，尝试打开
    if (circuitOpen.get() && System.currentTimeMillis() > timeCircuitOpenedOrWasLastTested + properties.circuitBreakerSleepWindowInMilliseconds().get()) {
 //已经过了休眠时间，允许一个请求尝试。
 //如果成功，断路器被关闭。
        if (circuitOpenedOrLastTestedTime.compareAndSet(timeCircuitOpenedOrWasLastTested, System.currentTimeMillis())) {
 //如果这个返回true，意味着我们设置了时间，因此我们将返回true以允许单次尝试
 //如果它返回false，这意味着另一个线程在我们之前运行并允许单次尝试
            return true;
 }
 }
 return false;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
private Observable getFallbackOrThrowException(final AbstractCommand _cmd, final HystrixEventType eventType, final FailureType failureType, final String message, final Exception originalException) {
 final HystrixRequestContext requestContext = HystrixRequestContext.getContextForCurrentThread();
 long latency = System.currentTimeMillis() - executionResult.getStartTimestamp();
 // record the executionResult
 // do this before executing fallback so it can be queried from within getFallback (see See https://github.com/Netflix/Hystrix/pull/144)
 executionResult = executionResult.addEvent((int) latency, eventType);

 if (shouldNotBeWrapped(originalException)){
 /* executionHook for all errors */
 Exception e = wrapWithOnErrorHook(failureType, originalException);
 return Observable.error(e);
 } else if (isUnrecoverable(originalException)) {
 logger.error("Unrecoverable Error for HystrixCommand so will throw HystrixRuntimeException and not apply fallback. ", originalException);

 /* executionHook for all errors */
 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()) {
 /* fallback behavior is permitted so attempt */

 final Action1super R>> setRequestContext = new Action1super R>>() {
 @Override
 public void call(Notification <span style="color: rgba(0, 0, 255, 1)"super R> rNotification) {
 setRequestContextIfNeeded(requestContext);
 }
 };

 final Action1 markFallbackEmit = new Action1() {
 @Override
 public void call(R r) {
 if (shouldOutputOnNextEvents()) {
 executionResult = executionResult.addEvent(HystrixEventType.FALLBACK_EMIT);
 eventNotifier.markEvent(HystrixEventType.FALLBACK_EMIT, commandKey);
 }
 }
 };

 final Action0 markFallbackCompleted = new Action0() {
 @Override
 public void call() {
 long latency = System.currentTimeMillis() - executionResult.getStartTimestamp();
 eventNotifier.markEvent(HystrixEventType.FALLBACK_SUCCESS, commandKey);
 executionResult = executionResult.addEvent((int) latency, HystrixEventType.FALLBACK_SUCCESS);
 }
 };

 final Func1> handleFallbackError = new Func1>() {
 @Override
 public Observable call(Throwable t) {
 Exception e = originalException;
 Exception fe = getExceptionFromThrowable(t);

 if (fe instanceof UnsupportedOperationException) {
 long latency = System.currentTimeMillis() - executionResult.getStartTimestamp();
 logger.debug("No fallback for HystrixCommand. ", fe); // debug only since we're throwing the exception and someone higher will do something with it
 eventNotifier.markEvent(HystrixEventType.FALLBACK_MISSING, commandKey);
 executionResult = executionResult.addEvent((int) latency, HystrixEventType.FALLBACK_MISSING);

 /* executionHook for all errors */
 e = wrapWithOnErrorHook(failureType, e);

 return Observable.error(new HystrixRuntimeException(failureType, _cmd.getClass(), getLogMessagePrefix() + " " + message + " and no fallback available.", e, fe));
 } else {
 long latency = System.currentTimeMillis() - executionResult.getStartTimestamp();
 logger.debug("HystrixCommand execution " + failureType.name() + " and fallback failed.", fe);
 eventNotifier.markEvent(HystrixEventType.FALLBACK_FAILURE, commandKey);
 executionResult = executionResult.addEvent((int) latency, HystrixEventType.FALLBACK_FAILURE);

 /* executionHook for all errors */
 e = wrapWithOnErrorHook(failureType, e);

 return Observable.error(new HystrixRuntimeException(failureType, _cmd.getClass(), getLogMessagePrefix() + " " + message + " and fallback failed.", e, fe));
 }
 }
 };

 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 fallbackExecutionChain;

 //上面那些定义的其实都不会在这里调用，主要是看下面的
 // acquire a permit
 if (fallbackSemaphore.tryAcquire()) {
 try {
 if (isFallbackUserDefined()) {
 executionHook.onFallbackStart(this);
 //HystrixCommand类#getFallbackObservable
 fallbackExecutionChain = getFallbackObservable();
 } else {
 //same logic as above without the hook invocation
 fallbackExecutionChain = getFallbackObservable();
 }
 } catch (Throwable ex) {
 //If hook or user-fallback throws, then use that as the result of the fallback lookup
 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类#getFallbackObservable
@Override
final protected Observable getFallbackObservable() {
 return Observable.defer(new Func0>() {
 @Override
 public Observable call() {
 try {
 //调用GenericCommand类的getFallback方法【子类重新写父类】
 return Observable.just(getFallback());
 } catch (Throwable ex) {
 return Observable.error(ex);
 }
 }
 });
}

//GenericCommand类#getFallback方法
@Override
protected Object getFallback() {
 final CommandAction commandAction = getFallbackAction();
 if (commandAction != null) {
 try {
 return process(new Action() {
 @Override
 Object execute() {
 MetaHolder metaHolder = commandAction.getMetaHolder();
 Object[] args = createArgsForFallback(metaHolder, getExecutionException());
 return commandAction.executeWithArgs(metaHolder.getFallbackExecutionType(), args);
 }
 });
 } catch (Throwable e) {
 LOGGER.error(FallbackErrorMessageBuilder.create()
 .append(commandAction, e).build());
 throw new FallbackInvocationException(unwrapCause(e));
 }
 } else {
 return super.getFallback();
 }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
private Observable executeCommandAndObserve(final AbstractCommand _cmd) {
 final HystrixRequestContext currentRequestContext = HystrixRequestContext.getContextForCurrentThread();
 //主要是来对HystrixCommand和HystrixObservableCommand记录的事件是不同的
 final Action1 markEmits = new Action1() {
 @Override
 public void call(R r) {
 if (shouldOutputOnNextEvents()) {
 executionResult = executionResult.addEvent(HystrixEventType.EMIT);
 eventNotifier.markEvent(HystrixEventType.EMIT, commandKey);
 }
 if (commandIsScalar()) {
 long latency = System.currentTimeMillis() - executionResult.getStartTimestamp();
 eventNotifier.markCommandExecution(getCommandKey(), properties.executionIsolationStrategy().get(), (int) latency, executionResult.getOrderedList());
 eventNotifier.markEvent(HystrixEventType.SUCCESS, commandKey);
 executionResult = executionResult.addEvent((int) latency, HystrixEventType.SUCCESS);
 circuitBreaker.markSuccess();
 }
 }
 };

 final Action0 markOnCompleted = new Action0() {
 @Override
 public void call() {
 if (!commandIsScalar()) {
 long latency = System.currentTimeMillis() - executionResult.getStartTimestamp();
 eventNotifier.markCommandExecution(getCommandKey(), properties.executionIsolationStrategy().get(), (int) latency, executionResult.getOrderedList());
 eventNotifier.markEvent(HystrixEventType.SUCCESS, commandKey);
 executionResult = executionResult.addEvent((int) latency, HystrixEventType.SUCCESS);
 circuitBreaker.markSuccess();
 }
 }
 };

 //执行失败的逻辑定义 
 final Func1> handleFallback = new Func1>() {
 @Override
 public Observable call(Throwable t) {
 Exception e = getExceptionFromThrowable(t);
 executionResult = executionResult.setExecutionException(e);
 if (e instanceof RejectedExecutionException) {
 return handleThreadPoolRejectionViaFallback(e);
 } else if (t instanceof HystrixTimeoutException) {
 return handleTimeoutViaFallback();
 } 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);
 }
 }
 };

 final Action1super R>> setRequestContext = new Action1super R>>() {
 @Override
 public void call(Notification <span style="color: rgba(0, 0, 255, 1)"super R> rNotification) {
 setRequestContextIfNeeded(currentRequestContext);
 }
 };

 //上面定义的都是一些异步调用事件，主体在这里
 Observable execution;
 //如果超时开启 使用HystrixObservableTimeoutOperator来对Observable做超时处理。
 //所以不管是信号量隔离还是线程池隔离都会走该逻辑进行超时控制。
 if (properties.executionTimeoutEnabled().get()) {
 //看名字就知道是特殊的隔离，也就是隔离逻辑所在
 execution = executeCommandWithSpecifiedIsolation(_cmd)
 .lift(new HystrixObservableTimeoutOperator(_cmd));
 } else {
 execution = executeCommandWithSpecifiedIsolation(_cmd);
 }

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

private Observable executeCommandWithSpecifiedIsolation(final AbstractCommand _cmd) {
 if (properties.executionIsolationStrategy().get() == ExecutionIsolationStrategy.THREAD) {
 // mark that we are executing in a thread (even if we end up being rejected we still were a THREAD execution and not SEMAPHORE)
 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) {
 // the command timed out in the wrapping thread so we will return immediately
 // and not increment any of the counters below or other such logic
 return Observable.error(new RuntimeException("timed out before executing run()"));
 }
 if (threadState.compareAndSet(ThreadState.NOT_USING_THREAD, ThreadState.STARTED)) {
 //we have not been unsubscribed, so should proceed
 HystrixCounters.incrementGlobalConcurrentThreads();
 threadPool.markThreadExecution();
 // store the command that is being run
 endCurrentThreadExecutingCommand = Hystrix.startCurrentThreadExecutingCommand(getCommandKey());
 executionResult = executionResult.setExecutedInThread();
 /**
 * If any of these hooks throw an exception, then it appears as if the actual execution threw an error
 */
 try {
 executionHook.onThreadStart(_cmd);
 executionHook.onRunStart(_cmd);
 executionHook.onExecutionStart(_cmd);
 return getUserExecutionObservable(_cmd);
 } catch (Throwable ex) {
 return Observable.error(ex);
 }
 } else {
 //command has already been unsubscribed, so return immediately
 return Observable.error(new RuntimeException("unsubscribed before executing run()"));
 }
 }
 }).doOnTerminate(new Action0() {
 @Override
 public void call() {
 if (threadState.compareAndSet(ThreadState.STARTED, ThreadState.TERMINAL)) {
 handleThreadEnd(_cmd);
 }
 if (threadState.compareAndSet(ThreadState.NOT_USING_THREAD, ThreadState.TERMINAL)) {
 //if it was never started and received terminal, then no need to clean up (I don't think this is possible)
 }
 //if it was unsubscribed, then other cleanup handled it
 }
 }).doOnUnsubscribe(new Action0() {
 @Override
 public void call() {
 if (threadState.compareAndSet(ThreadState.STARTED, ThreadState.UNSUBSCRIBED)) {
 handleThreadEnd(_cmd);
 }
 if (threadState.compareAndSet(ThreadState.NOT_USING_THREAD, ThreadState.UNSUBSCRIBED)) {
 //if it was never started and was cancelled, then no need to clean up
 }
 //if it was terminal, then other cleanup handled it
 }
 }).subscribeOn(threadPool.getScheduler(new Func0() {
 //线程池隔离调用
 @Override
 public Boolean call() {
 return properties.executionIsolationThreadInterruptOnTimeout().get() && _cmd.isCommandTimedOut.get() == TimedOutStatus.TIMED_OUT;
 }
 }));
 } else {
 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.SEMAPHORE);
 // semaphore isolated
 // store the command that is being run
 endCurrentThreadExecutingCommand = Hystrix.startCurrentThreadExecutingCommand(getCommandKey());
 try {
 executionHook.onRunStart(_cmd);
 executionHook.onExecutionStart(_cmd);
 return getUserExecutionObservable(_cmd); //the getUserExecutionObservable method already wraps sync exceptions, so this shouldn't throw
 } catch (Throwable ex) {
 //If the above hooks throw, then use that as the result of the run method
 return Observable.error(ex);
 }
 }
 });
 }
}

private Observable getUserExecutionObservable(final AbstractCommand _cmd) {
 Observable userObservable;

 try {
 userObservable = getExecutionObservable();
 } catch (Throwable ex) {
 // the run() method is a user provided implementation so can throw instead of using Observable.onError
 // so we catch it here and turn it into Observable.error
 userObservable = Observable.error(ex);
 }

 return userObservable
 .lift(new ExecutionHookApplication(_cmd))
 .lift(new DeprecatedOnRunHookApplication(_cmd));
}

@Override
final protected Observable getExecutionObservable() {
 return Observable.defer(new Func0>() {
 @Override
 public Observable call() {
 try {
 //调用GenericCommand类的run方法【子类重新写父类】
 return Observable.just(run());
 } catch (Throwable ex) {
 return Observable.error(ex);
 }
 }
 }).doOnSubscribe(new Action0() {
 @Override
 public void call() {
 // Save thread on which we get subscribed so that we can interrupt it later if needed
 executionThread.set(Thread.currentThread());
 }
 });
}
//GenericCommand类#run方法
@Override
protected Object run() throws Exception {
 LOGGER.debug("execute command: {}", getCommandKey().name());
 return process(new Action() {
 @Override
 Object execute() {
 return getCommandAction().execute(getExecutionType());
 }
 });
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
private static HystrixThreadPool initThreadPool(HystrixThreadPool fromConstructor, HystrixThreadPoolKey threadPoolKey, HystrixThreadPoolProperties.Setter threadPoolPropertiesDefaults) {
 if (fromConstructor == null) {
 // get the default implementation of HystrixThreadPool
 //通过塞入注解信息threadPoolPropertiesDefaults进行初始化
        return HystrixThreadPool.Factory.getInstance(threadPoolKey, threadPoolPropertiesDefaults);
 } else {
 return fromConstructor;
 }
}

static HystrixThreadPool getInstance(HystrixThreadPoolKey threadPoolKey, HystrixThreadPoolProperties.Setter propertiesBuilder) {
 // get the key to use instead of using the object itself so that if people forget to implement equals/hashcode things will still work
    String key = threadPoolKey.name();

 // this should find it for all but the first time
    HystrixThreadPool previouslyCached = threadPools.get(key);
 if (previouslyCached != null) {
 return previouslyCached;
 }

 // if we get here this is the first time so we need to initialize
    synchronized (HystrixThreadPool.class) {
 if (!threadPools.containsKey(key)) {
 threadPools.put(key, new HystrixThreadPoolDefault(threadPoolKey, propertiesBuilder));
 }
 }
 return threadPools.get(key);
}

public HystrixThreadPoolDefault(HystrixThreadPoolKey threadPoolKey, HystrixThreadPoolProperties.Setter propertiesDefaults) {
 this.properties = HystrixPropertiesFactory.getThreadPoolProperties(threadPoolKey, propertiesDefaults);
 HystrixConcurrencyStrategy concurrencyStrategy = HystrixPlugins.getInstance().getConcurrencyStrategy();
 this.queueSize = properties.maxQueueSize().get();

 this.metrics = HystrixThreadPoolMetrics.getInstance(threadPoolKey,
 concurrencyStrategy.getThreadPool(threadPoolKey, properties),
 properties);
 this.threadPool = this.metrics.getThreadPool();
 this.queue = this.threadPool.getQueue();

 /* strategy: HystrixMetricsPublisherThreadPool */
 HystrixMetricsPublisherFactory.createOrRetrievePublisherForThreadPool(threadPoolKey, this.metrics, this.properties);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
private static HystrixCircuitBreaker initCircuitBreaker(boolean enabled, HystrixCircuitBreaker fromConstructor,
 HystrixCommandGroupKey groupKey, HystrixCommandKey commandKey,
 HystrixCommandProperties properties, HystrixCommandMetrics metrics) {
 // 如果启用了熔断器
    if (enabled) {
 // 若commandKey没有对应的CircuitBreaker,则创建
        if (fromConstructor == null) {
 // get the default implementation of HystrixCircuitBreaker
            return HystrixCircuitBreaker.Factory.getInstance(commandKey, groupKey, properties, metrics);
 } else {
 // 如果有则返回现有的
            return fromConstructor;
 }
 } else {
 return new NoOpCircuitBreaker();
 }
}

//circuitBreaker以commandKey为维度，每个commandKey都会有对应的circuitBreaker
public static HystrixCircuitBreaker getInstance(HystrixCommandKey key, HystrixCommandGroupKey group, HystrixCommandProperties properties, HystrixCommandMetrics metrics) {
 // 如果有则返回现有的, key.name()即command的name作为检索条件
    HystrixCircuitBreaker previouslyCached = circuitBreakersByCommand.get(key.name());
 if (previouslyCached != null) {
 return previouslyCached;
 }

 // 如果没有则创建并cache
    HystrixCircuitBreaker cbForCommand = circuitBreakersByCommand.putIfAbsent(key.name(), new HystrixCircuitBreakerImpl(key, group, properties, metrics));
 if (cbForCommand == null) {
 // this means the putIfAbsent step just created a new one so let's retrieve and return it
        return circuitBreakersByCommand.get(key.name());
 } else {
 // this means a race occurred and while attempting to 'put' another one got there before
 // and we instead retrieved it and will now return it
        return cbForCommand;
 }
}

protected HystrixCircuitBreakerImpl(HystrixCommandKey key, HystrixCommandGroupKey commandGroup, HystrixCommandProperties properties, HystrixCommandMetrics metrics) {
 this.properties = properties;
 this.metrics = metrics;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
HystrixCommandMetrics(final HystrixCommandKey key, HystrixCommandGroupKey commandGroup, HystrixThreadPoolKey threadPoolKey, HystrixCommandProperties properties, HystrixEventNotifier eventNotifier) {
 super(null);
 this.key = key;
 this.group = commandGroup;
 this.threadPoolKey = threadPoolKey;
 this.properties = properties;

 healthCountsStream = HealthCountsStream.getInstance(key, properties);
 rollingCommandEventCounterStream = RollingCommandEventCounterStream.getInstance(key, properties);
 cumulativeCommandEventCounterStream = CumulativeCommandEventCounterStream.getInstance(key, properties);

 rollingCommandLatencyDistributionStream = RollingCommandLatencyDistributionStream.getInstance(key, properties);
 rollingCommandUserLatencyDistributionStream = RollingCommandUserLatencyDistributionStream.getInstance(key, properties);
 rollingCommandMaxConcurrencyStream = RollingCommandMaxConcurrencyStream.getInstance(key, properties);
}

public static HealthCountsStream getInstance(HystrixCommandKey commandKey, HystrixCommandProperties properties) {
 //默认是按500ms为一个段
    final int healthCountBucketSizeInMs = properties.metricsHealthSnapshotIntervalInMilliseconds().get();
 if (healthCountBucketSizeInMs == 0) {
 throw new RuntimeException(...);
 }
 //获得桶数：设置的时间/段的时间
    final int numHealthCountBuckets = properties.metricsRollingStatisticalWindowInMilliseconds().get() / healthCountBucketSizeInMs;
 //初始化
    return getInstance(commandKey, numHealthCountBuckets, healthCountBucketSizeInMs);
}

public static HealthCountsStream getInstance(HystrixCommandKey commandKey, int numBuckets, int bucketSizeInMs) {
 HealthCountsStream initialStream = streams.get(commandKey.name());
 if (initialStream != null) {
 return initialStream;
 } else {
 final HealthCountsStream healthStream;
 synchronized (HealthCountsStream.class) {
 HealthCountsStream existingStream = streams.get(commandKey.name());
 if (existingStream == null) {
 //新建逻辑
                HealthCountsStream newStream = new HealthCountsStream(commandKey, numBuckets, bucketSizeInMs,
 HystrixCommandMetrics.appendEventToBucket);

 streams.putIfAbsent(commandKey.name(), newStream);
 healthStream = newStream;
 } else {
 healthStream = existingStream;
 }
 }
 healthStream.startCachingStreamValuesIfUnstarted();
 return healthStream;
 }
}

private HealthCountsStream(final HystrixCommandKey commandKey, final int numBuckets, final int bucketSizeInMs,
 Func2<long[], HystrixCommandCompletion, long[]> reduceCommandCompletion) {
 super(HystrixCommandCompletionStream.getInstance(commandKey), numBuckets, bucketSizeInMs, reduceCommandCompletion, healthCheckAccumulator);
}

public abstract class BucketedRollingCounterStreamextends HystrixEvent, Bucket, Output> extends BucketedCounterStream {
 private Observable sourceStream;
 private final AtomicBoolean isSourceCurrentlySubscribed = new AtomicBoolean(false);

 protected BucketedRollingCounterStream(HystrixEventStream stream, final int numBuckets, int bucketSizeInMs, final Func2 appendRawEventToBucket, final Func2 reduceBucket) {
 //调用父级方法
 super(stream, numBuckets, bucketSizeInMs, appendRawEventToBucket);
 Func1, Observable> reduceWindowToSummary = new Func1, Observable>() {
 @Override
 public Observable call(Observable window) {
 return window.scan(getEmptyOutputValue(), reduceBucket).skip(numBuckets);
 }
 };
 this.sourceStream = bucketedStream //stream broken up into buckets
 .window(numBuckets, 1) //emit overlapping windows of buckets
 .flatMap(reduceWindowToSummary) //convert a window of bucket-summaries into a single summary
 .doOnSubscribe(new Action0() {
 @Override
 public void call() {
 isSourceCurrentlySubscribed.set(true);
 }
 })
 .doOnUnsubscribe(new Action0() {
 @Override
 public void call() {
 isSourceCurrentlySubscribed.set(false);
 }
 })
 .share() //multiple subscribers should get same data
 .onBackpressureDrop(); //if there are slow consumers, data should not buffer
 }

 @Override
 public Observable observe() {
 return sourceStream;
 }

 /* package-private */ boolean isSourceCurrentlySubscribed() {
 return isSourceCurrentlySubscribed.get();
 }
}

public abstract class BucketedCounterStreamextends HystrixEvent, Bucket, Output> {
 protected final int numBuckets;
 protected final Observable bucketedStream;
 protected final AtomicReference subscription = new AtomicReference(null);

 private final Func1, Observable> reduceBucketToSummary;

 private final BehaviorSubject counterSubject = BehaviorSubject.create(getEmptyOutputValue());

 protected BucketedCounterStream(final HystrixEventStream inputEventStream, final int numBuckets, final int bucketSizeInMs,
 final Func2 appendRawEventToBucket) {
 this.numBuckets = numBuckets;
 this.reduceBucketToSummary = new Func1, Observable>() {
 @Override
 public Observable call(Observable eventBucket) {
 return eventBucket.reduce(getEmptyBucketSummary(), appendRawEventToBucket);
 }
 };

 final List emptyEventCountsToStart = new ArrayList();
 for (int i = 0; i < numBuckets; i++) {
 emptyEventCountsToStart.add(getEmptyBucketSummary());
 }

 this.bucketedStream = Observable.defer(new Func0>() {
 @Override
 public Observable call() {
 return inputEventStream
 .observe()
 .window(bucketSizeInMs, TimeUnit.MILLISECONDS) //bucket it by the counter window so we can emit to the next operator in time chunks, not on every OnNext
 .flatMap(reduceBucketToSummary) //for a given bucket, turn it into a long array containing counts of event types
 .startWith(emptyEventCountsToStart); //start it with empty arrays to make consumer logic as generic as possible (windows are always full)
 }
 });
 }

 abstract Bucket getEmptyBucketSummary();

 abstract Output getEmptyOutputValue();

 /**
 * Return the stream of buckets
 * @return stream of buckets
 */
 public abstract Observable observe();

 public void startCachingStreamValuesIfUnstarted() {
 if (subscription.get() == null) {
 //the stream is not yet started
 Subscription candidateSubscription = observe().subscribe(counterSubject);
 if (subscription.compareAndSet(null, candidateSubscription)) {
 //won the race to set the subscription
 } else {
 //lost the race to set the subscription, so we need to cancel this one
 candidateSubscription.unsubscribe();
 }
 }
 }

 /**
 * Synchronous call to retrieve the last calculated bucket without waiting for any emissions
 * @return last calculated bucket
 */
 public Output getLatest() {
 startCachingStreamValuesIfUnstarted();
 if (counterSubject.hasValue()) {
 return counterSubject.getValue();
 } else {
 return getEmptyOutputValue();
 }
 }

 public void unsubscribe() {
 Subscription s = subscription.get();
 if (s != null) {
 s.unsubscribe();
 subscription.compareAndSet(s, null);
 }
 }
}

1
2
MetaHolder