熔断Hystrix使用尝鲜

文章目录
  1. 熔断Hystrix使用尝鲜
    1. I. 原理探究
    2. II. 使用尝鲜
      1. 1. 引入依赖
      2. 2. 简单使用
      3. 3. 实测理解
        1. a. 配置相关
        2. b. 使用相关
        3. c. 如何获取失败的原因
        4. d.如何获取统计信息
      4. 4. 小结
    3. III. 其他
      1. 个人博客: Z+|blog
      2. 声明
      3. 扫描关注

熔断Hystrix使用尝鲜

当服务有较多外部依赖时,如果其中某个服务的不可用,导致整个集群会受到影响(比如超时,导致大量的请求被阻塞,从而导致外部请求无法进来),这种情况下采用hystrix就很有用了

出于这个目的,了解了下hystrix框架,下面记录下,框架尝新的历程

I. 原理探究

通过官网和相关博文,可以简单的说一下这个工作机制,大致流程如下

首先是请求过来 -> 判断熔断器是否开 -> 服务调用 -> 异常则走fallback,失败计数+1 -> 结束

下面是主流程图

流程图

1
2
3
4
5
6
7
8
9
10
graph LR
A(请求)-->B{熔断器是否已开}
B --> | 熔断 | D[fallback逻辑]
B --> | 未熔断 | E[线程池/Semphore]
E --> F{线程池满/无可用信号量}
F --> | yes | D
F --> | no | G{创建线程执行/本线程运行}
G --> | yes | I(结束)
G --> | no | D
D --> I(结束)

熔断机制主要提供了两种,一个是基于线程池的隔离方式来做;还有一个则是根据信号量的抢占来做

线程池方式 : 支持异步,支持超时设置,支持限流

信号量方式 : 本线程执行,无异步,无超时,支持限流,消耗更小

基本上有上面这个简单的概念之后,开始进入我们的使用测试流程

II. 使用尝鲜

1. 引入依赖

1
2
3
4
5
<dependency>
<groupId>com.netflix.hystrix</groupId>
<artifactId>hystrix-core</artifactId>
<version>1.5.12</version>
</dependency>

2. 简单使用

从官方文档来看,支持两种Command方式,一个是基于观察者模式的ObserverCommand, 一个是基本的Command,先用简单的看以下

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
public class HystrixConfigTest extends HystrixCommand<String> {

private final String name;

public HystrixConfigTest(String name, boolean ans) {
// 注意的是同一个任务,
super(Setter.withGroupKey(
// CommandGroup是每个命令最少配置的必选参数,在不指定ThreadPoolKey的情况下,字面值用于对不同依赖的线程池/信号区分
HystrixCommandGroupKey.Factory.asKey("CircuitBreakerTestGroup"))
// 每个CommandKey代表一个依赖抽象,相同的依赖要使用相同的CommandKey名称。依赖隔离的根本就是对相同CommandKey的依赖做隔离.
.andCommandKey(HystrixCommandKey.Factory.asKey("CircuitBreakerTestKey_" + ans))
// 当对同一业务依赖做隔离时使用CommandGroup做区分,但是对同一依赖的不同远程调用如(一个是redis 一个是http),可以使用HystrixThreadPoolKey做隔离区分
.andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("CircuitBreakerTest_" + ans))
.andThreadPoolPropertiesDefaults( // 配置线程池
HystrixThreadPoolProperties.Setter()
.withCoreSize(12) // 配置线程池里的线程数,设置足够多线程,以防未熔断却打满threadpool
)
.andCommandPropertiesDefaults( // 配置熔断器
HystrixCommandProperties.Setter()
.withCircuitBreakerEnabled(true)
.withCircuitBreakerRequestVolumeThreshold(3)
.withCircuitBreakerErrorThresholdPercentage(80)
// .withCircuitBreakerForceOpen(true) // 置为true时,所有请求都将被拒绝,直接到fallback
// .withCircuitBreakerForceClosed(true) // 置为true时,将忽略错误
// .withExecutionIsolationStrategy(HystrixCommandProperties.ExecutionIsolationStrategy.SEMAPHORE) // 信号量隔离
.withExecutionIsolationSemaphoreMaxConcurrentRequests(20)
.withExecutionTimeoutEnabled(true)
.withExecutionTimeoutInMilliseconds(200)
.withCircuitBreakerSleepWindowInMilliseconds(1000) //熔断器打开到关闭的时间窗长度
// .withExecutionTimeoutInMilliseconds(5000)
)
);
this.name = name;
}

@Override
protected String run() throws Exception {
System.out.println("running run():" + name + " thread: " + Thread.currentThread().getName());
int num = Integer.valueOf(name);
if (num % 2 == 0 && num < 10) { // 直接返回
return name;
} else if (num < 40) {
Thread.sleep(300);
return "sleep+"+ name;
} else { // 无限循环模拟超时
return name;
}
}
//
// @Override
// protected String getFallback() {
// Throwable t = this.getExecutionException();
// if(t instanceof HystrixRuntimeException) {
// System.out.println(Thread.currentThread() + " --> " + ((HystrixRuntimeException) t).getFailureType());
// } else if (t instanceof HystrixTimeoutException) {
// System.out.println(t.getCause());
// } else {
// t.printStackTrace();
// }
// System.out.println(Thread.currentThread() + " --> ----------over------------");
// return "CircuitBreaker fallback: " + name;
// }

public static class UnitTest {

@Test
public void testSynchronous() throws IOException, InterruptedException {
for (int i = 0; i < 50; i++) {
if (i == 41) {
Thread.sleep(2000);
}
try {
System.out.println("===========" + new HystrixConfigTest(String.valueOf(i), i % 2 == 0).execute());
} catch (HystrixRuntimeException e) {
System.out.println(i + " : " + e.getFailureType() + " >>>> " + e.getCause() + " <<<<<");
} catch (Exception e) {
System.out.println("run()抛出HystrixBadRequestException时,被捕获到这里" + e.getCause());
}
}

System.out.println("------开始打印现有线程---------");
Map<Thread, StackTraceElement[]> map = Thread.getAllStackTraces();
for (Thread thread : map.keySet()) {
System.out.println("--->name-->" + thread.getName());
}
System.out.println("thread num: " + map.size());

System.in.read();
}
}
}

使用起来还是比较简单的,一般步骤如下:

  • 继承 HsytrixCommand
  • 重载构造方法,内部需要指定各种配置
  • 实现run方法,这个里面主要执行熔断监控的方法

写上面的代码比较简单,但是有几个地方不太好处理

  1. 配置项的具体含义,又是怎么生效的?
  2. 某些异常不进入熔断逻辑怎么办?
  3. 监控数据如何获取?
  4. 如何模拟各种不同的case(超时?服务异常?熔断已开启?线程池满?无可用信号量?半熔断的重试?)

3. 实测理解

根据上面那一段代码的删删改改,貌似理解了以下几个点,不知道对误

a. 配置相关

  • groupKey 用于区分线程池和信号量,即一个group对应一个
  • commandKey 很重要,这个是用于区分业务
    • 简单来讲,group类似提供服务的app,command则对应app提供的service,一个app可以有多个service,这里就是将一个app的所有请求都放在一个线程池(or共享一个信号量)
  • 开启熔断机制,指定触发熔断的最小请求数(10s内),指定打开熔断的条件(失败率)
  • 设置熔断策略(线程池or信号量)
  • 设置重试时间(默认熔断开启后5s,放几个请求进去,看服务是否恢复)
  • 设置线程池大小,设置信号量大小,设置队列大小
  • 设置超时时间,设置允许超时设置

b. 使用相关

run方法是核心执行服务调用,如果需要某些服务不统计到熔断的失败率(比如因为调用姿势不对导致服务内部的异常抛上来了,但是服务本身是正常的),这个时候,就需要包装下调用逻辑,将不需要的异常包装到 HystrixBadRequestException 类里

1
2
3
4
5
6
7
8
9
10
11
12
13
@Override
protected String run() {
try {
return func.apply(route, parameterDescs);
} catch (Exception e) {
if (exceptionExcept(e)) {
// 如果是不关注的异常case, 不进入熔断逻辑
throw new HystrixBadRequestException("unexpected exception!", e);
} else {
throw e;
}
}
}

c. 如何获取失败的原因

当发生失败时,hystrix会把原生的异常包装到 HystrixRuntimeException 这个类里,所以我们可以在调用的地方如下处理

1
2
3
4
5
6
7
try {
System.out.println("===========" + new HystrixConfigTest(String.valueOf(i), i % 2 == 0).execute());
} catch (HystrixRuntimeException e) {
System.out.println(i + " : " + e.getFailureType() + " >>>> " + e.getCause() + " <<<<<");
} catch (Exception e) {
System.out.println("run()抛出HystrixBadRequestException时,被捕获到这里" + e.getCause());
}

当定义了fallback逻辑时,异常则不会抛到具体的调用方,所以在 fallback 方法内,则有必要获取对应的异常信息

1
2
// 获取异常信息
Throwable t = this.getExecutionException();

然后下一步就是需要获取对应的异常原因了,通过FailureType来表明失败的根源

1
((HystrixRuntimeException) t).getFailureType()

d.如何获取统计信息

hystrix自己提供了一套监控插件,基本上公司内都会有自己的监控统计信息,因此需要对这个数据进行和自定义,目前还没看到可以如何优雅的处理这些统计信息

4. 小结

主要是看了下这个东西可以怎么玩,整个用下来的感觉就是,设计的比较有意思,但是配置参数太多,很多都没有完全摸透

其次就是一些特殊的case(如监控,报警,特殊情况过滤)需要处理时,用起来并不是很顺手,主要问题还是没有理解清楚这个框架的内部工作机制的问题

III. 其他

个人博客: Z+|blog

基于hexo + github pages搭建的个人博客,记录所有学习和工作中的博文,欢迎大家前去逛逛

声明

尽信书则不如,已上内容,纯属一家之言,因本人能力一般,见识有限,如发现bug或者有更好的建议,随时欢迎批评指正

扫描关注

QrCode

评论

Your browser is out-of-date!

Update your browser to view this website correctly. Update my browser now

×