spring-cloud-netflix-hystrix 原理源码分析：
本文主要针对 spring-cloud-dependencies Hoxton.SR4版本, spring-cloud-starter-netflix-hystrix 源码的解析。
对于未接触过 hystrix 的小伙伴可以参考 https://www.cnblogs.com/wuzhenzhao/p/9473073.html 进行一些基础知识的了解。
本文主要从以下几个点来分析：
- 手写实现简易版 Hystrix 体验。
- RXJava 基础知识体验。
- Hystrix 源码流程分析。
手写实现简易版 Hystrix 体验：
继上文的博客链接，我们知道了Hystrix 提供了多种方式让我们实现服务降级。我们可以通过注解 @HystrixCommand、或者继承 HystrixCommand 来实现降级，以及一些请求合并等操作。
我们需要知道的是，当我们采用 @HystrixCommand 注解来实现服务降级，在Hystrix 的内部是采用 AOP 的方式进行拦截处理请求的，我们这里就先来实现一下简易版的 Hystrix 来体会一下，主要分为以下步骤
- 定义自己的@HystrixCommand 注解。
- 实现拦截请求的处理逻辑。
- 测试调用。
1.自定义注解 @WuzzHystrixCommand
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface WuzzHystrixCommand {
/**
* 默认超时时间
*
* @return
*/
int timeout() default 1000;
/**
* 回退方法
*
* @return
*/
String fallback() default "";
}
2.编写切面类，实现简易的逻辑处理
@Component
@Aspect
public class WuzzHystrixCommandAspect {
//线程池的处理，基于这个线程池的处理统计可以达到 THREAD 资源限流
ExecutorService executorService = Executors.newFixedThreadPool(10);
//注解切点
@Pointcut(value = "@annotation(com.wuzz.demo.custom.hystrix.WuzzHystrixCommand)")
public void pointCut() {
}
//环绕通知
@Around(value = "pointCut()&&@annotation(hystrixCommand)")
public Object doPointCut(ProceedingJoinPoint joinPoint, WuzzHystrixCommand hystrixCommand) throws InterruptedException, ExecutionException, TimeoutException, NoSuchMethodException, InvocationTargetException, IllegalAccessException {
int timeout = hystrixCommand.timeout();
//前置的判断逻辑
Future future = executorService.submit(() -> {
try {
return joinPoint.proceed(); //执行目标方法
} catch (Throwable throwable) {
throwable.printStackTrace();
}
return null;
});
Object result;
try {// 使用 future 来实现超时
result = future.get(timeout, TimeUnit.MILLISECONDS);
} catch (InterruptedException | ExecutionException | TimeoutException e) {
e.printStackTrace();
future.cancel(true);
// ？
if (StringUtils.isBlank(hystrixCommand.fallback())) {
throw e;
}
//调用fallback
result = invokeFallback(joinPoint, hystrixCommand.fallback());
}
return result;
}
private Object invokeFallback(ProceedingJoinPoint joinPoint, String fallback) throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
//获取被代理的方法的参数和Method
MethodSignature signature = (MethodSignature) joinPoint.getSignature();
Method method = signature.getMethod();
Class<?>[] parameterTypes = method.getParameterTypes();
//得到fallback方法
try {
Method fallbackMethod = joinPoint.getTarget().getClass().getMethod(fallback, parameterTypes);
fallbackMethod.setAccessible(true);
//完成反射调用
return fallbackMethod.invoke(joinPoint.getTarget(), joinPoint.getArgs());
} catch (Exception e) {
e.printStackTrace();
throw e;
}
}
}
3. 编写测试，调用：
@WuzzHystrixCommand(fallback = "customFallback", timeout = 3000)
@GetMapping("/custom/hystrix/test")
public String test() {
Map map = new HashMap<>();
map.put("id", 666);
return restTemplate.getForObject(REST_URL_PREFIX + "/hello?id={id}", String.class, map);
}
public String customFallback() {
return "custom 请求被降级";
}
正常得调用是没有问题的，这个时候我们把服务提供方的服务接口里 sleep 3秒来模仿调用超时，在访问接口：
相信小伙伴们有了一些心得了，只不过Hystrix里面得实现是很复杂的 ，没有我们这么简单。
RXJava 基础知识体验：
上文的博文连接中也讲到了 RXJava的简单例子，这里由于马上我们要去看 Hystrix的源码了，我们这里写一个类似于源码中的例子，这样来帮助我们更容易理解。
public class RxJavaDemo {
// ReactiveX Java 响应式编程框架(android）
// Java stream() java8
//观察者模式
public static void main(String[] args) throws ExecutionException, InterruptedException {
final String[] datas = new String[]{"登录"};
final Action0 onComplated = new Action0() {
@Override
public void call() {
System.out.println("on Complated");
}
};
//老师（被观察者）
Observable<String> observable = Observable.defer(new Func0<Observable<String>>() {
@Override
public Observable<String> call() {
Observable observable1 = Observable.from(datas);
return observable1.doOnCompleted(onComplated);
}
});
//学生(观察者)
Observer observer = new Observer() {
@Override
public void onCompleted() {
System.out.println("Observer: onCompleted");
}
@Override
public void onError(Throwable throwable) {
System.out.println("Observer: onError");
}
@Override
public void onNext(Object o) {
System.out.println("on Next:" + o);
}
};
// observable.subscribe(observer); //建立订阅关系
String s = observable.toBlocking().toFuture().get();//建立订阅关系
System.out.println(s);
}
}
写这个例子的目的主要是想说明，我们可能对于 RXJava 的 API可能不是很熟悉，但是我们一定要知道对于 Observable 实例来说， call 方法才是关键，而 observable.toBlocking().toFuture().get() 是用于获取执行结果的。在 Hystrix的源码中能看到。了解一下，我们直接进入Hystrix的源码
Hystrix 源码流程分析：
需要注意的是，Hystrix用到了RxJava这个框架，它是一个响应式编程框架，在Android里面用得比较多，所以很多同学对它不是很了解。如果不了解的话，看Hystrix的源码就会有点困难。
Hystrix的数据统计是采用的滑动窗口，关于滑动窗口我这里就不深入研究了，又兴趣的同学可以参考我另外一篇博客， Sentinel 限流原理 进行了解，也可以直接访问 滑动窗口在线演示地址。
Hystrix熔断的@HystrixCommand注解，是通过HystrixCommandAspect这个切面来处理的。其中我们关注@Around注解声明的方法，它针对于请求合并，以及降级的注解进行代理。这里我们重点针对HystrixCommand这个注解进行详细分析。
@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();
}
// 熔断降级切点
@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("method cannot be annotated with HystrixCommand and HystrixCollapser " +
"annotations at the same time");
}
MetaHolderFactory metaHolderFactory = META_HOLDER_FACTORY_MAP.get(HystrixPointcutType.of(method));
// 获取元数据，比如调用方法，HystrixProperty注解数据、方法参数等
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) {
throw e.getCause();
} catch (HystrixRuntimeException e) {
throw hystrixRuntimeExceptionToThrowable(metaHolder, e);
}
return result;
}
// ......
}
然后进入 CommandExecutor#execute 方法这个方法主要用来执行命令，从代码中可以看出这里有三个执行类型，分别是同步、异步、以及响应式。其中，响应式又分为Cold Observable（observable.toObservable()） 和 HotObservable（observable.observe()）默认的executionType=SYNCHRONOUS ，同步请求。
- execute():同步执行，返回一个单一的对象结果，发生错误时抛出异常。
- queue():异步执行，返回一个 Future 对象，包含着执行结束后返回的单一结果。
- observe():这个方法返回一个 Observable 对象，它代表操作的多个结果，但是已经被订阅者消费掉了。
- toObservable():这个方法返回一个 Observable 对象，它代表操作的多个结果，需要咱们自己手动订阅并消费掉。
接着调用HystrixCommand.execute()方法，这个方法中，首先调用queue()，这个方法会返回一个future对象。
public R execute() {
try {
return queue().get();
} catch (Exception e) {
throw Exceptions.sneakyThrow(decomposeException(e));
}
}
queue这个方法中，返回了一个Future对象，这个future对象的实现是f，f是以匿名内部类，它是Java.util.concurrent中定一个的一个异步带返回值对象。当调用queue().get()方法时，最终是委派给了delegate.get 方法。
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) {
// ......省略代码
@Override
public R get() throws InterruptedException, ExecutionException {
return delegate.get();
}
@Override
public R get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
return delegate.get(timeout, unit);
}
};
// ...省略代码
return f;
}
在上述代码中，重点来了，构建了一个 java.util.concurrent.Future ，然后调用 get的时候委派给 delegate，而 delegate来自于 toObservable().toBlocking().toFuture(); 这正是我们上面例子里面得代码。所以我们现在的重点应该放在 toObservable() 方法中：
AbstractCommand.toObservable ：通过Observable定义一个被观察者，这个被观察者会被toObservable().toBlocking().toFuture() ，实际上就是返回可获得 run() 抽象方法执行结果的Future 。 run() 方法由子类实现，执行正常的业务逻辑。在下面这段代码中，当存在subscriber时，便会调用Func0#call() 方法，而这个subscriber是在 toBlocking() 中被订阅的。到这还是我们上面的例子里面的代码。该方法主要做了以下几件事：
-
创建一些命令供后续的回调使用
-
调用 isRequestCachingEnabled(); 判断请求结果缓存功能是否开启，如果开启并且命中了缓存，则会以Observable形式返回一个缓存结果
-
创建执行命令的Observable： hystrixObservable，
-
当缓存处于开启状态并且没有命中缓存时，则创建一个“订阅了执行命令的Observable”：HystrixCommandResponseFromCache
-
创建存储到缓存的Observable： HystrixCachedObservable当缓存特性没有开启时，则返回执行命令的Observable。
-
将toCache添加到缓存中，返回获取缓存的Observable：fromCache
-
如果添加失败: fromCache!=null, 则调用 toCache.unsubscribe() 方法，取消HystrixCachedObservable 的订阅
-
如果添加成功，则调用 toCache.toObservable(); 获得缓存Observable
-
-
当缓存特性没有开启时，则返回执行命令的Observable。
public Observable<R> toObservable() {
final AbstractCommand<R> _cmd = this;
//会在Observable结束前触发回调该call方法，无论是正常还是异常终止
//doOnCompleted handler already did all of the SUCCESS work
//doOnError handler already did all of the FAILURE/TIMEOUT/REJECTION/BAD_REQUEST work
final Action0 terminateCommandCleanup = new Action0() {
@Override
public void call() {
if (_cmd.commandState.compareAndSet(CommandState.OBSERVABLE_CHAIN_CREATED, CommandState.TERMINAL)) {
handleCommandEnd(false); //user code never ran
} else if (_cmd.commandState.compareAndSet(CommandState.USER_CODE_EXECUTED, CommandState.TERMINAL)) {
handleCommandEnd(true); //user code did run
}
}
};
//取消订阅时的监听会进行回调该 call方法
//mark the command as CANCELLED and store the latency (in addition to standard cleanup)
final Action0 unsubscribeCommandCleanup = new Action0() {
@Override
public void call() {
if (_cmd.commandState.compareAndSet(CommandState.OBSERVABLE_CHAIN_CREATED, CommandState.UNSUBSCRIBED)) {
if (!_cmd.executionResult.containsTerminalEvent()) {
_cmd.eventNotifier.markEvent(HystrixEventType.CANCELLED, _cmd.commandKey);
try {
executionHook.onUnsubscribe(_cmd);
} catch (Throwable hookEx) {
logger.warn("Error calling HystrixCommandExecutionHook.onUnsubscribe", hookEx);
}
_cmd.executionResultAtTimeOfCancellation = _cmd.executionResult
.addEvent((int) (System.currentTimeMillis() - _cmd.commandStartTimestamp), HystrixEventType.CANCELLED);
}
handleCommandEnd(false); //user code never ran
} else if (_cmd.commandState.compareAndSet(CommandState.USER_CODE_EXECUTED, CommandState.UNSUBSCRIBED)) {
if (!_cmd.executionResult.containsTerminalEvent()) {
_cmd.eventNotifier.markEvent(HystrixEventType.CANCELLED, _cmd.commandKey);
try {
executionHook.onUnsubscribe(_cmd);
} catch (Throwable hookEx) {
logger.warn("Error calling HystrixCommandExecutionHook.onUnsubscribe", hookEx);
}
_cmd.executionResultAtTimeOfCancellation = _cmd.executionResult
.addEvent((int) (System.currentTimeMillis() - _cmd.commandStartTimestamp), HystrixEventType.CANCELLED);
}
handleCommandEnd(true); //user code did run
}
}
};
final Func0<Observable<R>> applyHystrixSemantics = new Func0<Observable<R>>() {
@Override
public Observable<R> call() {
if (commandState.get().equals(CommandState.UNSUBSCRIBED)) {
return Observable.never(); // 立即终止整个流程。
}//返回执行命令的Observable
return applyHystrixSemantics(_cmd);
}
};
final Func1<R, R> wrapWithAllOnNextHooks = new Func1<R, R>() {
@Override
public R call(R r) {
R afterFirstApplication = r;
try {
afterFirstApplication = executionHook.onComplete(_cmd, r);
} catch (Throwable hookEx) {
logger.warn("Error calling HystrixCommandExecutionHook.onComplete", hookEx);
}
try {
return executionHook.onEmit(_cmd, afterFirstApplication);
} catch (Throwable hookEx) {
logger.warn("Error calling HystrixCommandExecutionHook.onEmit", hookEx);
return afterFirstApplication;
}
}
};
final Action0 fireOnCompletedHook = new Action0() {
@Override
public void call() {
try {
executionHook.onSuccess(_cmd);
} catch (Throwable hookEx) {
logger.warn("Error calling HystrixCommandExecutionHook.onSuccess", hookEx);
}
}
};
return Observable.defer(new Func0<Observable<R>>() {
@Override
public Observable<R> call() {
/* this is a stateful object so can only be used once */
// CAS保证命令只执行一次
if (!commandState.compareAndSet(CommandState.NOT_STARTED, CommandState.OBSERVABLE_CHAIN_CREATED)) {
IllegalStateException ex = new IllegalStateException("This instance can only be executed once. Please instantiate a new instance.");
//TODO make a new error type for this
throw new HystrixRuntimeException(FailureType.BAD_REQUEST_EXCEPTION, _cmd.getClass(), getLogMessagePrefix() + " command executed multiple times - this is not permitted.", ex, null);
}
// 命令开始时间戳
commandStartTimestamp = System.currentTimeMillis();
// 打印日志
if (properties.requestLogEnabled().get()) {
// log this command execution regardless of what happened
if (currentRequestLog != null) {
currentRequestLog.addExecutedCommand(_cmd);
}
}
// 缓存开关，缓存KEY（这个是Hystrix中请求缓存功能，hystrix支持将一个请求结果缓存起来，
// 下一个具有相同key的请求将直接从缓存中取出结果，减少请求开销）
final boolean requestCacheEnabled = isRequestCachingEnabled();
final String cacheKey = getCacheKey();
/* try from cache first */
if (requestCacheEnabled) {//如果开启了缓存机制，则从缓存中获取结果
HystrixCommandResponseFromCache<R> fromCache = (HystrixCommandResponseFromCache<R>) requestCache.get(cacheKey);
if (fromCache != null) {
isResponseFromCache = true;
return handleRequestCacheHitAndEmitValues(fromCache, _cmd);
}
}
// 声明执行命令的Observable
Observable<R> hystrixObservable =
Observable.defer(applyHystrixSemantics)
.map(wrapWithAllOnNextHooks);
Observable<R> afterCache;
// put in cache 保存请求结果到缓存中
if (requestCacheEnabled && cacheKey != null) {
// wrap it for caching
HystrixCachedObservable<R> toCache = HystrixCachedObservable.from(hystrixObservable, _cmd);
HystrixCommandResponseFromCache<R> fromCache = (HystrixCommandResponseFromCache<R>) requestCache.putIfAbsent(cacheKey, toCache);
if (fromCache != null) {
// another thread beat us so we'll use the cached value instead
toCache.unsubscribe();
isResponseFromCache = true;
return handleRequestCacheHitAndEmitValues(fromCache, _cmd);
} else {
// we just created an ObservableCommand so we cast and return it
afterCache = toCache.toObservable();
}
} else {
afterCache = hystrixObservable;
}
return afterCache
//会在Observable结束前触发回调，无论是正常还是异常终止
.doOnTerminate(terminateCommandCleanup) // perform cleanup once (either on normal terminal state (this line), or unsubscribe (next line))
//取消订阅时的监听
.doOnUnsubscribe(unsubscribeCommandCleanup) // perform cleanup once
//Observable正常终止时的监听
.doOnCompleted(fireOnCompletedHook);
}
});
}
所以在 AbstractCommand#toObservable 方法里，我们只需要看这个返回的 Observable 对象的 call 方法即可，而在这里 默认没有开启缓存的话就是 ：
Observable<R> hystrixObservable =
Observable.defer(applyHystrixSemantics)
.map(wrapWithAllOnNextHooks);
那么我们主要来看 applyHystrixSemantics ，在该方法上面定义了
final Func0<Observable<R>> applyHystrixSemantics = new Func0<Observable<R>>() {
@Override
public Observable<R> call() {
if (commandState.get().equals(CommandState.UNSUBSCRIBED)) {
return Observable.never();
}
return applyHystrixSemantics(_cmd);
}
};
假设缓存特性未开启或者未命中缓存，那么代码将执行 applyHystrixSemantics 。传入的_cmd是一个GenericCommand(可以断点看看)，最终执行这个command中的run方法，本质就是完成对queryOrder方法的代理。
circuitBreaker.allowRequest() 如果为true，表示当前不处于熔断状态，正常执行，否则，调用 handleShortCircuitViaFallback 实现服务降级，如果我们配置了fallback方法，则会获得我们配置的fallback执行。执行路径为 : handleShortCircuitViaFallback ->getFallbackOrThrowException ->getFallbackObservable->HystrixCommand.getFallbackObservable->GenericCommand.getFallback();
如果当前hystrix处于未熔断状态，则
- getExecutionSemaphore 判断当前策略是否为信号量（TryableSemaphoreNoOp/TryableSemaphoreActual），如果是，则调用 tryAcquire 来获取信号量。如果当前信号量满了，则调用 handleSemaphoreRejectionViaFallback 方法。
- 调用 executeCommandAndObserve 获取命令执行Observable。
private Observable<R> applyHystrixSemantics(final AbstractCommand<R> _cmd) {
// mark that we're starting execution on the ExecutionHook
// if this hook throws an exception, then a fast-fail occurs with no fallback. No state is left inconsistent
executionHook.onStart(_cmd);
/* determine if we're allowed to execute */
// 判断是否处于熔断状态
if (circuitBreaker.allowRequest()) {
final TryableSemaphore executionSemaphore = getExecutionSemaphore();
final AtomicBoolean semaphoreHasBeenReleased = new AtomicBoolean(false);
final Action0 singleSemaphoreRelease = new Action0() {//操作命令
@Override
public void call() {
if (semaphoreHasBeenReleased.compareAndSet(false, true)) {
executionSemaphore.release();
}
}
};
final Action1<Throwable> markExceptionThrown = new Action1<Throwable>() {
@Override
public void call(Throwable t) {
eventNotifier.markEvent(HystrixEventType.EXCEPTION_THROWN, commandKey);
}
};
// 是否开启信号量资源隔离，未配置走 com.netflix.hystrix.AbstractCommand.TryableSemaphoreNoOp#tryAcquire 默认返回通过
if (executionSemaphore.tryAcquire()) {
try {
/* used to track userThreadExecutionTime */
executionResult = executionResult.setInvocationStartTime(System.currentTimeMillis());
return executeCommandAndObserve(_cmd) // 执行命令，以下三个是回调，可以不看
.doOnError(markExceptionThrown)
.doOnTerminate(singleSemaphoreRelease)
.doOnUnsubscribe(singleSemaphoreRelease);
} catch (RuntimeException e) {
return Observable.error(e);
}
} else {
// 走服务降级方法
return handleSemaphoreRejectionViaFallback();
}
} else {
return handleShortCircuitViaFallback();
}
}
我们县来看一下执行失败进入降级的逻辑，这里我们直接进入到 HystrixCommand#getFallbackObservable
@Override
final protected Observable<R> getFallbackObservable() {
return Observable.defer(new Func0<Observable<R>>() {
@Override
public Observable<R> call() {
try {
return Observable.just(getFallback());
} catch (Throwable ex) {
return Observable.error(ex);
}
}
});
}
这里的 getFallback 我们应该熟悉了，因为通过集成 HystrixCommand 类来实现熔断降级的时候我们重写了这个方法，而通过注解的话是通过 GenericCommand进行代理实现得，我们Debug一下，看看该类的 getFallback 方法做了什么：
可以发现他拿到了我们配置在注解上的方法，这一点是不是跟上文的手写是一个道理呢？ 然后进行调用获取结果返回。
好了，回到 AbstractCommand#applyHystrixSemantics ，接下去我们按照正常逻辑走到 AbstractCommand#executeCommandAndObserve，主要做了以下三件事情
- 定义不同的回调，doOnNext、doOnCompleted、onErrorResumeNext、doOnEach。
- 调用executeCommandWithSpecifiedIsolation获得执行命令的Observable
- 若执行命令超时特性开启，调用 Observable.lift 方法实现执行命令超时功能。
private Observable<R> executeCommandAndObserve(final AbstractCommand<R> _cmd) {
final HystrixRequestContext currentRequestContext = HystrixRequestContext.getContextForCurrentThread();
// Action和Func都是定义的一个动作，Action是无返回值，Func是有返回值
// doOnNext中的回调。即命令执行之前执行的操作
final Action1<R> markEmits = new Action1<R>() {
@Override
public void call(R r) {
if (shouldOutputOnNextEvents()) {
executionResult = executionResult.addEvent(HystrixEventType.EMIT);
eventNotifier.markEvent(HystrixEventType.EMIT, commandKey);
}
if (commandIsScalar()) {
long latency = System.currentTimeMillis() - executionResult.getStartTimestamp();
eventNotifier.markCommandExecution(getCommandKey(), properties.executionIsolationStrategy().get(), (int) latency, executionResult.getOrderedList());
eventNotifier.markEvent(HystrixEventType.SUCCESS, commandKey);
executionResult = executionResult.addEvent((int) latency, HystrixEventType.SUCCESS);
circuitBreaker.markSuccess();
}
}
};
// doOnCompleted中的回调。命令执行完毕后执行的操作
final Action0 markOnCompleted = new Action0() {
@Override
public void call() {
if (!commandIsScalar()) {
long latency = System.currentTimeMillis() - executionResult.getStartTimestamp();
eventNotifier.markCommandExecution(getCommandKey(), properties.executionIsolationStrategy().get(), (int) latency, executionResult.getOrderedList());
eventNotifier.markEvent(HystrixEventType.SUCCESS, commandKey);
executionResult = executionResult.addEvent((int) latency, HystrixEventType.SUCCESS);
circuitBreaker.markSuccess();
}
}
};
// onErrorResumeNext中的回调。命令执行失败后的回退逻辑
final Func1<Throwable, Observable<R>> handleFallback = new Func1<Throwable, Observable<R>>() {
@Override
public Observable<R> call(Throwable t) {
Exception e = getExceptionFromThrowable(t);
executionResult = executionResult.setExecutionException(e);
if (e instanceof RejectedExecutionException) {
return handleThreadPoolRejectionViaFallback(e);
} else if (t instanceof HystrixTimeoutException) {
return handleTimeoutViaFallback();
} else if (t instanceof HystrixBadRequestException) {
return handleBadRequestByEmittingError(e);
} else {
/*
* Treat HystrixBadRequestException from ExecutionHook like a plain HystrixBadRequestException.
*/
if (e instanceof HystrixBadRequestException) {
eventNotifier.markEvent(HystrixEventType.BAD_REQUEST, commandKey);
return Observable.error(e);
}
return handleFailureViaFallback(e);
}
}
};
// doOnEach中的回调。`Observable`每发射一个数据都会执行这个回调，设置请求上下文
final Action1<Notification<? super R>> setRequestContext = new Action1<Notification<? super R>>() {
@Override
public void call(Notification<? super R> rNotification) {
setRequestContextIfNeeded(currentRequestContext);
}
};
Observable<R> execution;
// 是否开启超时降级
if (properties.executionTimeoutEnabled().get()) {
execution = executeCommandWithSpecifiedIsolation(_cmd)
.lift(new HystrixObservableTimeoutOperator<R>(_cmd));
} else {
execution = executeCommandWithSpecifiedIsolation(_cmd);
}
// 发射
return execution.doOnNext(markEmits)
.doOnCompleted(markOnCompleted)
.onErrorResumeNext(handleFallback)
.doOnEach(setRequestContext);
}
executeCommandWithSpecifiedIsolation:这个方法首先是根据当前不同的资源隔离策略执行不同的逻辑，THREAD、SEMAPHORE:
private Observable<R> executeCommandWithSpecifiedIsolation(final AbstractCommand<R> _cmd) {
// 是否开启 THREAD 资源隔离降级
if (properties.executionIsolationStrategy().get() == ExecutionIsolationStrategy.THREAD) {
// ......省略代码
} else { // 否则进入这里
return Observable.defer(new Func0<Observable<R>>() {
@Override
public Observable<R> call() {
executionResult = executionResult.setExecutionOccurred();
if (!commandState.compareAndSet(CommandState.OBSERVABLE_CHAIN_CREATED, CommandState.USER_CODE_EXECUTED)) {
return Observable.error(new IllegalStateException("execution attempted while in state : " + commandState.get().name()));
}
metrics.markCommandStart(commandKey, threadPoolKey, ExecutionIsolationStrategy.SEMAPHORE);
// semaphore isolated
// store the command that is being run
endCurrentThreadExecutingCommand = Hystrix.startCurrentThreadExecutingCommand(getCommandKey());
try {
executionHook.onRunStart(_cmd);
executionHook.onExecutionStart(_cmd);
// 真正的执行
return getUserExecutionObservable(_cmd); //the getUserExecutionObservable method already wraps sync exceptions, so this shouldn't throw
} catch (Throwable ex) {
//If the above hooks throw, then use that as the result of the run method
return Observable.error(ex);
}
}
});
}
}
这里就不展开实现细节，我们直接看执行的方法 getUserExecutionObservable 。然后会执行 HystrixCommand#getExecutionObservable
@Override
final protected Observable<R> getExecutionObservable() {
return Observable.defer(new Func0<Observable<R>>() {
@Override
public Observable<R> call() {
try {
return Observable.just(run());
} catch (Throwable ex) {
return Observable.error(ex);
}
}
}).doOnSubscribe(new Action0() {
@Override
public void call() {
// Save thread on which we get subscribed so that we can interrupt it later if needed
executionThread.set(Thread.currentThread());
}
});
}
又看到熟悉的代码 ，这个 run() 方法在通过集成 HystrixCommand 类来实现熔断降级的时候我们重写了这个方法，是真正的执行方法。
这里最终调用的是run方法，通过Observable.just， just是RxJava中的一个操作符，它可以接受一个或者多个参数来创建一个Observable对象。而这个run（）方法是一个抽象方法，在HystrixCommand中并没有实现，而是在子类中实现，而此时传递的cmd=GenricCommand正好实现了HystrixCommand，重写了run方法。
@Override
protected Object run() throws Exception {
LOGGER.debug("execute command: {}", getCommandKey().name());
return process(new Action() {
@Override
Object execute() {
return getCommandAction().execute(getExecutionType());
}
});
}
大家有没有发现，这里的实现和我们前面自定义的 HystrixCommandService 实现是一样的，同样是集成HystrixCommand，重写run方法。这里也是如此。
- 首先调用 getCommandAction() 方法获取 CommandAction ，我们的示例中获取到的是MethodExecutionAction 。
- 然后调用 MethodExecutionAction.execute 方法，传入 ExecutionType 参数，我们的示例中传入的是 ExecutionType.SYNCHRONOUS 。
拿到我们的真实方法进行调用返回。下面附上整个过程的流程图：
