热门标签
热门文章
- 1Python打包成linux可执行程序_python打包成linux可执行文件
- 2Yolov5-v7.0使用CBAM注意力机制记录
- 3iphone，pad等真机不被xcode识别，解决方法_ios 开发 xcode15经常不能识别手机
- 4【数据结构与算法python】完全二叉树的实现（优先队列和二叉堆）_python 消息队列完全二叉树
- 5分享75个ASP江湖论坛源码，总有一款适合您_互动力量论坛
- 6递归-常规问题详解
- 72025计算机视觉领域顶会新方向！_cvpr2025
- 820位大佬，勾勒出一个中国网络安全江湖_网络安全专家
- 9Stable diffusion 用DeOldify给黑白照片、视频上色_stablediffusion怎么上色
- 10使用RAG-GPT和Ollama搭建智能客服_ollama部署智能客服
当前位置: article > 正文
hystrix源码分析_hystrix 线程池源码分析
作者：爱喝兽奶帝天荒 | 2024-08-14 07:32:55
赞
踩
hystrix 线程池源码分析
最近遇到了一些生产问题涉及到了hystrix，想要了解下底层的原理。hystrix中大量运用了Rxjava的响应式编程，不懂Rxjava,理解起来有点费劲。
基本准备
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <properties> <maven.compiler.source>8</maven.compiler.source> <maven.compiler.target>8</maven.compiler.target> <spring-boot.version>2.3.12.RELEASE</spring-boot.version> <spring-cloud.version>Hoxton.SR11</spring-cloud.version> </properties> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-netflix-hystrix</artifactId> </dependency> <dependencyManagement> <dependencies> <!--导入依赖的pom文件--> <!--顺序靠前，先被引入--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-dependencies</artifactId> <version>${spring-boot.version}</version> <type>pom</type> <scope>provided</scope> </dependency> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-dependencies</artifactId> <version>${spring-cloud.version}</version> <type>pom</type> <scope>import</scope> </dependency> </dependencies> </dependencyManagement> </project>
- 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
启用hystrix
//在主类上添加 @EnableHystrix 我们看下源码 @Target({ElementType.TYPE}) @Retention(RetentionPolicy.RUNTIME) @Documented @Inherited @EnableCircuitBreaker public @interface EnableHystrix { } 最终@EnableHystrix继承了@EnableCircuitBreaker @Target(ElementType.TYPE) @Retention(RetentionPolicy.RUNTIME) @Documented @Inherited @Import(EnableCircuitBreakerImportSelector.class) public @interface EnableCircuitBreaker { }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
按照之前的模式，我们一定是去分析EnableCircuitBreakerImportSelector，至于如何执行@Import这里不做过多说明，之前已经讲过，感兴趣的看下这里：具体代码在ConfigurationClassParser.processImports里的String[] importClassNames = selector.selectImports(currentSourceClass.getMetadata());
通过EnableCircuitBreakerImportSelector导入HystrixCircuitBreakerConfiguration
//优先级不是特别高，毕竟是运行时的切面
@Order(Ordered.LOWEST_PRECEDENCE - 100)
public class EnableCircuitBreakerImportSelector
extends SpringFactoryImportSelector<EnableCircuitBreaker> {
@Override
protected boolean isEnabled() {
//默认是开启的
return getEnvironment().getProperty("spring.cloud.circuit.breaker.enabled",
Boolean.class, Boolean.TRUE);
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
Import导入后，会通过selectImports从配置文件里加载对应的类 org.springframework.cloud.client.circuitbreaker.EnableCircuitBreaker
public abstract class SpringFactoryImportSelector<T>
implements DeferredImportSelector, BeanClassLoaderAware, EnvironmentAware {
//将泛型T解析为this.annotationClass
protected SpringFactoryImportSelector() {
this.annotationClass = (Class<T>) GenericTypeResolver.resolveTypeArgument(this.getClass(), SpringFactoryImportSelector.class);
}
@Override
public String[] selectImports(AnnotationMetadata metadata) {
//通过SpringFactoriesLoader.loadFactoryNames从spring.factories中获取org.springframework.cloud.client.circuitbreaker.EnableCircuitBreaker对应的配置
List<String> factories = new ArrayList<>(new LinkedHashSet<>(SpringFactoriesLoader.loadFactoryNames(this.annotationClass, this.beanClassLoader)));
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
具体到最后对应的是2.2.8.RELEASE/spring-cloud-netflix-hystrix-2.2.8.RELEASE.jar!/META-INF/spring.factories
spring.factories文件的内容
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.springframework.cloud.netflix.hystrix.HystrixAutoConfiguration,\
org.springframework.cloud.netflix.hystrix.HystrixCircuitBreakerAutoConfiguration,\
org.springframework.cloud.netflix.hystrix.ReactiveHystrixCircuitBreakerAutoConfiguration,\
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
在@SpringBootApplication中，对@EnableAutoConfiguration有引用，所以一启动，就会加载HystrixAutoConfiguration这些类
@SpringBootConfiguration
@EnableAutoConfiguration
@ComponentScan(excludeFilters = { @Filter(type = FilterType.CUSTOM, classes = TypeExcludeFilter.class),
@Filter(type = FilterType.CUSTOM, classes = AutoConfigurationExcludeFilter.class) })
public @interface SpringBootApplication {}
- 1
- 2
- 3
- 4
- 5
- 6
HystrixCircuitBreakerConfiguration解析
@Configuration(proxyBeanMethods = false)
public class HystrixCircuitBreakerConfiguration {
//核心是这个切面
@Bean
public HystrixCommandAspect hystrixCommandAspect() {
return new HystrixCommandAspect();
}
....
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
在HystrixCommandAspect中主要是对HystrixCommand和HystrixCollapser这两个注解进行处理
@Aspect public class HystrixCommandAspect { static { //通过静态方法将两个注解的两个工厂实例化 META_HOLDER_FACTORY_MAP = ImmutableMap.<HystrixPointcutType, MetaHolderFactory>builder() .put(HystrixPointcutType.COMMAND, new CommandMetaHolderFactory()) .put(HystrixPointcutType.COLLAPSER, new CollapserMetaHolderFactory()) .build(); } //定义切入点注解HystrixCommand @Pointcut("@annotation(com.netflix.hystrix.contrib.javanica.annotation.HystrixCommand)") public void hystrixCommandAnnotationPointcut() { } //定义切入点注解HystrixCollapser @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("method cannot be annotated with HystrixCommand and HystrixCollapser " + "annotations at the same time"); } //获取MetaHolderFactory不同实现的factory MetaHolderFactory metaHolderFactory = META_HOLDER_FACTORY_MAP.get(HystrixPointcutType.of(method)); //获取目标方法的的元数据，方法签名，参数等 MetaHolder metaHolder = metaHolderFactory.create(joinPoint); /** * 创建处理器CommandCollapser 或 GenericCommand （同步） 或GenericObservableCommand（异步） * GenericCommand里有很多super，最终通过HystrixCommandBuilderFactory.getInstance().create(metaHolder) 构建了一个HystrixCommandBuilder作为GenericCommad的参数 * new GenericCommand 通过super到AbstractHystrixCommand， * AbstractHystrixCommand 通过super到HystrixCommand， * HystrixCommand最终到了AbstractCommand 一路传递 * 一会在AbstractCommand中分析下 */ HystrixInvokable invokable = HystrixCommandFactory.getInstance().create(metaHolder); //根据返回值推断执行类型 ExecutionType executionType = metaHolder.isCollapserAnnotationPresent() ? metaHolder.getCollapserExecutionType() : metaHolder.getExecutionType(); //返回结果 Object result; try { //不使用observable模式 if (!metaHolder.isObservable()) { //execute执行 result = CommandExecutor.execute(invokable, executionType, metaHolder); } else { result = executeObservable(invokable, executionType, metaHolder); } } catch (HystrixBadRequestException e) { throw e.getCause(); } catch (HystrixRuntimeException e) { throw hystrixRuntimeExceptionToThrowable(metaHolder, e); } return result; } //HystrixCommand的时候MetaHolder的创建 private static class CommandMetaHolderFactory extends MetaHolderFactory { @Override public MetaHolder create(Object proxy, Method method, Object obj, Object[] args, final ProceedingJoinPoint joinPoint) { //获取注解HystrixCommand 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)); } //这里没有多少参数，最重要的一个hystrixCommand，你在注解里加了什么 return builder.defaultCommandKey(method.getName()) .hystrixCommand(hystrixCommand) .observableExecutionMode(hystrixCommand.observableExecutionMode()) //执行模式 .executionType(executionType) //执行方式 .observable(ExecutionType.OBSERVABLE == executionType) .build(); } } } //在枚举ExecutionType类里 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
- 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
我们重点分析下同步处理，通过代码我们可以看到HystrixInvokable 是 GenericCommand，我们同步里的看下 CommandExecutor.execute(invokable, executionType, metaHolder)
public class CommandExecutor { public static Object execute(HystrixInvokable invokable, ExecutionType executionType, MetaHolder metaHolder) throws RuntimeException { Validate.notNull(invokable); Validate.notNull(metaHolder); switch (executionType) { case SYNCHRONOUS: { //重点看同步处理这个，先把GenericCommand 转成HystrixExecutable 再执行execute 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); } } }
- 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
我们先看下类关系
通过GenericCommand一层层的往上翻，最终定位到HystrixCommand有个execute()
HystrixCommand
public abstract class HystrixCommand<R> extends AbstractCommand<R> implements HystrixExecutable<R>, HystrixInvokableInfo<R>, HystrixObservable<R> { //同步执行 public R execute() { try { //通过queue().get()来同步执行 return queue().get(); } catch (Exception e) { throw Exceptions.sneakyThrow(decomposeException(e)); } } //异步执行，什么时候get()，由调用者决定，get()的时候会阻塞 public Future<R> queue() { //核心处理，最终定位到了AbstractCommand里的toObservable()里 final Future<R> delegate = toObservable().toBlocking().toFuture(); final Future<R> f = new Future<R>() { ..... @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); } }; //特殊处理了下，已经执行完了，get()也不会阻塞了 if (f.isDone()) { try { f.get(); return f; } catch (Exception e) { ... } } 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
我们看下官方的流程图：（来源地址：https://github.com/Netflix/Hystrix/wiki/How-it-Works）通过代码，我们也可以看到确实到了这里
AbstractCommand
abstract class AbstractCommand<R> implements HystrixInvokableInfo<R>, HystrixObservable<R> { // 一个 new GenericCommand 最终走到了这里，参数都包装到了这里，只是从GenericCommand 来说 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) { //设置group this.commandGroup = initGroupKey(group); //正常为方法名，不设置，就是类 this.commandKey = initCommandKey(key, getClass()); //这里做了优化，将组装好的CommandProperties放入了ConcurrentHashMap缓存了起来（如果是动态生成方法溢出？） this.properties = initCommandProperties(this.commandKey, propertiesStrategy, commandPropertiesDefaults); //初始化线程池的key，如果为null，则用groupKey.name() this.threadPoolKey = initThreadPoolKey(threadPoolKey, this.commandGroup, this.properties.executionIsolationThreadPoolKeyOverride().get()); //初始化指标，以commandKey缓存 this.metrics = initMetrics(metrics, this.commandGroup, this.threadPoolKey, this.commandKey, this.properties); //初始化断路器，也是以commandKey缓存 this.circuitBreaker = initCircuitBreaker(this.properties.circuitBreakerEnabled().get(), circuitBreaker, this.commandGroup, this.commandKey, this.properties, this.metrics); //初始化线程池，传入有就用，没有就创建一个缓存，还是commandKey维度，最终是通过concurrencyStrategy.getThreadPool(threadPoolKey, properties)创建 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; } //核心处理逻辑 public Observable<R> toObservable() { //拿到当前的对象 final AbstractCommand<R> _cmd = this; /** * terminateCommandCleanup 用于中断命令清理的回调函数 * unsubscribeCommandCleanup 取消订阅命令的清理回调函数 * applyHystrixSemantics 创建带执行逻辑的Observable，用于订阅后回调 * wrapWithAllOnNextHooks onNext执行的钩子 * fireOnCompletedHook 完成后触发的钩子 */ 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); } }; //onNext执行后的钩子 final Func1<R, R> wrapWithAllOnNextHooks = new Func1<R, R>() { @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; } } }; //通过defer创建Observable return Observable.defer(new Func0<Observable<R>>() { //创建观察者，观察者的执行，得由subscribe()触发 @Override public Observable<R> call() { //通过cas来判断命令状态，初始为NOT_STARTED，如果进来不是NOT_STARTED说明已经有别线程执行了，就抛异常 if (!commandState.compareAndSet(CommandState.NOT_STARTED, CommandState.OBSERVABLE_CHAIN_CREATED)) { IllegalStateException ex = new IllegalStateException("This instance can only be executed once. Please instantiate a new instance."); throw new HystrixRuntimeException(FailureType.BAD_REQUEST_EXCEPTION, _cmd.getClass(), getLogMessagePrefix() + " command executed multiple times - this is not permitted.", ex, null); } //命令开始时间 commandStartTimestamp = System.currentTimeMillis(); if (properties.requestLogEnabled().get()) { // 不管发生什么都要用日志记录下来 if (currentRequestLog != null) { currentRequestLog.addExecutedCommand(_cmd); } } //是否允许请求缓存， final boolean requestCacheEnabled = isRequestCachingEnabled(); //cacheKey是需要设置@CacheKey 在HystrixCacheKeyGenerator中会把方法和指定的key的value进行拼接 final String cacheKey = getCacheKey(); //允许请求缓存的情况下，先从缓存中获取，最终缓存用的是ConcurrentHashMap if (requestCacheEnabled) { HystrixCommandResponseFromCache<R> fromCache = (HystrixCommandResponseFromCache<R>) requestCache.get(cacheKey); if (fromCache != null) { isResponseFromCache = true; return handleRequestCacheHitAndEmitValues(fromCache, _cmd); } } //通过defer创建Observable Observable<R> hystrixObservable = Observable.defer(applyHystrixSemantics).map(wrapWithAllOnNextHooks); Observable<R> afterCache; // 开启缓存的情况下，重新放入缓存 if (requestCacheEnabled && cacheKey != null) { // 对hystrixObservable进行包装后放入缓存 HystrixCachedObservable<R> toCache = HystrixCachedObservable.from(hystrixObservable, _cmd); HystrixCommandResponseFromCache<R> fromCache = (HystrixCommandResponseFromCache<R>) requestCache.putIfAbsent(cacheKey, toCache); } else { afterCache = hystrixObservable; } return afterCache .doOnTerminate(terminateCommandCleanup) //中断命令清理 .doOnUnsubscribe(unsubscribeCommandCleanup) // 不订阅也清理 .doOnCompleted(fireOnCompletedHook);//完成触发的钩子 } }); } private Observable<R> applyHystrixSemantics(final AbstractCommand<R> _cmd) { //开始执行 executionHook.onStart(_cmd); //断路器允许执行，里面有好多逻辑，具体看下面的解说 if (circuitBreaker.allowRequest()) { //信号量处理（前提是你配置的是信号量处理机制，如果使用线程池则使用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<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 { //申请不到信号量就直接执行Fallback return handleSemaphoreRejectionViaFallback(); } } else { //断路器不允许执行，就直接执行Fallback方法 return handleShortCircuitViaFallback(); } } private Observable<R> executeCommandAndObserve(final AbstractCommand<R> _cmd) { //HystrixRequestContext hystrix线程的上下文，可以通过它传递一些内容，防止异步过程中数据丢失 final HystrixRequestContext currentRequestContext = HystrixRequestContext.getContextForCurrentThread(); /** * 又是定义一堆Action和Func * markEmits 用于onNext回调 * markOnCompleted 用于完成后处理，会采样一个SUCCESS事件 * handleFallback,有返回结果 用于处理Fallback, * setRequestContext 设置请求上下文 * markEmits、markOnCompleted、handleFallback都会产生一个事件 */ Observable<R> execution; //执行命令 if (properties.executionTimeoutEnabled().get()) { //配置超时时间的情况下会加一个HystrixObservableTimeoutOperator 这里会发送超时事件 execution = executeCommandWithSpecifiedIsolation(_cmd).lift(new HystrixObservableTimeoutOperator<R>(_cmd)); } else { execution = executeCommandWithSpecifiedIsolation(_cmd); } return execution.doOnNext(markEmits) .doOnCompleted(markOnCompleted) .onErrorResumeNext(handleFallback) .doOnEach(setRequestContext); } 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(); try { executionHook.onThreadStart(_cmd); executionHook.onRunStart(_cmd); executionHook.onExecutionStart(_cmd); return getUserExecutionObservable(_cmd); } catch (Throwable ex) { return Observable.error(ex); } } else { 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); } } }).doOnUnsubscribe(new Action0() { @Override public void call() { //未订阅处理 if (threadState.compareAndSet(ThreadState.STARTED, ThreadState.UNSUBSCRIBED)) { handleThreadEnd(_cmd); } } }).subscribeOn(threadPool.getScheduler(new Func0<Boolean>() { @Override public Boolean call() { return properties.executionIsolationThreadInterruptOnTimeout().get() && _cmd.isCommandTimedOut.get() == TimedOutStatus.TIMED_OUT; } })); } else { //信号量的执行 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.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); } } }); } } }
- 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
- 229
- 230
- 231
- 232
- 233
- 234
- 235
- 236
- 237
- 238
- 239
- 240
- 241
- 242
- 243
- 244
- 245
- 246
- 247
- 248
- 249
- 250
- 251
- 252
- 253
- 254
- 255
- 256
- 257
- 258
- 259
- 260
- 261
- 262
- 263
- 264
- 265
- 266
- 267
- 268
- 269
- 270
- 271
- 272
- 273
- 274
- 275
- 276
- 277
- 278
- 279
- 280
- 281
- 282
- 283
- 284
- 285
- 286
- 287
- 288
- 289
- 290
- 291
- 292
- 293
- 294
- 295
- 296
- 297
断路器
public interface HystrixCircuitBreaker { //初始化实现 public static class Factory { public static HystrixCircuitBreaker getInstance(HystrixCommandKey key, HystrixCommandGroupKey group, HystrixCommandProperties properties, HystrixCommandMetrics metrics) { // 缓存中有，就直接返回 HystrixCircuitBreaker previouslyCached = circuitBreakersByCommand.get(key.name()); if (previouslyCached != null) { return previouslyCached; } // 第一次进来，没有就初始化并缓存到map里 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 { return cbForCommand; } } } //断路器实现 static class HystrixCircuitBreakerImpl implements HystrixCircuitBreaker { //hystrix 的配置都在HystrixCommandProperties @Override public boolean allowRequest() { if (properties.circuitBreakerForceOpen().get()) { //强制打开断路器，就直接返回，这个时候就熔断了 return false; } //断路器关闭的情况下，再次验证 if (properties.circuitBreakerForceClosed().get()) { isOpen(); //不管本次什么样的结果，至少我进来的是开着的，设置为关闭以后，我还得继续执行 return true; } return !isOpen() || allowSingleTest(); } @Override public boolean isOpen() { //如果是open的，拦截，直接返回true if (circuitOpen.get()) { return true; } /** * HealthCounts里存储的是一个滑动窗口期间的请求数。 * totalCount 总请求数 包含：数(失败+成功+超时+ threadPoolRejected +信号量拒绝) * errorCount 异常数据 刨除成功就是失败 * errorPercentage 异常百分比= errorCount/totalCount *100; */ HealthCounts health = metrics.getHealthCounts(); //如果总请求数小于配置的值，不拦截，配置值是hystrix.对应的comandKey.circuitBreaker.requestVolumeThreshold 默认值是20 if (health.getTotalRequests() < properties.circuitBreakerRequestVolumeThreshold().get()) { return false; } //如果异常率小于配置的值，也不拦截，配置值是hystrix.对应的comandKey.circuitBreaker.errorThresholdPercentage 默认值是50，也就是有50%异常，就熔断 if (health.getErrorPercentage() < properties.circuitBreakerErrorThresholdPercentage().get()) { return false; } else { // 异常比率较高，就设置断路器为open if (circuitOpen.compareAndSet(false, true)) { // 设置断路器器打开的时间 circuitOpenedOrLastTestedTime.set(System.currentTimeMillis()); return true; } else { //走到这里，说明本线程没有设置成功，别的线程已经打开了断路器，直接返回 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
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 初始化时通过ConcurrentHashMap来缓存HystrixCircuitBreaker来提升性能。
断路器的三种状态：转化关系如下：
断路器初始是Closed状态，如果调用持续出错或者超时，断路器会进入Open状态，熔断请求，后续的一段时间内所有的调用都会触发fallback
-
Open 状态：请求不再进行调用当前服务，内部设置时钟一般为MTTR（平均故障处理时间），当打开时长达到所设时钟则进入半熔断状态
-
Closed 关闭：路器器关闭不会对服务进行熔断部分请求
-
Half Open 半开：根据规则调用当前服务，如果请求成功且符合规则则认为当前服务恢复正常，关闭熔断
看下官方的流程图 在官方：https://github.com/Netflix/Hystrix/wiki/How-it-Works 有图有说明
-
断路器是以commandKey为维度
-
断路器是打开状态，直接熔断
-
在窗口期内，如果请求的量大于设置的值，熔断（默认是10秒超过20个请求）
-
如果异常率大于配置的值，熔断（默认是10秒超过50%的失败）
-
一段时间之后，这个断路器是掰开状态，会允许一个请求进来，如果成功，断路器会关闭
ps: HystrixCommand 的配置属性在类：HystrixCommandProperties
最后上下逻辑图
同时，在hystrix-core的jar包com.netflix.hystrix.metric.consumer下，有很多的HystrixEvent的消费流，这些根据配置执行不同的限流手段，有滑动窗口，有令牌桶，这些流都会在HystrixCommandMetrics实例化的时候启动。如果想了解这一块可以看下HystrixConfigurationStream和HystrixCommandMetrics。后期也会针对这块分析下。
流程是理顺了，但是整起来磕磕绊绊的，里面一层套一层的，想要真正的了解它的底层运作，需要好好的了解下Rxjava,后续研究下再分享。
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/爱喝兽奶帝天荒/article/detail/978706
推荐阅读
相关标签
