[享学Netflix] 三十、Hystrix的fallback回退/降级逻辑源码解读

AbstractCommand：

	// _cmd：命令对象
	// eventType：事件类型
	// failureType：失败类型枚举（BAD_REQUEST_EXCEPTION...） 见下面枚举定义
	// message：失败的消息。如timed-out、failed、short-circuited等
	// Exception：导致失败的异常（一定只有异常才能导致失败），如java.util.concurrent.TimeoutException
	private Observable<R> getFallbackOrThrowException(
		AbstractCommand<R> _cmd, HystrixEventType eventType, 
		FailureType failureType, String message, 
		Exception originalException) {
	
		// 因为fallabck也是在线程池里面执行，所以也是需要传递数据的
		HystrixRequestContext requestContext = HystrixRequestContext.getContextForCurrentThread();
		// 统计相关事件的次数：详见事件计数器
		executionResult = executionResult.addEvent((int) latency, eventType);
		...
		// 如果异常类型是ExceptionNotWrappedByHystrix类型
		if (shouldNotBeWrapped(originalException)){
			... // 触发executionHook.onError()动作
			return Observable.error(e);
		// 是否是不可恢复的异常类型：比如StackOverflowError/VirtualMachineError...
		// 因为并不是所有的Error都不会恢复，所以这里例举出来。如果是不可恢复的错误，就包装一下抛出
		} else if (isUnrecoverable(originalException)) {
			...
			// 若是不可恢复错误，转未HystrixRuntimeException异常抛出
			return Observable.error(new HystrixRuntimeException(failureType, ... );
		} else {
			// 若是可自己恢复的Error，如IOError，那就输入一句日志即可
			if (isRecoverableError(originalException)) { ... }

			// =======普通异常类型（比如NPE之类的），那就开始执行fallabck函数========
			if (properties.fallbackEnabled().get()) { // 显然默认是开启的，不建议关闭它
				// 若你没指定，默认使用的是TryableSemaphoreActual 10个信号量
				// 你可以通过fallbackIsolationSemaphoreMaxConcurrentRequests指定这个值
				// 该信号量用于杜绝你的fallabck还非常耗时的case，以防万一
				TryableSemaphore fallbackSemaphore = getFallbackSemaphore();

				...
				// 最终return的可观察对象
				// getFallbackObservable是个抽象方法，由子类提供
				// 比如command它就是使用Observable.just(getFallback())把任意对象转换为Observable
				// 当然它是个defer的Observable
				Observable<R> fallbackExecutionChain;
				// 若还有信号量资源就继续执行（否则会会做异常处理，见下面）
				if (fallbackSemaphore.tryAcquire()) { 
					// 用户是否自定义了fallabck函数
					// 也就是是否重写了getFallabck()方法嘛
					if (isFallbackUserDefined()) { 
						... // executionHook.onFallbackStart(this);
						fallbackExecutionChain = getFallbackObservable();
					} else { // 区别是：若没有复写这个方法，就不会触发onFallbackStart()动作
						fallbackExecutionChain = getFallbackObservable();
					}
					// 绑定固定的处理函数
					return fallbackExecutionChain
								// 作用于每个iter身上：setRequestContextIfNeeded(requestContext);
								// 确保上下文是同一个，数据是通的
			                    .doOnEach(setRequestContext)
			                    // 主要是为了触发executionHook相关回调
			                    .lift(new FallbackHookApplication(_cmd))
			                    .lift(new DeprecatedOnFallbackHookApplication(_cmd))
			                    // 发布事件：HystrixEventType.FALLBACK_EMIT
			                    // executionResult.addEvent(HystrixEventType.FALLBACK_EMIT)
			                    .doOnNext(markFallbackEmit)
			                    // 完成了。发布事件：HystrixEventType.FALLBACK_SUCCESS
			                    // 证明fallabck成功
			                    .doOnCompleted(markFallbackCompleted)
			                    // fallabck失败（出现异常了），发布事件
			                    // HystrixEventType.FALLBACK_MISSING、FALLBACK_FAILURE
			                    .onErrorResumeNext(handleFallbackError)
			                    // 释放信号量：fallbackSemaphore.release();
			                    .doOnTerminate(singleSemaphoreRelease)
			                    // fallbackSemaphore.release();
			                    .doOnUnsubscribe(singleSemaphoreRelease);
				} else { // 此else对应逻辑：信号量不够的情况
					return handleFallbackRejectionByEmittingError();
				}
			} else { // 此else对应逻辑：properties禁用了fallabck的情况
				return handleFallbackDisabledByEmittingError(originalException, failureType, message);
			}
		}
	}


// 定义失败枚举：一个7种类型
HystrixRuntimeException：
    public static enum FailureType {
        BAD_REQUEST_EXCEPTION, COMMAND_EXCEPTION, TIMEOUT, SHORTCIRCUIT, REJECTED_THREAD_EXECUTION, REJECTED_SEMAPHORE_EXECUTION, REJECTED_SEMAPHORE_FALLBACK
    }

	// 若异常类型是该类型（比如你使用HystrixCommand，但没重写getFallabck()方法，执行就抛出此异常）
	// 就包装为HystrixRuntimeException给你抛出
	// 并且有你熟悉的抛错消息： message + " and no fallback available."
	if (fe instanceof UnsupportedOperationException) {
		...
		// 记录结果：事件类型是FALLBACK_MISSING
		executionResult.addEvent((int) latency, HystrixEventType.FALLBACK_MISSING);
		return Observable.error(new HystrixRuntimeException(failureType, ...);
	} else { // 你的fallabck方法里抛出了其它异常
		...
		// 记录事件，此时事件类型是Fallabck执行失败FALLBACK_FAILURE
		executionResult.addEvent((int) latency, HystrixEventType.FALLBACK_FAILURE);
		return Observable.error(new HystrixRuntimeException(failureType ...  message + " and fallback failed." ...);
	}

AbstractCommand：

	private Observable<R> handleFallbackDisabledByEmittingError(Exception underlying, FailureType failureType, String message) {
		... // 触发executionHook.onError动作
		// 转换为一个HystrixRuntimeException抛出
		return Observable.error(new HystrixRuntimeException(failureType, ... " and fallback disabled." ... );
	}

AbstractCommand：

	private Observable<R> handleFallbackRejectionByEmittingError() {
		...
		// 这里记录了结果哦。统计对应事件的次数(毕竟出现此case的数据还是蛮有意义的)
		executionResult = executionResult.addEvent((int) latencyWithFallback, HystrixEventType.FALLBACK_REJECTION);
		return Observable.error(new HystrixRuntimeException(FailureType.REJECTED_SEMAPHORE_FALLBACK,... " fallback execution rejected." ...);
    }