服务保障框架-Hystrix快速入门

//Gradle中添加Hystrix核心
compile('org.springframework.cloud:spring-cloud-starter-netflix-hystrix')

@Configuration
public class HystrixConfiguration {
  @Bean
  public HystrixCommandAspect hystrixAspect() {
    return new HystrixCommandAspect();
  }
}

@DefaultProperties(groupKey = "UserQueryGroup", threadPoolProperties = {
        @HystrixProperty(name = "coreSize", value = "20"),
        @HystrixProperty(name = "maxQueueSize", value = "20")
})
@Service
public class UserQueryManager {
    @HystrixCommand(commandKey = "GetUserCommand",fallbackMethod = "getUserFallback")
    public String getUser(long id) {
        return "test_user";
    }
    
    public String getUserFallback(long id) {
        return "test_user_fallback";
    }
}

//Gradle中添加Hystrix面板
compile('org.springframework.cloud:spring-cloud-starter-netflix-hystrix-dashboard')

@Bean
public ServletRegistrationBean getServlet() {
    HystrixMetricsStreamServlet streamServlet = new HystrixMetricsStreamServlet();
    ServletRegistrationBean registrationBean = new ServletRegistrationBean(streamServlet);
    registrationBean.setLoadOnStartup(1);
    registrationBean.addUrlMappings("/hystrix.stream");
    registrationBean.setName("HystrixMetricsStreamServlet");
    return registrationBean;
}

@Around("hystrixCommandAnnotationPointcut() || hystrixCollapserAnnotationPointcut()")
public Object methodsAnnotatedWithHystrixCommand(final ProceedingJoinPoint joinPoint) throws Throwable {
    Method method = getMethodFromTarget(joinPoint);
    ...
    MetaHolderFactory metaHolderFactory = META_HOLDER_FACTORY_MAP.get(HystrixPointcutType.of(method));
    MetaHolder metaHolder = metaHolderFactory.create(joinPoint);
    HystrixInvokable invokable = HystrixCommandFactory.getInstance().create(metaHolder);//1.创建metaHolder
    ExecutionType executionType = metaHolder.isCollapserAnnotationPresent() ?
            metaHolder.getCollapserExecutionType() : metaHolder.getExecutionType();
    Object result;
    try {//2.命令执行
        if (!metaHolder.isObservable()) {
            result = CommandExecutor.execute(invokable, executionType, metaHolder);
        } else {
            result = executeObservable(invokable, executionType, metaHolder);
        }
    } ...
}

public static Object execute(HystrixInvokable invokable, ExecutionType executionType, MetaHolder metaHolder) throws RuntimeException {
    ...
    switch (executionType) {
        case SYNCHRONOUS: {//1.同步执行，其实其内部也是用的异步执行queue().get()
            return castToExecutable(invokable, executionType).execute();
        }
        case ASYNCHRONOUS: {//2.异步执行
            HystrixExecutable executable = castToExecutable(invokable, executionType);
            if (metaHolder.hasFallbackMethodCommand()
                    && ExecutionType.ASYNCHRONOUS == metaHolder.getFallbackExecutionType()) {
                return new FutureDecorator(executable.queue());
            }
            return executable.queue();
        }
        case OBSERVABLE: {//3.响应式执行，observable.toObservable()是核心方法
            HystrixObservable observable = castToObservable(invokable);
            return ObservableExecutionMode.EAGER == metaHolder.getObservableExecutionMode() ? observable.observe() : observable.toObservable();
        }
      ...
}

public Observable<R> toObservable() {
   ...
    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);//1.关键步骤，命令处理
        }
    };
  ...
  
private Observable<R> applyHystrixSemantics(final AbstractCommand<R> _cmd) {
    ...
    if (circuitBreaker.attemptExecution()) {//1.【断路器相关处理】，之后HystrixCircuitBreaker中展示
        ..
        if (executionSemaphore.tryAcquire()) {//2.获取信号量，如果是THREAD线程池策略，【直接返回true】，这里需要注意，不然流程将进行不下去
            try {
                executionResult = executionResult.setInvocationStartTime(System.currentTimeMillis());
                return executeCommandAndObserve(_cmd)//3.核心执行方法
                        .doOnError(markExceptionThrown)
                        .doOnTerminate(singleSemaphoreRelease)
                        .doOnUnsubscribe(singleSemaphoreRelease);
            } ...
}

private Observable<R> executeCommandAndObserve(final AbstractCommand<R> _cmd) {
    ...
    final Func1<Throwable, Observable<R>> handleFallback = new Func1<Throwable, Observable<R>>() {
        @Override
        public Observable<R> call(Throwable t) {
            circuitBreaker.markNonSuccess();
            Exception e = getExceptionFromThrowable(t);
            executionResult = executionResult.setExecutionException(e);
            if (e instanceof RejectedExecutionException) {
                return handleThreadPoolRejectionViaFallback(e);//1.处理线程池拒绝场景
            } else if (t instanceof HystrixTimeoutException) {
                return handleTimeoutViaFallback();//2.处理请求超时场景
            } 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);//3.处理请求失败场景
            }
        }
    };

@Override
public boolean attemptExecution() {
    if (properties.circuitBreakerForceOpen().get()) {//1.断路器是否强制打开
        return false;
    }
    if (properties.circuitBreakerForceClosed().get()) {//2.断路器是否强制关闭
        return true;
    }
    if (circuitOpened.get() == -1) {//3.短路器关闭
        return true;
    } else {
        if (isAfterSleepWindow()) {//4.是否经过了指定的窗口时间
            if (status.compareAndSet(Status.OPEN, Status.HALF_OPEN)) {//5.设置为半打开状态
                //only the first request after sleep window should execute
                return true;
            } else {
                return false;
            }
        } else {
            return false;
        }
    }
}