Spring Cloud——Hystrix源码分析

spring-cloud-netflix-hystrix 原理源码分析：

本文主要针对 spring-cloud-dependencies   Hoxton.SR4版本, spring-cloud-starter-netflix-hystrix 源码的解析。

 

　　对于未接触过 hystrix 的小伙伴可以参考 https://www.cnblogs.com/wuzhenzhao/p/9473073.html 进行一些基础知识的了解。

 

本文主要从以下几个点来分析：

• 1、手写实现简易版 Hystrix 体验。
• 2、RXJava 基础知识体验。
• 3、Hystrix 源码流程分析。

手写实现简易版Hystrix 体验

Hystrix 提供了熔断、请求超时、资源隔离等多种方式实现服务降级。可以通过注解 @HystrixCommand、或者继承 HystrixCommand 来实现降级，以及一些请求合并等操作。

 

我们需要知道的是，当采用 @HystrixCommand 注解来实现服务降级，在Hystrix 的内部是采用AOP的方式进行拦截处理请求的，我们这里就先来实现一下简易版的 Hystrix 来体会一下，主要分为以下步骤

• 1、定义自己的@HystrixCommand 注解。
• 2、实现拦截请求的处理逻辑。
• 3、测试调用。

1、自定义注解 @HybHystrixCommand

@Target(ElementType.METHOD) @Retention(RetentionPolicy.RUNTIME) @Documented public @interface HybHystrixCommand { /** * 默认超时时间 * * @return */ int timeout() default 1000; /** * 回退方法 * * @return */ String fallback() default ""; }

2、编写切面类，实现简易的逻辑处理

@Component @Aspect public class HybHystrixCommandAspect { //线程池的处理，基于这个线程池的处理统计可以达到 THREAD 资源限流 ExecutorService executorService = Executors.newFixedThreadPool(10); //注解切点 @Pointcut(value = "@annotation(com.yibo.demo.custom.hystrix.HybHystrixCommand)") 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、编写测试，调用：

@HybHystrixCommand(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 基础知识体验

由于马上我们要去看 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的数据统计是采用的滑动窗口，关于滑动窗口可以直接访问 滑动窗口在线演示地址。

Spring Boot-SPI机制

根据自动装载原则可以在spring-cloud-netflix-hystrix-2.2.5.RELEASE.jar下的META-INF目录下找到 spring.factories文件

org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ org.springframework.cloud.netflix.hystrix.HystrixAutoConfiguration,\ org.springframework.cloud.netflix.hystrix.HystrixCircuitBreakerAutoConfiguration,\ org.springframework.cloud.netflix.hystrix.ReactiveHystrixCircuitBreakerAutoConfiguration,\ org.springframework.cloud.netflix.hystrix.security.HystrixSecurityAutoConfiguration org.springframework.cloud.client.circuitbreaker.EnableCircuitBreaker=\ org.springframework.cloud.netflix.hystrix.HystrixCircuitBreakerConfiguration

HystrixCircuitBreakerConfiguration.class

@Configuration(proxyBeanMethods = false) public class HystrixCircuitBreakerConfiguration { // 注册切面 @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)); } /** * 当 ApplicationContext 关闭时，确保清除 Hystrix内部状态。 */ private class HystrixShutdownHook implements DisposableBean { @Override public void destroy() throws Exception { // Just call Hystrix to reset thread pool etc. Hystrix.reset(); } } }

Hystrix熔断的@HystrixCommand注解，是通过HystrixCommandAspect这个切面来处理的。其中我们关注@Around注解声明的方法，它针对于请求合并，以及降级的注解进行代理。这里我们重点针对HystrixCommand这个注解进行详细分析。

HystrixCommandAspect.class

@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); // 同时有 @HystrixCommand 和@HystrixCollapser 注解,抛出异常 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, 创建MetaHolder, MetaHolder 里面包含了所有信息 MetaHolderFactory metaHolderFactory = META_HOLDER_FACTORY_MAP.get(HystrixPointcutType.of(method)); // 获取元数据，比如调用方法，HystrixProperty注解数据、方法参数等 MetaHolder metaHolder = metaHolderFactory.create(joinPoint); //获取调用者，它持有一个命令对象，并且可以在合适的时候通过这个命令对象完成具体的业务逻辑 HystrixInvokable invokable = HystrixCommandFactory.getInstance().create(metaHolder); // 命令类型: 异步 ASYNCHRONOUS, 同步 SYNCHRONOUS, 响应式（异步回调） OBSERVABLE; 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; } // 省略后续代码...... }

metaHolderFactory.create(joinPoint);

实现类：CommandMetaHolderFactory，CollapserMetaHolderFactory；

public class HystrixCommandAspect { 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(); } } }

HystrixCommandFactory.getInstance().create(metaHolder)

public class HystrixCommandFactory { private static final HystrixCommandFactory INSTANCE = new HystrixCommandFactory(); public static HystrixCommandFactory getInstance() { return INSTANCE; } // 根据 metaHolder 创建 HystrixInvokable 命令 public HystrixInvokable create(MetaHolder metaHolder) { HystrixInvokable executable; if (metaHolder.isCollapserAnnotationPresent()) { executable = new CommandCollapser(metaHolder); } else if (metaHolder.isObservable()) { executable = new GenericObservableCommand(HystrixCommandBuilderFactory.getInstance().create(metaHolder)); } else { executable = new GenericCommand(HystrixCommandBuilderFactory.getInstance().create(metaHolder)); } return executable; } }

CommandExecutor.execute(invokable, executionType, metaHolder);

public class CommandExecutor { public static Object execute(HystrixInvokable invokable, ExecutionType executionType, MetaHolder metaHolder) throws RuntimeException { Validate.notNull(invokable); Validate.notNull(metaHolder); switch (executionType) { case SYNCHRONOUS: { // 强转成HystrixExecutable 同步执行 return castToExecutable(invokable, executionType).execute(); } case ASYNCHRONOUS: { // 强转成HystrixExecutable 异步执行 HystrixExecutable executable = castToExecutable(invokable, executionType); // 如果有 fallback方法，且是异步执行，则执行并返回包装结果 if (metaHolder.hasFallbackMethodCommand() && ExecutionType.ASYNCHRONOUS == metaHolder.getFallbackExecutionType()) { return new FutureDecorator(executable.queue()); } return executable.queue(); } case OBSERVABLE: { // 强转成 HystrixObservable HystrixObservable observable = castToObservable(invokable); // 判断执行模式是不是急切/懒惰，来选择模式执行 return ObservableExecutionMode.EAGER == metaHolder.getObservableExecutionMode() ? observable.observe() : observable.toObservable(); } default: throw new RuntimeException("unsupported execution type: " + executionType); } } }

然后进入 CommandExecutor#execute 方法这个方法主要用来执行命令，从代码中可以看出这里有三个执行类型，分别是同步、异步、以及响应式。其中，响应式又分为Cold Observable（observable.toObservable()） 和 HotObservable（observable.observe()）默认的executionType=SYNCHRONOUS ，同步请求。

• execute()：同步执行，返回一个单一的对象结果，发生错误时抛出异常。
• queue()：异步执行，返回一个 Future 对象，包含着执行结束后返回的单一结果。
• observe()：这个方法返回一个 Observable 对象，它代表操作的多个结果，但是已经被订阅者消费掉了。
• toObservable()：这个方法返回一个 Observable 对象，它代表操作的多个结果，需要咱们自己手动订阅并消费掉。

接着HystrixCommand.execute()方法中，首先调用queue()，这个方法会返回一个future对象。

public abstract class HystrixCommand<R> extends AbstractCommand<R> implements HystrixExecutable<R>, HystrixInvokableInfo<R>, HystrixObservable<R> { public R execute() { try { // 返回Feture对象（封装异步处理的结果） return queue().get(); } catch (Exception e) { throw Exceptions.sneakyThrow(decomposeException(e)); } } public Future<R> queue() { // toObservable转换为Observable,toBlocking转换为BlockingObservable, // toFuture转换为Future,完成了Observable的创建和订阅 final Future<R> delegate = toObservable().toBlocking().toFuture(); final Future<R> f = new Future<R>() { // 通过cancel方法中断执行线程 @Override public boolean cancel(boolean mayInterruptIfRunning) { if (delegate.isCancelled()) { return false; } // 如果 execution.isolation.thread.interruptOnFutureCancel = true（默认false） if (HystrixCommand.this.getProperties().executionIsolationThreadInterruptOnFutureCancel().get()) { // CAS 设置 interruptOnFutureCancel.compareAndSet(false, mayInterruptIfRunning); } // 执行目标future的cancel final boolean res = delegate.cancel(interruptOnFutureCancel.get()); if (!isExecutionComplete() && interruptOnFutureCancel.get()) { // 获取执行线程 final Thread t = executionThread.get(); if (t != null && !t.equals(Thread.currentThread())) { // 中断线程 t.interrupt(); } } return res; } @Override public boolean isCancelled() { return delegate.isCancelled(); } @Override public boolean isDone() { return delegate.isDone(); } @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); } }; // 如果已经执行完毕了就返回 if (f.isDone()) { try { f.get(); return f; } catch (Exception e) { //省略异常处理...... } } 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() 中被订阅的。到这还是我们上面的例子里面的代码。该方法主要做了以下几件事：

• 1、创建一些命令供后续的回调使用

• 2、调用 isRequestCachingEnabled(); 判断请求结果缓存功能是否开启，如果开启并且命中了缓存，则会以Observable形式返回一个缓存结果

• 3、创建执行命令的Observable：hystrixObservable，

• 4、当缓存处于开启状态并且没有命中缓存时，则创建一个“订阅了执行命令的Observable”：HystrixCommandResponseFromCache

• 5、创建存储到缓存的Observable： HystrixCachedObservable当缓存特性没有开启时，则返回执行命令的Observable。

  • 将toCache添加到缓存中，返回获取缓存的Observable：fromCache
  • 如果添加失败: fromCache!=null, 则调用 toCache.unsubscribe() 方法，取消HystrixCachedObservable 的订阅
  • 如果添加成功，则调用 toCache.toObservable(); 获得缓存Observable

• 6、当缓存特性没有开启时，则返回执行命令的Observable。

AbstractCommand.toObservable()

abstract class AbstractCommand<R> implements HystrixInvokableInfo<R>, HystrixObservable<R> { public Observable<R> toObservable() { final AbstractCommand<R> _cmd = this; //doOnCompleted handler already did all of the SUCCESS work //doOnError handler already did all of the FAILURE/TIMEOUT/REJECTION/BAD_REQUEST work // 命令执行完的回调操作 终止命令清理 //会在Observable结束前触发回调该call方法，无论是正常还是异常终止 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 } } }; //mark the command as CANCELLED and store the latency (in addition to standard cleanup) // 将命令标记为已取消并存储延迟（除了标准清理） //取消订阅时的监听会进行回调该 call方法 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); } } }; // 创建Observable,设置各种处理操作 return Observable.defer(new Func0<Observable<R>>() { @Override public Observable<R> call() { // 设置已启动标志, 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(); // 如果配置允许缓存，先试图从缓存获取，默认 false if (requestCacheEnabled) { HystrixCommandResponseFromCache<R> fromCache = (HystrixCommandResponseFromCache<R>) requestCache.get(cacheKey); if (fromCache != null) { isResponseFromCache = true; return handleRequestCacheHitAndEmitValues(fromCache, _cmd); } } // 声明执行命令的Observable // 创建Observable, applyHystrixSemantics() 来生成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) //取消订阅时的监听 .doOnUnsubscribe(unsubscribeCommandCleanup) //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。

abstract class AbstractCommand<R> implements HystrixInvokableInfo<R>, HystrixObservable<R> { 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); // 是否允许请求，即断路器是否开启 ，这里也有好几种情况 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); } }; // 获取信号，并返回对应的 Observable // 是否开启信号量资源隔离，未配置走 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

public abstract class HystrixCommand<R> extends AbstractCommand<R> implements HystrixExecutable<R>, HystrixInvokableInfo<R>, HystrixObservable<R> { @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，主要做了以下三件事情

• 1、定义不同的回调，doOnNext、doOnCompleted、onErrorResumeNext、doOnEach。
• 2、调用executeCommandWithSpecifiedIsolation获得执行命令的Observable。
• 3、若执行命令超时特性开启，调用 Observable.lift 方法实现执行命令超时功能。

abstract class AbstractCommand<R> implements HystrixInvokableInfo<R>, HystrixObservable<R> { 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) { // HystrixBadRequestException 异常回调 return handleBadRequestByEmittingError(e); } else { 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，实现 线程隔离、请求发送 等操作 Observable<R> execution; // 判断 超时监控功能是否打开 if (properties.executionTimeoutEnabled().get()) { // HystrixObservableTimeoutOperator 转换对应的 Observable 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。

abstract class AbstractCommand<R> implements HystrixInvokableInfo<R>, HystrixObservable<R> { private Observable<R> executeCommandWithSpecifiedIsolation(final AbstractCommand<R> _cmd) { // 线程隔离, 是否开启 THREAD 资源隔离降级 if (properties.executionIsolationStrategy().get() == ExecutionIsolationStrategy.THREAD) { //创建一个Observable 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.THREAD); // 该命令在包装线程中超时，将立即返回，并且不会增加任何计数器或其他此类逻辑 if (isCommandTimedOut.get() == TimedOutStatus.TIMED_OUT) { // the command timed out in the wrapping thread so we will return immediately // and not increment any of the counters below or other such logic return Observable.error(new RuntimeException("timed out before executing run()")); } // 设置线程启动 if (threadState.compareAndSet(ThreadState.NOT_USING_THREAD, ThreadState.STARTED)) { //we have not been unsubscribed, so should proceed HystrixCounters.incrementGlobalConcurrentThreads(); threadPool.markThreadExecution(); // store the command that is being run endCurrentThreadExecutingCommand = Hystrix.startCurrentThreadExecutingCommand(getCommandKey()); executionResult = executionResult.setExecutedInThread(); try { executionHook.onThreadStart(_cmd); executionHook.onRunStart(_cmd); executionHook.onExecutionStart(_cmd); //返回 Observable,这个函数最终会返回一个封装了我们的run()逻辑的Observable return getUserExecutionObservable(_cmd); } catch (Throwable ex) { return Observable.error(ex); } } else { //command has already been unsubscribed, so return immediately return Observable.error(new RuntimeException("unsubscribed before executing run()")); } } }).doOnTerminate(new Action0() { @Override public void call() { if (threadState.compareAndSet(ThreadState.STARTED, ThreadState.TERMINAL)) { handleThreadEnd(_cmd); } if (threadState.compareAndSet(ThreadState.NOT_USING_THREAD, ThreadState.TERMINAL)) { //if it was never started and received terminal, then no need to clean up (I don't think this is possible) } //if it was unsubscribed, then other cleanup handled it } }).doOnUnsubscribe(new Action0() { @Override public void call() { if (threadState.compareAndSet(ThreadState.STARTED, ThreadState.UNSUBSCRIBED)) { handleThreadEnd(_cmd); } if (threadState.compareAndSet(ThreadState.NOT_USING_THREAD, ThreadState.UNSUBSCRIBED)) { //if it was never started and was cancelled, then no need to clean up } //if it was terminal, then other cleanup handled it } }).subscribeOn(threadPool.getScheduler(new Func0<Boolean>() { @Override public Boolean call() { return properties.executionIsolationThreadInterruptOnTimeout().get() && _cmd.isCommandTimedOut.get() == TimedOutStatus.TIMED_OUT; } })); } 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); } catch (Throwable ex) { //If the above hooks throw, then use that as the result of the run method return Observable.error(ex); } } }); } } }

• 判断是否允许发送请求，这基于 断路器 实现，如果 断路器 打开，则进行对应回调处理（失败或降级）。
• 如果 断路器 关闭，则进行请求，先获取 信号，获取失败则处理对应回调。
• 获取成功，则由方法 executeCommandAndObserve 创建对应的 Observable 实现 线程隔离、请求发送 等操作，同时注册了对应的 生命周期回调。

getUserExecutionObservable

然后会执行 HystrixCommand#getExecutionObservable

abstract class AbstractCommand<R> implements HystrixInvokableInfo<R>, HystrixObservable<R> { private Observable<R> getUserExecutionObservable(final AbstractCommand<R> _cmd) { Observable<R> userObservable; try { userObservable = getExecutionObservable(); } catch (Throwable ex) { userObservable = Observable.error(ex); } return userObservable .lift(new ExecutionHookApplication(_cmd)) .lift(new DeprecatedOnRunHookApplication(_cmd)); } } public abstract class HystrixCommand<R> extends AbstractCommand<R> implements HystrixExecutable<R>, HystrixInvokableInfo<R>, HystrixObservable<R> { @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方法。

@ThreadSafe public class GenericCommand extends AbstractHystrixCommand<Object> { @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 。

拿到我们的真实方法进行调用返回。下面附上整个过程的流程图：

参考： https://www.cnblogs.com/wuzhenzhao/p/13726372.html

https://www.cnblogs.com/lucky-yqy/p/14607522.html

https://www.cnblogs.com/dabenxiang/p/13764076.html