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

🚀 优质资源分享 🚀

Hystrix的介绍

【1】Hystrix是springCloud的组件之一，Hystrix 可以让我们在分布式系统中对服务间的调用进行控制加入一些调用延迟或者依赖故障的容错机制。

【2】Hystrix 通过将依赖服务进行资源隔离进而阻止某个依赖服务出现故障时在整个系统所有的依赖服务调用中进行蔓延；【防止服务雪崩】

【3】其核心功能：

1）服务隔离（服务限流）

通过线程池或者信号量判断是否已满，超出容量的请求直接降级，以达到限流的作用。

2）服务熔断

当失败率达到阈值自动触发降级，熔断器触发的快速失败会有助于系统防止崩溃。【可以说熔断是特定条件的降级】

3）服务降级

服务降级是当服务器压力剧增的情况下，根据当前业务情况及流量对一些服务和页面有策略的降级，以此释放服务器资源以保证核心任务的正常运行。

Hystrix的简单使用

【1】引入依赖

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

1
2
3
4

【2】启动类开启hystrix功能

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

【3】注解@HystrixCommand参数分析

@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 [] 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

【4】使用示例

//线程池隔离的设置，线程池隔离与信号量隔离的最大区别在于发送请求的线程，信号量是采用调用方法的线程，而线程池则是用池内的线程去发送请求
@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
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

Hystrix源码总结

1.采用了AOP的方式来对方法进行了增强，

2.采用了大量的RxJava响应式编程，利用了Future+线程池的方法进行了大量的异步

3.涉及到了滑动窗口的设计，来进行统计失败率

Hystrix源码分析

【1】分析注解@EnableCircuitBreaker是如何开启断路器功能

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

【2】深入分析EnableCircuitBreakerImportSelector类做了什么

//会发现什么都没做，只是将环境变量中的某个值设置为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
9

【3】分析SpringBoot自动装配会导入什么

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

【4】分析HystrixCircuitBreakerConfiguration类做了什么

@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

【5】分析HystrixCommandAspect类在做了什么

//先是定义了两个切入点
@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

【5.1】模式分析—分析MetaHolder的构成

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.SYNCHRO
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