二十一.SpringCloud源码剖析-Hystrix的初始化

深碍√TFBOYSˉ_ 2022-10-15 08:00 141阅读 0赞

## 前言 ##

Hystrix不是停更了吗？你在这写什么？是，Hystrix是停止更新版本了，说不定后面又继续更新了呢？比如阿里的dubbo不也是停更一段时间后又继续更新了么。Hystrix只是停止开发新的版本，并不是完全停止维护，有Bug依然会修复，Hystrix已经是比较稳定的，很多项目依旧在使用它。

再者说Hystrix是SpringCloud 第一代技术标准中的非常重要的一个组件，可以看做是我们学习SpringCloud全家桶的一个必不可少的过程。在实际项目中你当然可以使用Spring Cloud Alibaba 等更先进的技术来作为微服务架构，使用sentinel代替Hystrix，但是如果你只会使用SpringCloudAlibaba,却不懂netflix技术体系，那我觉得你的技术栈是不完整的。即使你不使用Hystrix，那么它的一些思想和原理也是你需要掌握的，你觉得呢？？？

所以本着技术研究的态度，我还是写了这篇文章，Hystrix 源码剖析，对于Hystrix的使用请移步《[熔断器Hystrix][Hystrix]》

## 1.回顾Hystrix使用 ##

为了防止服务之间的调用异常造成的连锁反应，在SpringCloud中提供了Hystrix组件来实现服务调用异常的处理，或对高并发情况下的服务降级处理 。这里简单回顾一下Hystrix的使用：

1.要使用 Hystrix熔断机制处理引入它本身的依赖之外，我们需要在主程序配置类上贴 @EnableCircuitBreaker标签 开启Hystrix功能，如下

@EnableCircuitBreaker
    @EnableEurekaClient
    @SpringBootApplication
    ...
    public class ConsumerApplication {

2.开启Hystrix熔断机制后，对方法进行熔断处理

@Service
    public class HelloService { 
    
        @Autowired
        private RestTemplate restTemplate;
    
        //该注解对该方法创建了熔断器的功能，并指定了fallbackMethod熔断方法
        @HystrixCommand(fallbackMethod = "hiError")
        public String hiService(String name){ 
            //调用接口进行消费
            String result = restTemplate.getForObject("http://PRODUCER/hello?name="+name,String.class);
            return result;
        }
        public String hiError(String name) { 
            return "hi,"+name+"error!";
        }
    }

当hiService方法第调用异常，会触发 fallbackMethod做降级处理，那么我们就沿着我们的使用方式来跟踪一下 @HystrixCommand注解的背后工作原理。

## 2.Hystrix的自动配置 ##

首先我们看一下标签：@EnableCircuitBreaker ，他的作用从名字就能看出就是开启Hystrix ，我们看一下它的源码

/** * Annotation to enable a CircuitBreaker implementation. * http://martinfowler.com/bliki/CircuitBreaker.html * @author Spencer Gibb */
    @Target(ElementType.TYPE)
    @Retention(RetentionPolicy.RUNTIME)
    @Documented
    @Inherited
    @Import(EnableCircuitBreakerImportSelector.class)
    public @interface EnableCircuitBreaker { 
    
    }

@EnableCircuitBreaker标签引入了一个`@Import(EnableCircuitBreakerImportSelector.class)` 类，翻译类的名字就是 ， 开启熔断器的导入选择器 ，导入什么东西呢？看源码

/** * Import a single circuit breaker implementation Configuration * @author Spencer Gibb */
    @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);
    	}
    
    }

选择器中的isEnabled方法用来读取Hytrix配置，判断是否开启熔断支持，默认是true，  
翻译类上的注释 “Import a single circuit breaker implementation Configuration”，其实EnableCircuitBreakerImportSelector的作用就是去导入熔断器的配置

这个就是走的SpringBoot自动配置了，其实Spring中也有类似于JAVA SPI 的加载机制, 即会自动加载 jar包 spring-cloud-netflix-core 中的META-INF/spring.factories 中的Hystrix相关的自动配置类  
注：SPI : 通过将服务的接口与实现分离以实现解耦，提高程序拓展性的机制，达到插拔式的效果 。

![image.png][]  
HystrixCircuitBreakerConfiguration 就是针对于 Hystrix熔断器的配置

/** * @author Spencer Gibb * @author Christian Dupuis * @author Venil Noronha */
    @Configuration
    public class HystrixCircuitBreakerConfiguration { 
    
    	@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));
    	}
    ......

## 3.HystrixCommandAspect Aop切面类 ##

在该配置类中创建了 HystrixCommandAspect ,它其实就是针对于@HystrixCommand的切面类，看一下源码：

/** * AspectJ aspect to process methods which annotated with {@link HystrixCommand} annotation. */
    @Aspect
    public class HystrixCommandAspect { 
    
        private static final Map<HystrixPointcutType, MetaHolderFactory> META_HOLDER_FACTORY_MAP;
    
        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() { 
        }
    
        @Pointcut("@annotation(com.netflix.hystrix.contrib.javanica.annotation.HystrixCollapser)")
        public void hystrixCollapserAnnotationPointcut() { 
        }
    	//针对切点：@hystrixCommand切点的处理
        @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);
    		//判断方法上不能同时存在@HystrixCommand标签和HystrixCollapser标签
            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 metaHolderFactory = META_HOLDER_FACTORY_MAP.get(HystrixPointcutType.of(method));
            //获取到@HystrixCommand注解的元数据
            MetaHolder metaHolder = metaHolderFactory.create(joinPoint);
    		//把方法封装成 HystrixInvokable ,HystrixInvokable就是调用代理
            HystrixInvokable invokable = HystrixCommandFactory.getInstance().create(metaHolder);
            //这里在判断方法上是否使用 @HystrixCollapser ,我们使用的是@HystrixCommand
            //所以这里的执行类型ExecutionType是 SYNCHRONOUS 同步的方式去调用
            ExecutionType executionType = metaHolder.isCollapserAnnotationPresent() ?
                    metaHolder.getCollapserExecutionType() : metaHolder.getExecutionType();
    
            Object result;
            try { 
    		// 通过CommandExecutor来执行方法
                if (!metaHolder.isObservable()) { 
                	//没有isObservable，走这里
                    result = CommandExecutor.execute(invokable, executionType, metaHolder);
                } else { 
             
                    result = executeObservable(invokable, executionType, metaHolder);
                }
            } catch (HystrixBadRequestException e) { 
                throw e.getCause() != null ? e.getCause() : e;
            } catch (HystrixRuntimeException e) { 
                throw hystrixRuntimeExceptionToThrowable(metaHolder, e);
            }
            return result;

HystrixCommandAspect 其实就是对 贴了@HystrixCommand标签的方法使用 Aop机制实现处理 。代码中通过把目标方法封装成 HystrixInvokable对象，通过CommandExecutor工具来执行目标方法。

### 3.1.HystrixCommand创建 ###

HystrixInvokable 是HystrixCommand的父接口，HystrixCommand是hystrix的核心组件，Hystrix的初始化、执行、限流、熔断等都是在这个组件中完成的，具体在run()方法中执行具体业务逻辑，那么他是如何创建的我们看代码`HystrixInvokable invokable = HystrixCommandFactory.getInstance().create(metaHolder);`的create方法

public HystrixInvokable create(MetaHolder metaHolder) { 
            HystrixInvokable executable;
            ...省略代码...
                executable = new GenericCommand(HystrixCommandBuilderFactory.getInstance().create(metaHolder));
            }
            return executable;
        }

其实是new了一个 GenericCommand 对象，很明显他们是实现关系，我们看一下关系图  
![image.png][image.png 1]  
GenericCommand主要是通过AbstractCommand进行初始化,见：com.netflix.hystrix.AbstractCommand\#AbstractCommand

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

### 3.2.熔断器的创建 ###

HystrixCircuitBreaker是处理熔断业务的熔断器，跟一下 com.netflix.hystrix.AbstractCommand\#initCircuitBreaker 方法，看一下熔断器是如何创建的

private static HystrixCircuitBreaker initCircuitBreaker(boolean enabled, HystrixCircuitBreaker fromConstructor,
                                                                HystrixCommandGroupKey groupKey, HystrixCommandKey commandKey,
                                                                HystrixCommandProperties properties, HystrixCommandMetrics metrics) { 
            if (enabled) { 
                if (fromConstructor == null) { 
                    // get the default implementation of HystrixCircuitBreaker
                    return HystrixCircuitBreaker.Factory.getInstance(commandKey, groupKey, properties, metrics);
                } else { 
                    return fromConstructor;
                }
            } else { 
                return new NoOpCircuitBreaker();
            }
        }

*  HystrixCommandKey：一个HystrixCommandKey一个HystrixCommand，服务熔断，监控都会用到这个key
 *  HystrixCommandGroupKey：一组HystrixCommand可以用HystrixCommandGroupKey来配置
 *  HystrixCommandProperties ：读取和设置HystrixCommand配置的配置类
 *  HystrixCommandMetrics ：用于统计HystrixCommand的调用，包括成功、超时、异常、线程池队列/semaphore满、短路器打开等

这里使用 HystrixCircuitBreaker.Factory 来创建HystrixCircuitBreaker 熔断器，跟一下源码：

class Factory { 
            // String is HystrixCommandKey.name() (we can't use HystrixCommandKey directly as we can't guarantee it implements hashcode/equals correctly)
            //存放熔断器的容器ConcurrentHashMap
            private static ConcurrentHashMap<String, HystrixCircuitBreaker> circuitBreakersByCommand = new ConcurrentHashMap<String, HystrixCircuitBreaker>();
            public static HystrixCircuitBreaker getInstance(HystrixCommandKey key, HystrixCommandGroupKey group, HystrixCommandProperties properties, HystrixCommandMetrics metrics) { 
                // this should find it for all but the first time
                HystrixCircuitBreaker previouslyCached = circuitBreakersByCommand.get(key.name());
                if (previouslyCached != null) { 
                    return previouslyCached;
                }
    
                // if we get here this is the first time so we need to initialize
    
                // Create and add to the map ... use putIfAbsent to atomically handle the possible race-condition of
                // 2 threads hitting this point at the same time and let ConcurrentHashMap provide us our thread-safety
                // If 2 threads hit here only one will get added and the other will get a non-null response instead.
                //熔断器使用的是HystrixCircuitBreakerImpl
                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;
                }
            }

HystrixCircuitBreaker熔断器使用了ConcurrentHashMap来存储，具体的实现类是 HystrixCircuitBreakerImpl

### 3.3.线程池的创建 ###

回到com.netflix.hystrix.AbstractCommand\#initThreadPool方法看一下线程池是如何创建的

private static HystrixThreadPool initThreadPool(HystrixThreadPool fromConstructor, HystrixThreadPoolKey threadPoolKey, HystrixThreadPoolProperties.Setter threadPoolPropertiesDefaults) { 
            if (fromConstructor == null) { 
                // get the default implementation of HystrixThreadPool
                return HystrixThreadPool.Factory.getInstance(threadPoolKey, threadPoolPropertiesDefaults);
            } else { 
                return fromConstructor;
            }
        }

*  HystrixThreadPool ： 对线程池进行统一管理，任务提交、队列限制等
 *  HystrixThreadPoolKey：一个key对应着hystrix的一个HystrixThreadPool
 *  HystrixThreadPoolProperties：读取线程池配置

这里使用HystrixThreadPool.Factory来创建HystrixThreadPool ，它其实和熔断器创建很相似：

static class Factory { 
            /* * Use the String from HystrixThreadPoolKey.name() instead of the HystrixThreadPoolKey instance as it's just an interface and we can't ensure the object * we receive implements hashcode/equals correctly and do not want the default hashcode/equals which would create a new threadpool for every object we get even if the name is the same */
            /* package */final static ConcurrentHashMap<String, HystrixThreadPool> threadPools = new ConcurrentHashMap<String, HystrixThreadPool>();
    
            /** * Get the {@link HystrixThreadPool} instance for a given {@link HystrixThreadPoolKey}. * <p> * This is thread-safe and ensures only 1 {@link HystrixThreadPool} per {@link HystrixThreadPoolKey}. * * @return {@link HystrixThreadPool} instance */
            /* package */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);
            }

线程池使用的是HystrixThreadPoolDefault创建实例，然后使用一个ConcurrentHashMap进行存储的。

## 4. HystrixCommand的执行流程 ##

回到 HystrixCommandAspect的methodsAnnotatedWithHystrixCommand方法中，我们看下 CommandExecutor.execute是如何执行的

public class CommandExecutor { 
        public CommandExecutor() { 
        }
    
        public static Object execute(HystrixInvokable invokable, ExecutionType executionType, MetaHolder metaHolder) throws RuntimeException { 
            Validate.notNull(invokable);
            Validate.notNull(metaHolder);
            switch(executionType) { 
    			//异步
            case SYNCHRONOUS:
                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);
            }
        }
    
        private static HystrixExecutable castToExecutable(HystrixInvokable invokable, ExecutionType executionType) { 
            if (invokable instanceof HystrixExecutable) { 
                return (HystrixExecutable)invokable;
            } else { 
                throw new RuntimeException("Command should implement " + HystrixExecutable.class.getCanonicalName() + " interface to execute in: " + executionType + " mode");
            }
        }

这里有两种执行方式 SYNCHRONOUS 异步 ，ASYNCHRONOUS同步 ，我们先看异步： castToExecutable(invokable, executionType).execute(); 这里代码把HystrixInvokable对象转成 HystrixExecutable并调用execute方法执行 ，跟踪execute方法进入HystrixCommand.execute方法中

public R execute() { 
            try { 
                return queue().get();
            } catch (Exception e) { 
                throw Exceptions.sneakyThrow(decomposeException(e));
            }
        }
    --------------
     public Future<R> queue() { 
            /* * The Future returned by Observable.toBlocking().toFuture() does not implement the * interruption of the execution thread when the "mayInterrupt" flag of Future.cancel(boolean) is set to true; * thus, to comply with the contract of Future, we must wrap around it. */
            final Future<R> delegate = toObservable().toBlocking().toFuture();
        	
            final Future<R> f = new Future<R>() { 
    
                @Override
                public boolean cancel(boolean mayInterruptIfRunning) { 
                    if (delegate.isCancelled()) { 
                        return false;
                    }
    
                    if (HystrixCommand.this.getProperties().executionIsolationThreadInterruptOnFutureCancel().get()) { 
                        /* * The only valid transition here is false -> true. If there are two futures, say f1 and f2, created by this command * (which is super-weird, but has never been prohibited), and calls to f1.cancel(true) and to f2.cancel(false) are * issued by different threads, it's unclear about what value would be used by the time mayInterruptOnCancel is checked. * The most consistent way to deal with this scenario is to say that if *any* cancellation is invoked with interruption, * than that interruption request cannot be taken back. */
                        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;
    			}
    ....省略...

在 HystrixCommand.execute方法中 其实是Future 来异步执行，调用过程中会触发 GenericCommand来完成调用，执行完成后调用 Future.get()方法拿到执行结果 。

文章就到这里吧，如果喜欢请给个好评哦!!!

[Hystrix]: https://blog.csdn.net/u014494148/article/details/105055231
[image.png]: /images/20221014/a68aa75022044bc19694f704f1399fc4.png
[image.png 1]: /images/20221014/a0a31855bb73471abe3aca9db7fec072.png