Hystrix最基本的支持高可用的技术:资源隔离+限流。
创建command;
执行这个command;
配置这个command对应的group和线程池。
开始执行command,调用了这个command的execute()方法之后,Hystrix底层的执行流程和步骤以及原理是什么。
步骤一:创建command
一个HystrixCommand或HystrixObservableCommand对象,代表了对某个依赖服务发起的一次请求或者调用。创建的时候,可以在构造函数中传入任何需要的参数。
HystrixCommand主要用于仅仅会返回一个结果的调用。
HystrixObservableCommand主要用于可能会返回多条结果的调用。
HystrixCommand hystrixCommand = new HystrixCommand(arg1,arg2);
HystrixObservableCommand hystrixObservableCommand = new HystrixObservableCommand(arg1,arg2);
步骤二:调用command执行方法
执行command,就可以发起一次对依赖服务的调用。
要执行command,可以在4个方法中选择其中一个:execute()、queue()、observe()、toObservable()。
其中execute()和queue()方法仅仅对HystrixCommand适用。
execute():调用后直接block住,属于同步调用,直到依赖服务返回单条结果,或者抛出异常。
queue():返回一个Future,属于异步调用,后面可以通过Future获取单条结果。
observe():订阅一个Observable对象,Observable代表的是依赖服务返回的对象,获取到一个那个代表结果的Observable对象的拷贝对象。
toObservable():返回一个Observable对象,如果订阅这个对象,就会执行command并且获取返回结果。
K value = hystrixCommand.execute();
Futrue<K> fValue = hystrixCommand.queue();
Observable<K> oValue = hystrixObservableCommand.observe();
Observable<K> toValue = hystrixObservableCommand.toObservable();
execute()实际上会调用queue.get()方法,可以看下Hystrix源码。
public R execute(){
try{
return queue.get();
}catch(Exception e){
throw Exceptions.sneakyThrow(decompseException(e));
}
}
其中queue()方法中,会调用toObservable().toBlocking().toFuture();
final Future<R> delegate = toObservable().toBlocking().toFuture();
也就是说,先通过toObservable()获得Future对象,然后调用Future的get()方法。那么,其实无论是那种方式执行command,最终都是依赖于toObservable()去执行的。
步骤三:检查是否开启缓存
从这一步开始,就进入到Hystrix底层运行原理,如果这个command开启了请求缓存Request Cache,而且这个调用的结果在缓存中存在,那么直接从缓存中返回结果,否则,继续往后的步骤。
步骤四:检查是否开启了断路器
检查这个command对应的依赖服务是否开启了断路器。如果断路器被打开了,那么Hystrix就不会执行这个command,而是直接去执行fallback降级机制,返回降级结果。
步骤五:检查线程池/队列/信号量是否已满
如果这个command线程池和队列已满,或者semaphore信号量已满,那么也不会执行command,而是直接去调用fallback降级机制,同时发送reject信息给断路器统计。
步骤六:执行command
调用HystrixObservableCommand对象的construct()方法,或者HystrixCommand的run()方法来实际执行command。
HystrixCommand.run()返回单条结果,或者抛出异常。
HystrixObservableCommand.construct()返回一个Observable对象,可以获取多条结果。
如果是采用线程池方式,并且HystrixCommand.run()或者HystrixObservableCommand.construct()的执行时间超过timeout时长的话,那么command所在的线程会抛出一个TimeoutException,这时会执行fallback降级机制,不会去管run()或construct()返回的值了。另一种情况,如果command执行出错抛出了其他异常,那么也会走fallback降级。这两种情况,Hystrix都会发送异常事件给断路器统计。
如果没有timeout,也正常执行的话,那么调用线程就会拿到一些调用依赖服务获取到的结果,然后Hystrix也会做一些logging记录和metrix度量统计。
步骤七:断路健康检查
Hystrix会把每一个依赖服务的调用成功、失败、Reject、timeout等事件发送给circuit breaker断路器。断路器就会把这些事件的次数进行统计,根据异常事件发送的比例来决定是否要进行断路(熔断)。如果打开了断路器,那么在接下来一段时间内,会直接断路,返回降级结果。
如果在志宏,断路器尝试执行command,调用没有出错,返回了正常结果,那么Hystrix就会把断路器关闭。
步骤八:调用fallback降级机制
在以下几种情况下,Hystrix会调用fallback降级机制。
断路器处于打开状态
线程池/队列/semaphore满了
command执行超时
run()或construct()抛出异常。
一般在降级机制中,都建议给出一些默认的返回值,比如静态的一些代码逻辑,或者从内存中的缓存中提取一些数据,在这里尽量不要再进行网络请求了。
在降级中,如果一定要进行网络调用的话,也应该将那个调用放在一个HystrixCommand中进行隔离。
HystrixCommand中,实现getFallback()方法,可以提供降级机制。
HystrixObservableCommand中,实现resumeWithFallback()方法,返回一个Observable,但是不会返回任何数据。
不同的command执行方式,其fallback为空或者异常时的返回结果不同。
对于execute(),直接抛出异常
对于queue(),返回一个Future,调用get()时抛出异常。
对于observe(),返回一个Observable对象,但是调用subscribe()方法订阅它时,立即抛出调用者的onError()方法。
对于toObservable(),返回一个Observable对象,但是调用subscribe()方法订阅它时,立即抛出调用者的onError()方法。
不同的执行方式
execute(),获取一个Future.get(),然后拿到单个结果。
queue(),返回一个Future。
observe(),立即订阅Observable,然后启动8大执行步骤,返回一个拷贝的Observable,订阅时立即回调给你结果。
toObservable(),返回一个原始的Observable,必须手动订阅才会去执行8大步骤。
标签:返回,Observable,调用,Hystrix,command,深入,原理,执行
From: https://blog.51cto.com/u_11315052/8171440