当服务有较多外部依赖时,如果其中某个服务的不可用,导致整个集群会受到影响(比如超时,导致大量的请求被阻塞,从而导致外部请求无法进来),这种情况下采用hystrix就很有用了
出于这个目的,了解了下hystrix框架,下面记录下,框架尝新的历程
通过官网和相关博文,可以简单的说一下这个工作机制,大致流程如下
首先是请求过来 -> 判断熔断器是否开 -> 服务调用 -> 异常则走fallback,失败计数+1 -> 结束
下面是主流程图
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(结束)
熔断机制主要提供了两种,一个是基于线程池的隔离方式来做;还有一个则是根据信号量的抢占来做
线程池方式 : 支持异步,支持超时设置,支持限流
信号量方式 : 本线程执行,无异步,无超时,支持限流,消耗更小
基本上有上面这个简单的概念之后,开始进入我们的使用测试流程
<dependency>
<groupId>com.netflix.hystrix</groupId>
<artifactId>hystrix-core</artifactId>
<version>1.5.12</version>
</dependency>
从官方文档来看,支持两种Command方式,一个是基于观察者模式的ObserverCommand, 一个是基本的Command,先用简单的看以下
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方法是核心执行服务调用,如果需要某些服务不统计到熔断的失败率(比如因为调用姿势不对导致服务内部的异常抛上来了,但是服务本身是正常的),这个时候,就需要包装下调用逻辑,将不需要的异常包装到 HystrixBadRequestException
类里
如
@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;
}
}
}
当发生失败时,hystrix会把原生的异常包装到 HystrixRuntimeException
这个类里,所以我们可以在调用的地方如下处理
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 方法内,则有必要获取对应的异常信息
// 获取异常信息
Throwable t = this.getExecutionException();
然后下一步就是需要获取对应的异常原因了,通过FailureType来表明失败的根源
((HystrixRuntimeException) t).getFailureType()
hystrix自己提供了一套监控插件,基本上大家都会有自己的监控统计信息,因此需要对这个数据进行和自定义,目前还没想到可以如何优雅的处理这些统计信息
主要是看了下这个东西可以怎么玩,整个用下来的感觉就是,设计的比较有意思,但是配置参数太多,很多都没有完全摸透
其次就是一些特殊的case(如监控,报警,特殊情况过滤)需要处理时,用起来并不是很顺手,主要问题还是没有理解清楚这个框架的内部工作机制的问题
基于hexo + github pages搭建的个人博客,记录所有学习和工作中的博文,欢迎大家前去逛逛
尽信书则不如,以上内容,纯属一家之言,因本人能力一般,见识有限,如发现bug或者有更好的建议,随时欢迎批评指正