【SpringCloud】Hystrix源码解析

hystrix是一个微服务容错组件，提供了资源隔离、服务降级、服务熔断的功能。这一章重点分析hystrix的实现原理

1、服务降级

CAP原则是分布式系统的一个理论基础，它的三个关键属性分别是一致性、可用性和容错性。当服务实例所在服务器承受过大的压力或者受到网络因素影响没法及时响应请求时，整个任务将处于阻塞状态，这样的系统容错性不高，稍有不慎就会陷入瘫痪，hystrix为此提供了一种容错机制：当服务实例没法及时响应请求，可以采用服务降级的方式快速失败，维持系统的稳定性

服务降级和@HystrixCommand注解绑定，查看它的源码

@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
public @interface HystrixCommand {

    ...

    String fallbackMethod() default "";

}

源码提供的信息很少，想要分析注解的功能，还得找到处理注解信息的类：HystrixCommandAspect

@Aspect
public class HystrixCommandAspect {
    ...
    
    // 环绕通知
    @Around("hystrixCommandAnnotationPointcut() || hystrixCollapserAnnotationPointcut()")
    public Object methodsAnnotatedWithHystrixCommand(ProceedingJoinPoint joinPoint) throws Throwable {
        Method method = AopUtils.getMethodFromTarget(joinPoint);
        Validate.notNull(method, "failed to get method from joinPoint: %s", new Object[]{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");
        } else {
            MetaHolderFactory metaHolderFactory = (MetaHolderFactory)META_HOLDER_FACTORY_MAP.get(HystrixCommandAspect.HystrixPointcutType.of(method));
            MetaHolder metaHolder = metaHolderFactory.create(joinPoint);
            HystrixInvokable invokable = HystrixCommandFactory.getInstance().create(metaHolder);
            ExecutionType executionType = metaHolder.isCollapserAnnotationPresent() ? metaHolder.getCollapserExecutionType() : metaHolder.getExecutionType();

            try {
                Object result;
                if (!metaHolder.isObservable()) {
                    // 代理执行方法
                    result = CommandExecutor.execute(invokable, executionType, metaHolder);
                } else {
                    result = this.executeObservable(invokable, executionType, metaHolder);
                }

                return result;
            } catch (HystrixBadRequestException var9) {
                throw var9.getCause();
            } catch (HystrixRuntimeException var10) {
                throw this.hystrixRuntimeExceptionToThrowable(metaHolder, var10);
            }
        }
    }
}

从命名上我们能看出这是一个切面，说明服务降级是通过aop代理实现的，跟踪CommandExecutor的execute方法

调用链：
-> CommandExecutor.execute
-> castToExecutable(invokable, executionType).execute()
-> HystrixCommand.execute
-> this.queue().get()

public Future<R> queue() {

	// 获取Future对象
	final Future<R> delegate = this.toObservable().toBlocking().toFuture();
	Future<R> f = new Future<R>() {
		...

		public R get() throws InterruptedException, ExecutionException {
			return delegate.get();
		}

		public R get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
			return delegate.get(timeout, unit);
		}
	};
	...
}

HystrixCommand类的queue方法返回了一个Future对象，在线程任务中常用Future对象来获取任务执行的结果。这里的Future对象是通过this.toObservable().toBlocking().toFuture()创建的，点击查看toObservable方法，它返回一个Observable对象

public Observable<R> toObservable() {
   ...
   final Func0<Observable<R>> applyHystrixSemantics = new Func0<Observable<R>>() {
        public Observable<R> call() {
              return 
((CommandState)AbstractCommand.this.commandState.get()).equals(AbstractCommand.CommandState.UNSUBSCRIBED) ? Observable.never() : 
              // 传入指令执行任务
              AbstractCommand.this.applyHystrixSemantics(AbstractCommand.this);
        }
   };
   ...
	return Observable.defer(new Func0<Observable<R>>() {
		public Observable<R> call() {
                ...
                // 有订阅者订阅了才创建Observable对象
				Observable<R> hystrixObservable = 
                Observable.defer(applyHystrixSemantics).map(wrapWithAllOnNextHooks);
				Observable afterCache;
				if (requestCacheEnabled && cacheKey != null) {
					HystrixCachedObservable<R> toCache = HystrixCachedObservable.from(hystrixObservable, AbstractCommand.this);
					HystrixCommandResponseFromCache<R> fromCache = (HystrixCommandResponseFromCache)AbstractCommand.this.requestCache.putIfAbsent(cacheKey, toCache);
					if (fromCache != null) {
						toCache.unsubscribe();
						AbstractCommand.this.isResponseFromCache = true;
						return AbstractCommand.this.handleRequestCacheHitAndEmitValues(fromCache, AbstractCommand.this);
					}

					afterCache = toCache.toObservable();
				} else {
					afterCache = hystrixObservable;
				}

				return afterCache.doOnTerminate(terminateCommandCleanup).doOnUnsubscribe(unsubscribeCommandCleanup).doOnCompleted(fireOnCompletedHook);
			...
		}
	});        
}

Observable对象的创建任务委托了给了AbstractCommand.this.applyHystrixSemantics方法

private Observable<R> applyHystrixSemantics(AbstractCommand<R> _cmd) {
	this.executionHook.onStart(_cmd);
	// 是否允许请求，判断熔断状态
	if (this.circuitBreaker.allowRequest()) {
		final TryableSemaphore executionSemaphore = this.getExecutionSemaphore();
		final AtomicBoolean semaphoreHasBeenReleased = new AtomicBoolean(false);
		Action0 singleSemaphoreRelease = new Action0() {
			public void call() {
				if (semaphoreHasBeenReleased.compareAndSet(false, true)) {
					executionSemaphore.release();
				}

			}
		};
		Action1<Throwable> markExceptionThrown = new Action1<Throwable>() {
			public void call(Throwable t) {
				AbstractCommand.this.eventNotifier.markEvent(HystrixEventType.EXCEPTION_THROWN, AbstractCommand.this.commandKey);
			}
		};
		if (executionSemaphore.tryAcquire()) {
			try {
				this.executionResult = this.executionResult.setInvocationStartTime(System.currentTimeMillis());
				// 执行任务
				return this.executeCommandAndObserve(_cmd).doOnError(markExceptionThrown).doOnTerminate(singleSemaphoreRelease).doOnUnsubscribe(singleSemaphoreRelease);
			} catch (RuntimeException var7) {
				return Observable.error(var7);
			}
		} else {
			return this.handleSemaphoreRejectionViaFallback();
		}
	} else {
		// 处于熔断状态，执行备用任务
		return this.handleShortCircuitViaFallback();
	}
}

this.circuitBreaker.allowReques返回true表示没有熔断，走executeCommandAndObserve方法

private Observable<R> executeCommandAndObserve(AbstractCommand<R> _cmd) {
	...
	Observable execution;
	if ((Boolean)this.properties.executionTimeoutEnabled().get()) {
		// 添加了超时监控
		execution = this.executeCommandWithSpecifiedIsolation(_cmd).lift(new HystrixObservableTimeoutOperator(_cmd));
	} else {
		execution = this.executeCommandWithSpecifiedIsolation(_cmd);
	}

   ...
	// handleFallback：不同异常状况下使用不同的处理方法
	Func1<Throwable, Observable<R>> handleFallback = new Func1<Throwable, Observable<R>>() {
		public Observable<R> call(Throwable t) {
			Exception e = AbstractCommand.this.getExceptionFromThrowable(t);
			AbstractCommand.this.executionResult = AbstractCommand.this.executionResult.setExecutionException(e);
			if (e instanceof RejectedExecutionException) {
				return AbstractCommand.this.handleThreadPoolRejectionViaFallback(e);
			} else if (t instanceof HystrixTimeoutException) {
				// 抛出超时异常时，做超时处理
				return AbstractCommand.this.handleTimeoutViaFallback();
			} else if (t instanceof HystrixBadRequestException) {
				return AbstractCommand.this.handleBadRequestByEmittingError(e);
			} else if (e instanceof HystrixBadRequestException) {
				AbstractCommand.this.eventNotifier.markEvent(HystrixEventType.BAD_REQUEST, AbstractCommand.this.commandKey);
				return Observable.error(e);
			} else {
				return AbstractCommand.this.handleFailureViaFallback(e);
			}
		}
	};
   ...

	return execution.doOnNext(markEmits).doOnCompleted(markOnCompleted)
		  // 调用handleFallback处理异常
		  .onErrorResumeNext(handleFallback).doOnEach(setRequestContext);
}

private static class HystrixObservableTimeoutOperator<R> implements Observable.Operator<R, R> {
        final AbstractCommand<R> originalCommand;

        public HystrixObservableTimeoutOperator(AbstractCommand<R> originalCommand) {
            this.originalCommand = originalCommand;
        }

        public Subscriber<? super R> call(final Subscriber<? super R> child) {
            final CompositeSubscription s = new CompositeSubscription();
            child.add(s);
            final HystrixContextRunnable timeoutRunnable = new HystrixContextRunnable(this.originalCommand.concurrencyStrategy, new Runnable() {
                public void run() {
                    // 3.抛出超时异常
                    child.onError(new HystrixTimeoutException());
                }
            });
            HystrixTimer.TimerListener listener = new HystrixTimer.TimerListener() {
                // 1.判断是否超时
                public void tick() {
                    if (HystrixObservableTimeoutOperator.this.originalCommand.isCommandTimedOut.compareAndSet(AbstractCommand.TimedOutStatus.NOT_EXECUTED, AbstractCommand.TimedOutStatus.TIMED_OUT)) {
                        HystrixObservableTimeoutOperator.this.originalCommand.eventNotifier.markEvent(HystrixEventType.TIMEOUT, HystrixObservableTimeoutOperator.this.originalCommand.commandKey);
                        s.unsubscribe();
                        // 2.执行超时任务
                        timeoutRunnable.run();
                    }
                }
            };
            
        }
    }

executeCommandAndObserve方法添加超时监控，如果任务执行超出限制时间会抛出超时异常，由handleTimeoutViaFallback方法处理异常

private Observable<R> handleTimeoutViaFallback() {
	// 1.根据异常类型处理异常
	return this.getFallbackOrThrowException(this, HystrixEventType.TIMEOUT, FailureType.TIMEOUT, "timed-out", new TimeoutException());
}

private Observable<R> getFallbackOrThrowException(final AbstractCommand<R> _cmd, HystrixEventType eventType, final HystrixRuntimeException.FailureType failureType, final String message, final Exception originalException) {
	...
	// 获取回调观察者
	fallbackExecutionChain = this.getFallbackObservable();
	...
}    

protected final Observable<R> getFallbackObservable() {
	return Observable.defer(new Func0<Observable<R>>() {
		public Observable<R> call() {
			try {
				// 执行备用方法
				return Observable.just(HystrixCommand.this.getFallback());
			} catch (Throwable var2) {
				return Observable.error(var2);
			}
		}
	});
}

到这里终于看到了getFallback方法，它会调用注解中fallback指向的方法，快速失败返回响应结果

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();
	}
}

回到AbstractCommand.this.applyHystrixSemantics方法，当this.circuitBreaker.allowReques返回true是请求正常往下走，当它返回false时表示服务进入熔断状态，会走else分支，同样会进入getFallback方法

调用链
-> AbstractCommand.handleShortCircuitViaFallback
-> getFallbackOrThrowException
-> this.getFallbackObservable
-> GenericCommand.getFallback

2、服务熔断

服务熔断是hystrix提供的一种保护机制，当一段时间内服务响应的异常的次数过多，hystrix会让服务降级快速返回失败信息，避免累积压力造成服务崩溃。
联系上文找到circuitBreaker.allowRequest方法，该方法判断是否允许请求往下走

public boolean allowRequest() {
	// 是否强制打开熔断
	if ((Boolean)this.properties.circuitBreakerForceOpen().get()) {
		return false;
	// 是否强制关闭熔断
	} else if ((Boolean)this.properties.circuitBreakerForceClosed().get()) {
		this.isOpen();
		return true;
	} else {
		return !this.isOpen() || this.allowSingleTest();
	}
}

public boolean isOpen() {
	if (this.circuitOpen.get()) {
		return true;
	} else {
		HystrixCommandMetrics.HealthCounts health = this.metrics.getHealthCounts();
		// 请求次数是否超过单位时间内请求数阈值
		if (health.getTotalRequests() < (long)(Integer)this.properties.circuitBreakerRequestVolumeThreshold().get()) {
			return false;
		// 请求异常次数占比
		} else if (health.getErrorPercentage() < (Integer)this.properties.circuitBreakerErrorThresholdPercentage().get()) {
			return false;
		} else if (this.circuitOpen.compareAndSet(false, true)) {
			this.circuitOpenedOrLastTestedTime.set(System.currentTimeMillis());
			return true;
		} else {
			return true;
		}
	}
}

isOpen方法内有针对请求的各种量化计算，当请求异常情况过多，就会触发熔断，走服务降级

3、总结

hystrix组件会根据请求状态判断是否执行请求，当请求超时或者存在其他异常会走备用方法，当异常次数过多会进入熔断状态快速失败，避免服务累积过多压力