【一起学源码-微服务】Hystrix 源码二：Hystrix核心流程：Hystix非降级逻辑流程梳理

2020 年 2 月 13 日
筆記

说明

原创不易，如若转载请标明来源！

欢迎关注本人微信公众号：壹枝花算不算浪漫更多内容也可查看本人博客：一枝花算不算浪漫

前言

前情回顾

上一讲我们讲了配置了feign.hystrix.enabled=true之后，默认的Targeter就会构建成HystrixTargter, 然后通过对应的HystrixInvocationHandler 生成对应的动态代理。

本讲目录

这一讲开始讲解Hystrix相关代码，当然还是基于上一个组件Feign的基础上开始讲解的，这里默认你已经对Feign有过大致了解。

目录如下：

线程池初始化过程
HystrixCommand通过线程池执行原理

由于这里面代码比较多，所以我都是将一些主要核心代码发出来，这里后面会汇总一个流程图，可以参考流程图自己一点点调试。

这里建议在回调的地方都加上断点，而且修改feign和hystrix超时时间，浏览器发送请求后，一步步debug代码。

源码分析

线程池初始化过程

上一讲已经讲过激活Hystrix后，构造的InvocationHandler为HystrixInvocationHandler，所以当调用FeignClient服务实例的时候，会先执行HystrixInvocationHandler.invoke()方法，这里我们先跟进这个方法：

final class HystrixInvocationHandler implements InvocationHandler {        @Override      public Object invoke(final Object proxy, final Method method, final Object[] args)              throws Throwable {            // 构建一个HystrixCommand          // HystrixCommand构造参数需要Setter对象          HystrixCommand<Object> hystrixCommand = new HystrixCommand<Object>(setterMethodMap.get(method)) {              @Override              protected Object run() throws Exception {                  try {                      // 执行SynchronousMethodHandler.invoke方法                      return HystrixInvocationHandler.this.dispatch.get(method).invoke(args);                  } catch (Exception e) {                      throw e;                  } catch (Throwable t) {                      throw (Error) t;                  }              }          }            // 省略部分代码...            return hystrixCommand.execute();      }  }

这里主要是构造HystrixCommand，我们先看看它的构造函数以及线程池池初始化的代码：

public abstract class HystrixCommand<R> extends AbstractCommand<R> implements HystrixExecutable<R>, HystrixInvokableInfo<R>, HystrixObservable<R> {        protected HystrixCommand(HystrixCommandGroupKey group) {          super(group, null, null, null, null, null, null, null, null, null, null, null);      }  }    abstract class AbstractCommand<R> implements HystrixInvokableInfo<R>, HystrixObservable<R> {      protected AbstractCommand(HystrixCommandGroupKey group, HystrixCommandKey key, HystrixThreadPoolKey threadPoolKey, HystrixCircuitBreaker circuitBreaker, HystrixThreadPool threadPool,              HystrixCommandProperties.Setter commandPropertiesDefaults, HystrixThreadPoolProperties.Setter threadPoolPropertiesDefaults,              HystrixCommandMetrics metrics, TryableSemaphore fallbackSemaphore, TryableSemaphore executionSemaphore,              HystrixPropertiesStrategy propertiesStrategy, HystrixCommandExecutionHook executionHook) {            this.commandGroup = initGroupKey(group);          this.commandKey = initCommandKey(key, getClass());          this.properties = initCommandProperties(this.commandKey, propertiesStrategy, commandPropertiesDefaults);          this.threadPoolKey = initThreadPoolKey(threadPoolKey, this.commandGroup, this.properties.executionIsolationThreadPoolKeyOverride().get());          this.metrics = initMetrics(metrics, this.commandGroup, this.threadPoolKey, this.commandKey, this.properties);          this.circuitBreaker = initCircuitBreaker(this.properties.circuitBreakerEnabled().get(), circuitBreaker, this.commandGroup, this.commandKey, this.properties, this.metrics);          // 初始化线程池          this.threadPool = initThreadPool(threadPool, this.threadPoolKey, threadPoolPropertiesDefaults);          // 省略部分代码...      }        private static HystrixThreadPool initThreadPool(HystrixThreadPool fromConstructor, HystrixThreadPoolKey threadPoolKey, HystrixThreadPoolProperties.Setter threadPoolPropertiesDefaults) {          if (fromConstructor == null) {              // get the default implementation of HystrixThreadPool              return HystrixThreadPool.Factory.getInstance(threadPoolKey, threadPoolPropertiesDefaults);          } else {              return fromConstructor;          }      }  }    public interface HystrixThreadPool {      final static ConcurrentHashMap<String, HystrixThreadPool> threadPools = new ConcurrentHashMap<String, HystrixThreadPool>();        static HystrixThreadPool getInstance(HystrixThreadPoolKey threadPoolKey, HystrixThreadPoolProperties.Setter propertiesBuilder) {          // 这个线程池的key就是我们feignClient定义的value名称，其他服务的projectName          // 在我们的demo中：key = serviceA          String key = threadPoolKey.name();            // threadPools是一个map，key就是serviceA          HystrixThreadPool previouslyCached = threadPools.get(key);          if (previouslyCached != null) {              return previouslyCached;          }            // 初始化线程池          synchronized (HystrixThreadPool.class) {              if (!threadPools.containsKey(key)) {                  threadPools.put(key, new HystrixThreadPoolDefault(threadPoolKey, propertiesBuilder));              }          }          return threadPools.get(key);      }  }      public abstract class HystrixThreadPoolProperties {      /* defaults */      static int default_coreSize = 10;      static int default_maximumSize = 10;      static int default_keepAliveTimeMinutes = 1;      static int default_maxQueueSize = -1;      static boolean default_allow_maximum_size_to_diverge_from_core_size = false;      static int default_queueSizeRejectionThreshold = 5;      static int default_threadPoolRollingNumberStatisticalWindow = 10000;      static int default_threadPoolRollingNumberStatisticalWindowBuckets = 10;        // 省略部分代码...  }

这里主要是初始化线程池的逻辑，从HystrixCommand一直到HystrixThreadPoolProperties。这里的threadPools 是一个Map，一个serviceName会对应一个线程池。

线程池的默认配置都在HystrixThreadPoolProperties中。线程池的核心线程和最大线程数都是10，队列的大小为-1，这里意思是不使用队列。

HystrixCommand构造函数需要接收一个Setter对象，Setter中包含两个很重要的属性，groupKey和commandKey, 这里看下Setter是如何构造的：

final class HystrixInvocationHandler implements InvocationHandler {        HystrixInvocationHandler(Target<?> target, Map<Method, MethodHandler> dispatch,                             SetterFactory setterFactory, FallbackFactory<?> fallbackFactory) {          this.target = checkNotNull(target, "target");          this.dispatch = checkNotNull(dispatch, "dispatch");          this.fallbackFactory = fallbackFactory;          this.fallbackMethodMap = toFallbackMethod(dispatch);          this.setterMethodMap = toSetters(setterFactory, target, dispatch.keySet());      }        static Map<Method, Setter> toSetters(SetterFactory setterFactory, Target<?> target,                                         Set<Method> methods) {          Map<Method, Setter> result = new LinkedHashMap<Method, Setter>();          for (Method method : methods) {              method.setAccessible(true);              result.put(method, setterFactory.create(target, method));          }          return result;      }  }    public interface SetterFactory {      HystrixCommand.Setter create(Target<?> target, Method method);      final class Default implements SetterFactory {          @Override          public HystrixCommand.Setter create(Target<?> target, Method method) {              // groupKey既是调用的服务服务名称：serviceA              String groupKey = target.name();              // commandKey即是方法的名称+入参定义等，一个commandKey能够确定这个类中唯一的一个方法              String commandKey = Feign.configKey(target.type(), method);              return HystrixCommand.Setter                  .withGroupKey(HystrixCommandGroupKey.Factory.asKey(groupKey))                  .andCommandKey(HystrixCommandKey.Factory.asKey(commandKey));              }          }      }  }

构建一个HystrixCommand时必须要传入这两个参数。

groupKey: 就是调用的服务名称，例如我们demo中的ServiceA，groupKey对应着一个线程池。
commandKey: 一个FeignClient接口中的一个方法就是一个commandKey, 其组成为方法名和入参等信息。

groupkey和commandKey是一对多的关系，例如ServiceA中的2个方法，那么groupKey就对应着这个ServiceA中的2个commandKey。

groupKey -> target.name() -> ServiceA -> @FeignClient注解里设置的服务名称

commanKey -> ServiceAFeignClient#sayHello(String)

这里回调函数执行HystrixInvocationHandler.this.dispatch.get(method).invoke(args) 其实就是执行SynchronousMethodHandler.invoke() 方法了。但是什么时候才会回调回来呢？后面接着看吧。

HystrixCommand通过线程池执行原理

上面已经看了线程池的初始化过程，当一个服务第一次被调用的时候，会判断threadPools (数据结构为ConcurrentHashMap) 中是否存在这个serviceName对应的线程池，如果没有的话则会初始化一个对应的线程池。线程池默认配置属性在HystrixThreadPoolProperties中可以看到。

Hystrix线程池默认是不使用队列进行线程排队的，核心线程数为10。接下来我们看看创建HystrixCommand后，线程池是如何将HystrixCommand 命令提交的：

public abstract class HystrixCommand<R> extends AbstractCommand<R> implements HystrixExecutable<R>, HystrixInvokableInfo<R>, HystrixObservable<R> {      public R execute() {          try {              return queue().get();          } catch (Exception e) {              throw Exceptions.sneakyThrow(decomposeException(e));          }      }        public Future<R> queue() {          final Future<R> delegate = toObservable().toBlocking().toFuture();            final Future<R> f = new Future<R>() {                @Override              public boolean cancel(boolean mayInterruptIfRunning) {                  if (delegate.isCancelled()) {                      return false;                  }                    if (HystrixCommand.this.getProperties().executionIsolationThreadInterruptOnFutureCancel().get()) {                      interruptOnFutureCancel.compareAndSet(false, mayInterruptIfRunning);                  }                    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) {                  Throwable t = decomposeException(e);                  if (t instanceof HystrixBadRequestException) {                      return f;                  } else if (t instanceof HystrixRuntimeException) {                      HystrixRuntimeException hre = (HystrixRuntimeException) t;                      switch (hre.getFailureType()) {                      case COMMAND_EXCEPTION:                      case TIMEOUT:                          // we don't throw these types from queue() only from queue().get() as they are execution errors                          return f;                      default:                          // these are errors we throw from queue() as they as rejection type errors                          throw hre;                      }                  } else {                      throw Exceptions.sneakyThrow(t);                  }              }          }            return f;      }  }

这里又是一堆的回调函数，我们可以在每个回调函数中打上断点，然后一点点调试。这里主要是通过toObservable()方法构造了一个Future<R>, 然后包装此Future，添加了中断等逻辑，后面使用f.get() 阻塞获取线程执行结果，最后返回Future对象。

这里我们的重点在于寻找哪里将HystrixCommand丢入线程池，然后返回一个Future的。接着往后跟进代码：

abstract class AbstractCommand<R> implements HystrixInvokableInfo<R>, HystrixObservable<R> {      public Observable<R> toObservable() {          // _cmd就是HystrixInvocationHandler对象          // 里面包含要请求的method信息，threadPool信息，groupKey，commandKey等信息          final AbstractCommand<R> _cmd = this;          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);              }          };            // 省略部分回调函数代码...            return Observable.defer(new Func0<Observable<R>>() {              @Override              public Observable<R> call() {                  // 是否使用请求缓存，默认为false                  final boolean requestCacheEnabled = isRequestCachingEnabled();                  // 请求缓存相关                  final String cacheKey = getCacheKey();                    // 省略部分代码...                    Observable<R> hystrixObservable =                          Observable.defer(applyHystrixSemantics)                                  .map(wrapWithAllOnNextHooks);                    Observable<R> afterCache;                    // put in cache                  if (requestCacheEnabled && cacheKey != null) {                      // 省略部分代码...                  } else {                      afterCache = hystrixObservable;                  }                    return afterCache                          .doOnTerminate(terminateCommandCleanup)                          .doOnUnsubscribe(unsubscribeCommandCleanup)                          .doOnCompleted(fireOnCompletedHook);              }          });      }  }

toObservable()是比较核心的代码，这里也是定义了很多回调函数，上面代码做了精简，留下一些核心逻辑，在defer()中构造返回了一个Observable对象，这个Observable是包含上面的一些回调函数的。

通过debug代码，这里会直接执行到applyHystrixSemantics这个构造函数Func0中的call()方法中，通过语意我们可以大致猜到这个函数的意思：应用Hystrix语义接着往下跟进代码：

abstract class AbstractCommand<R> implements HystrixInvokableInfo<R>, HystrixObservable<R> {      private Observable<R> applyHystrixSemantics(final AbstractCommand<R> _cmd) {          executionHook.onStart(_cmd);          // 判断是否短路          if (circuitBreaker.attemptExecution()) {              final TryableSemaphore executionSemaphore = getExecutionSemaphore();              final AtomicBoolean semaphoreHasBeenReleased = new AtomicBoolean(false);              // 如果不使用Semaphore配置，那么tryAcquire使用的是TryableSemaphoreNoOp中的方法，返回true              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();          }      }  }

这里面我们默认使用的线程池的隔离配置，所以executionSemaphore.tryAcquire()都会返回true，这里有个重要的方法：executeCommandAndObserve(_cmd), 我们继续往后跟进这个方法：

abstract class AbstractCommand<R> implements HystrixInvokableInfo<R>, HystrixObservable<R> {      private Observable<R> executeCommandAndObserve(final AbstractCommand<R> _cmd) {          final HystrixRequestContext currentRequestContext = HystrixRequestContext.getContextForCurrentThread();            // 省略部分回调函数...            Observable<R> execution;          // 默认配置timeOutEnabled为true          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);      }  }

对于Hystrix来说，默认是开启超时机制的，这里会执行executeCommandWithSpecifiedIsolation(), 返回一个执行的Observable.还是通过方法名我们可以猜测这个方法是：使用指定的隔离执行命令继续往里面跟进：

abstract class AbstractCommand<R> implements HystrixInvokableInfo<R>, HystrixObservable<R> {      private Observable<R> executeCommandWithSpecifiedIsolation(final AbstractCommand<R> _cmd) {          if (properties.executionIsolationStrategy().get() == ExecutionIsolationStrategy.THREAD) {              // mark that we are executing in a thread (even if we end up being rejected we still were a THREAD execution and not SEMAPHORE)              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) {                          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);                              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()"));                      }                  }              }).subscribeOn(threadPool.getScheduler(new Func0<Boolean>() {                  @Override                  public Boolean call() {                      return properties.executionIsolationThreadInterruptOnTimeout().get() && _cmd.isCommandTimedOut.get() == TimedOutStatus.TIMED_OUT;                  }              }));          }      }  }

这里就是我们千辛万苦需要找的核心方法了，里面仍然是一个回调函数，通过断点调试，这里会先执行：subscribeOn回调函数，执行threadPool.getScheduler方法，我们进一步往后跟进：

public interface HystrixThreadPool {      @Override      public Scheduler getScheduler(Func0<Boolean> shouldInterruptThread) {          touchConfig();          return new HystrixContextScheduler(HystrixPlugins.getInstance().getConcurrencyStrategy(), this, shouldInterruptThread);      }        private void touchConfig() {          final int dynamicCoreSize = properties.coreSize().get();          final int configuredMaximumSize = properties.maximumSize().get();          int dynamicMaximumSize = properties.actualMaximumSize();          final boolean allowSizesToDiverge = properties.getAllowMaximumSizeToDivergeFromCoreSize().get();          boolean maxTooLow = false;            // 动态调整最大线程池的数量          if (allowSizesToDiverge && configuredMaximumSize < dynamicCoreSize) {              //if user sets maximum < core (or defaults get us there), we need to maintain invariant of core <= maximum              dynamicMaximumSize = dynamicCoreSize;              maxTooLow = true;          }            // In JDK 6, setCorePoolSize and setMaximumPoolSize will execute a lock operation. Avoid them if the pool size is not changed.          if (threadPool.getCorePoolSize() != dynamicCoreSize || (allowSizesToDiverge && threadPool.getMaximumPoolSize() != dynamicMaximumSize)) {              if (maxTooLow) {                  logger.error("Hystrix ThreadPool configuration for : " + metrics.getThreadPoolKey().name() + " is trying to set coreSize = " +                          dynamicCoreSize + " and maximumSize = " + configuredMaximumSize + ".  Maximum size will be set to " +                          dynamicMaximumSize + ", the coreSize value, since it must be equal to or greater than the coreSize value");              }              threadPool.setCorePoolSize(dynamicCoreSize);              threadPool.setMaximumPoolSize(dynamicMaximumSize);          }            threadPool.setKeepAliveTime(properties.keepAliveTimeMinutes().get(), TimeUnit.MINUTES);      }  }    public class HystrixContextScheduler extends Scheduler {      public HystrixContextScheduler(HystrixConcurrencyStrategy concurrencyStrategy, HystrixThreadPool threadPool, Func0<Boolean> shouldInterruptThread) {          this.concurrencyStrategy = concurrencyStrategy;          this.threadPool = threadPool;          this.actualScheduler = new ThreadPoolScheduler(threadPool, shouldInterruptThread);      }        @Override      public Worker createWorker() {          // 构建一个默认的Worker          return new HystrixContextSchedulerWorker(actualScheduler.createWorker());      }        private static class ThreadPoolScheduler extends Scheduler {            private final HystrixThreadPool threadPool;          private final Func0<Boolean> shouldInterruptThread;            public ThreadPoolScheduler(HystrixThreadPool threadPool, Func0<Boolean> shouldInterruptThread) {              this.threadPool = threadPool;              this.shouldInterruptThread = shouldInterruptThread;          }            @Override          public Worker createWorker() {              // 默认的worker为：ThreadPoolWorker              return new ThreadPoolWorker(threadPool, shouldInterruptThread);          }        }        private class HystrixContextSchedulerWorker extends Worker {          // 执行schedule方法          @Override          public Subscription schedule(Action0 action) {              if (threadPool != null) {                  if (!threadPool.isQueueSpaceAvailable()) {                      throw new RejectedExecutionException("Rejected command because thread-pool queueSize is at rejection threshold.");                  }              }              // 默认的worker为：ThreadPoolWorker              return worker.schedule(new HystrixContexSchedulerAction(concurrencyStrategy, action));          }      }          // 执行command的核心类      private static class ThreadPoolWorker extends Worker {            private final HystrixThreadPool threadPool;          private final CompositeSubscription subscription = new CompositeSubscription();          private final Func0<Boolean> shouldInterruptThread;            public ThreadPoolWorker(HystrixThreadPool threadPool, Func0<Boolean> shouldInterruptThread) {              this.threadPool = threadPool;              this.shouldInterruptThread = shouldInterruptThread;          }            @Override          public void unsubscribe() {              subscription.unsubscribe();          }            @Override          public boolean isUnsubscribed() {              return subscription.isUnsubscribed();          }            @Override          public Subscription schedule(final Action0 action) {              if (subscription.isUnsubscribed()) {                  // don't schedule, we are unsubscribed                  return Subscriptions.unsubscribed();              }                // This is internal RxJava API but it is too useful.              ScheduledAction sa = new ScheduledAction(action);                subscription.add(sa);              sa.addParent(subscription);                ThreadPoolExecutor executor = (ThreadPoolExecutor) threadPool.getExecutor();              FutureTask<?> f = (FutureTask<?>) executor.submit(sa);              sa.add(new FutureCompleterWithConfigurableInterrupt(f, shouldInterruptThread, executor));                return sa;          }            @Override          public Subscription schedule(Action0 action, long delayTime, TimeUnit unit) {              throw new IllegalStateException("Hystrix does not support delayed scheduling");          }      }  }

touchConfig() 方法主要是重新设置最大线程池actualMaximumSize的，这里默认的allowMaximumSizeToDivergeFromCoreSize是false。

在HystrixContextScheduler类中有HystrixContextSchedulerWorker、ThreadPoolScheduler、ThreadPoolWorker 这几个内部类。看看它们的作用：

HystrixContextSchedulerWorker: 对外提供schedule()方法，这里会判断线程池队列是否已经满，如果满了这会抛出异常：Rejected command because thread-pool queueSize is at rejection threshold。如果配置的队列大小为-1 则默认返回true。
ThreadPoolScheduler：执行createWorker()方法，默认使用ThreadPoolWorker()类
ThreadPoolWorker：执行command的核心逻辑

private static class ThreadPoolWorker extends Worker {        private final HystrixThreadPool threadPool;      private final CompositeSubscription subscription = new CompositeSubscription();      private final Func0<Boolean> shouldInterruptThread;        @Override      public Subscription schedule(final Action0 action) {          if (subscription.isUnsubscribed()) {              return Subscriptions.unsubscribed();          }            ScheduledAction sa = new ScheduledAction(action);          subscription.add(sa);          sa.addParent(subscription);          // 获取线程池          ThreadPoolExecutor executor = (ThreadPoolExecutor) threadPool.getExecutor();          // 将包装后的HystrixCommand submit到线程池，然后返回FutureTask          FutureTask<?> f = (FutureTask<?>) executor.submit(sa);          sa.add(new FutureCompleterWithConfigurableInterrupt(f, shouldInterruptThread, executor));            return sa;      }  }

原来一个command就是在这里被提交到线程池的，再次回到AbstractCommand.executeCommandWithSpecifiedIsolation()方法中，这里会回调到这个回调函数的call()方法中，这里一路执行逻辑如下：

getUserExecutionObservable(_cmd)==>getExecutionObservable()==>hystrixCommand.run()==>SynchronousMethodHandler.invoke()

这里最后执行到HystrixInvocationHandler中的invoke()方法中的回调函数run()中，最后执行SynchronousMethodHandler.invoke()方法。

一个正常的feign请求，经过hystrix走一遍也就返回对应的response。

总结

上面一顿分析，不知道大家有没有对hystrix 线程池及command执行是否有些理解了？

这个是一个正向流程，没有涉及超时、熔断、降级等代码。关于这些异常降级的源码会在后面一篇文章涉及。

还是之前的建议，大家可以在每个相关的回调函数打上断点，然后一点点调试。

最后再总结一下简单的流程：

浏览器发送请求，执行HystrixTargter
创建HystrixCommand，根据serviceName构造线程池
AbstractCommand中一堆回调函数，最后将command交由线程池submit处理

画一张流程图加深理解：

高清大图：https://www.processon.com/view/link/5e1c128ce4b0169fb51ce77e

申明

本文章首发自本人博客：https://www.cnblogs.com/wang-meng 和公众号：壹枝花算不算浪漫，如若转载请标明来源!

感兴趣的小伙伴可关注个人公众号：壹枝花算不算浪漫