本文作者：陳剛，叩丁狼高級(jí)講師。原創(chuàng)文章，轉(zhuǎn)載請(qǐng)注明出處。

回顧

為了防止服務(wù)之間的調(diào)用異常造成的連鎖反應(yīng)，在SpringCloud中提供了Hystrix組件來實(shí)現(xiàn)服務(wù)調(diào)用異常的處理，或?qū)Ω卟l(fā)情況下的服務(wù)降級(jí)處理 。簡(jiǎn)單回顧一下Hystrix的使用：
1.要使用 Hystrix熔斷機(jī)制處理引入它本身的依賴之外，我們需要在主程序配置類上貼 @EnableHystrix 標(biāo)簽 開啟Hystrix功能，如下

@EnableHystrix
@EnableEurekaClient
@SpringBootApplication
...
public class ConsumerApplication {

2.開啟Hystrix熔斷機(jī)制后，對(duì)方法進(jìn)行熔斷處理

@Service
public class HelloService {

    @Autowired
    private RestTemplate restTemplate;

    //該注解對(duì)該方法創(chuàng)建了熔斷器的功能，并指定了fallbackMethod熔斷方法
    @HystrixCommand(fallbackMethod = "hiError")
    public String hiService(String name){
        //調(diào)用接口進(jìn)行消費(fèi)
        String result = restTemplate.getForObject("http://PRODUCER/hello?name="+name,String.class);
        return result;
    }
    public String hiError(String name) {
        return "hi,"+name+"error!";
    }
}

當(dāng)hiService方法第調(diào)用異常，會(huì)觸發(fā) fallbackMethod執(zhí)行的hiError方法做成一些補(bǔ)救處理。

那么我們就沿著我們的使用方式來跟蹤一下 Hystrix的 工作原理。

首先我們看一下標(biāo)簽：@EnableHystrix ，他的作用從名字就能看出就是開啟Hystrix ，我們看一下它的源碼

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
@EnableCircuitBreaker
public @interface EnableHystrix {

}

它上面有一個(gè)注解：@ EnableCircuitBreaker ，翻譯單詞意思就是啟用熔斷器(斷路器),那么@ EnableHystrix標(biāo)簽的本質(zhì)其實(shí)是@ EnableCircuitBreaker ,我們看一下他的源碼

/**
 * Annotation to enable a CircuitBreaker implementation.
 * http://martinfowler.com/bliki/CircuitBreaker.html
 * @author Spencer Gibb
 */
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
@Import(EnableCircuitBreakerImportSelector.class)
public @interface EnableCircuitBreaker {

}

@EnableCircuitBreaker標(biāo)簽引入了一個(gè)@Import(EnableCircuitBreakerImportSelector.class) 類，翻譯類的名字就是 ， 開啟熔斷器的導(dǎo)入選擇器 ，導(dǎo)入什么東西呢？看源碼

/**
 * Import a single circuit breaker implementation Configuration
 * @author Spencer Gibb
 */
@Order(Ordered.LOWEST_PRECEDENCE - 100)
public class EnableCircuitBreakerImportSelector extends
        SpringFactoryImportSelector<EnableCircuitBreaker> {

    @Override
    protected boolean isEnabled() {
        return getEnvironment().getProperty(
                "spring.cloud.circuit.breaker.enabled", Boolean.class, Boolean.TRUE);
    }

}

翻譯類上的注釋 “Import a single circuit breaker implementation Configuration”，其實(shí)EnableCircuitBreakerImportSelector的作用就是去導(dǎo)入熔斷器的配置 。其實(shí)Spring中也有類似于JAVA SPI 的加載機(jī)制, 即會(huì)自動(dòng)加載 jar包 spring-cloud-netflix-core 中的META-INF/spring.factories 中的Hystrix相關(guān)的自動(dòng)配置類
注：SPI : 通過將服務(wù)的接口與實(shí)現(xiàn)分離以實(shí)現(xiàn)解耦，提高程序拓展性的機(jī)制，達(dá)到插拔式的效果 。

image.png

HystrixCircuitBreakerConfiguration 就是針對(duì)于 Hystrix熔斷器的配置

/**
 * @author Spencer Gibb
 * @author Christian Dupuis
 * @author Venil Noronha
 */
@Configuration
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));
    }
......

在該配置類中創(chuàng)建了 HystrixCommandAspect

/**
 * AspectJ aspect to process methods which annotated with {@link HystrixCommand} annotation.
 */
@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();
    }

//定義切點(diǎn)，切到 @HystrixCommand標(biāo)簽所在的方法  
  @Pointcut("@annotation(com.netflix.hystrix.contrib.javanica.annotation.HystrixCommand)")

    public void hystrixCommandAnnotationPointcut() {
    }

    @Pointcut("@annotation(com.netflix.hystrix.contrib.javanica.annotation.HystrixCollapser)")
    public void hystrixCollapserAnnotationPointcut() {
    }
//針對(duì)切點(diǎn)：@hystrixCommand切點(diǎn)的處理
    @Around("hystrixCommandAnnotationPointcut() || hystrixCollapserAnnotationPointcut()")
    public Object methodsAnnotatedWithHystrixCommand(final ProceedingJoinPoint joinPoint) throws Throwable {
//獲取到目標(biāo)方法
  Method method = getMethodFromTarget(joinPoint);
        Validate.notNull(method, "failed to get method from joinPoint: %s", joinPoint);
//判斷方法上不能同時(shí)存在@HystrixCommand標(biāo)簽和HystrixCollapser標(biāo)簽
        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));
        MetaHolder metaHolder = metaHolderFactory.create(joinPoint);
//把方法封裝成 HystrixInvokable 
        HystrixInvokable invokable = HystrixCommandFactory.getInstance().create(metaHolder);
        ExecutionType executionType = metaHolder.isCollapserAnnotationPresent() ?
                metaHolder.getCollapserExecutionType() : metaHolder.getExecutionType();

        Object result;
        try {
// 通過CommandExecutor來執(zhí)行方法
            if (!metaHolder.isObservable()) {
                result = CommandExecutor.execute(invokable, executionType, metaHolder);
            } else {
                result = executeObservable(invokable, executionType, metaHolder);
            }
        } catch (HystrixBadRequestException e) {
            throw e.getCause() != null ? e.getCause() : e;
        } catch (HystrixRuntimeException e) {
            throw hystrixRuntimeExceptionToThrowable(metaHolder, e);
        }
        return result;

HystrixCommandAspect 其實(shí)就是對(duì) 貼了@HystrixCommand標(biāo)簽的方法使用 Aop機(jī)制實(shí)現(xiàn)處理 。代碼中通過把目標(biāo)方法封裝成 HystrixInvokable對(duì)象，通過CommandExecutor工具來執(zhí)行目標(biāo)方法。

HystrixInvokable是用來干嘛的？看源碼知道，其實(shí)他是一個(gè)空行法的接口，他的目的只是用來標(biāo)記可被執(zhí)行，那么他是如何創(chuàng)建的我們看代碼HystrixInvokable invokable = HystrixCommandFactory.getInstance().create(metaHolder);的create方法

   public HystrixInvokable create(MetaHolder metaHolder) {
        HystrixInvokable executable;
        ...省略代碼...
            executable = new GenericCommand(HystrixCommandBuilderFactory.getInstance().create(metaHolder));
        }
        return executable;
    }

其實(shí)是new了一個(gè) GenericCommand 對(duì)象，很明顯他們是實(shí)現(xiàn)關(guān)系，我們看一下關(guān)系圖

叩丁狼教育.png

跟蹤 GenericCommand 的源碼

@ThreadSafe
public class GenericCommand extends AbstractHystrixCommand<Object> {
    private static final Logger LOGGER = LoggerFactory.getLogger(GenericCommand.class);

    public GenericCommand(HystrixCommandBuilder builder) {
        super(builder);
    }

    protected Object run() throws Exception {
        LOGGER.debug("execute command: {}", this.getCommandKey().name());
        return this.process(new AbstractHystrixCommand<Object>.Action() {
            Object execute() {
                return GenericCommand.this.getCommandAction().execute(GenericCommand.this.getExecutionType());
            }
        });
    }

    protected Object getFallback() {
        final CommandAction commandAction = this.getFallbackAction();
        if (commandAction != null) {
            try {
                return this.process(new AbstractHystrixCommand<Object>.Action() {
                    Object execute() {
                        MetaHolder metaHolder = commandAction.getMetaHolder();
                        Object[] args = CommonUtils.createArgsForFallback(metaHolder, GenericCommand.this.getExecutionException());
                        return commandAction.executeWithArgs(metaHolder.getFallbackExecutionType(), args);
                    }
                });
            } catch (Throwable var3) {
                LOGGER.error(FallbackErrorMessageBuilder.create().append(commandAction, var3).build());
                throw new FallbackInvocationException(ExceptionUtils.unwrapCause(var3));
            }
        } else {
            return super.getFallback();
        }
    }
}

它本身對(duì)目標(biāo)方法的正常執(zhí)行和對(duì) fallback方法的 執(zhí)行做了實(shí)現(xiàn) 。
GenericCommand.this.getCommandAction().execute(...)獲取到目標(biāo)方法并執(zhí)行，底層會(huì)交給 MethodExecutionAction 使用反射去執(zhí)行方法，

回到 HystrixCommandAspect的methodsAnnotatedWithHystrixCommand方法中，我們看下 CommandExecutor.execute是如何執(zhí)行的

public class CommandExecutor {
    public CommandExecutor() {
    }

    public static Object execute(HystrixInvokable invokable, ExecutionType executionType, MetaHolder metaHolder) throws RuntimeException {
        Validate.notNull(invokable);
        Validate.notNull(metaHolder);
        switch(executionType) {
//異步
        case SYNCHRONOUS:
            return castToExecutable(invokable, executionType).execute();
//同步
        case ASYNCHRONOUS:
            HystrixExecutable executable = castToExecutable(invokable, executionType);
            if (metaHolder.hasFallbackMethodCommand() && ExecutionType.ASYNCHRONOUS == metaHolder.getFallbackExecutionType()) {
                return new FutureDecorator(executable.queue());
            }

            return executable.queue();
        case OBSERVABLE:
            HystrixObservable observable = castToObservable(invokable);
            return ObservableExecutionMode.EAGER == metaHolder.getObservableExecutionMode() ? observable.observe() : observable.toObservable();
        default:
            throw new RuntimeException("unsupported execution type: " + executionType);
        }
    }

    private static HystrixExecutable castToExecutable(HystrixInvokable invokable, ExecutionType executionType) {
        if (invokable instanceof HystrixExecutable) {
            return (HystrixExecutable)invokable;
        } else {
            throw new RuntimeException("Command should implement " + HystrixExecutable.class.getCanonicalName() + " interface to execute in: " + executionType + " mode");
        }
    }

這里有兩種執(zhí)行方式 SYNCHRONOUS 異步 ，ASYNCHRONOUS同步 ，我們先看異步： castToExecutable(invokable, executionType).execute(); 這里代碼把HystrixInvokable對(duì)象轉(zhuǎn)成 HystrixExecutable并調(diào)用execute方法執(zhí)行 ，跟蹤execute方法進(jìn)入HystrixCommand.execute方法中

 public R execute() {
        try {
            return queue().get();
        } catch (Exception e) {
            throw Exceptions.sneakyThrow(decomposeException(e));
        }
    }
--------------
 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) {
                if (delegate.isCancelled()) {
                    return false;
                }

                if (HystrixCommand.this.getProperties().executionIsolationThreadInterruptOnFutureCancel().get()) {
                    /*
                     * The only valid transition here is false -> true. If there are two futures, say f1 and f2, created by this command
                     * (which is super-weird, but has never been prohibited), and calls to f1.cancel(true) and to f2.cancel(false) are
                     * issued by different threads, it's unclear about what value would be used by the time mayInterruptOnCancel is checked.
                     * The most consistent way to deal with this scenario is to say that if *any* cancellation is invoked with interruption,
                     * than that interruption request cannot be taken back.
                     */
                    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;
            }
....省略...

在 HystrixCommand.execute方法中 其實(shí)是Future 來異步執(zhí)行，調(diào)用過程中會(huì)觸發(fā) GenericCommand來完成調(diào)用，執(zhí)行完成后調(diào)用 Future.get()方法拿到執(zhí)行結(jié)果 。

想獲取更多技術(shù)干貨，請(qǐng)前往叩丁狼官網(wǎng)：http://www.wolfcode.cn/all_article.html