[享学Netflix] 三十五、Hystrix执行过程集大成者：AbstractCommand详解

abstract class AbstractCommand<R> implements HystrixInvokableInfo<R>, HystrixObservable<R> {
	
	// command的id
	protected final HystrixCommandKey commandKey; 
	// 线程池分组名（理论上不同的Command可以共用一个线程池，节约资源嘛）
    protected final HystrixThreadPoolKey threadPoolKey; 
    // 逻辑分组。用于统计
    protected final HystrixCommandGroupKey commandGroup;

	// 各种properties配置  均可以通过SPI方式提供
	protected final HystrixCommandProperties properties;	
	
	// SPI接口（这几个接口有详细介绍，不陌生）
    protected final HystrixEventNotifier eventNotifier;
    protected final HystrixConcurrencyStrategy concurrencyStrategy;
    protected final HystrixCommandExecutionHook executionHook;

	// 熔断器。若你使用配置显示enable=false了
	// 其实现就是NoOpCircuitBreaker，否则就是默认实现
    protected final HystrixCircuitBreaker circuitBreaker;
    
    // 线程池。默认实现是HystrixThreadPoolDefault
    // 线程池参数使用HystrixThreadPoolProperties配置
    // 通过HystrixConcurrencyStrategy#getThreadPool()得到ThreadPoolExecutor执行器
    // 说明：每次getThreadPool()一下都会用最新的配置配置执行器。所以可以达到动态化的目的
    // 比如说CorePoolSize、MaximumPoolSize等等都可以在properties里动态改变
    protected final HystrixThreadPool threadPool;
	// Command指标收集器
	protected final HystrixCommandMetrics metrics;

	// 执行时候时使用的信号量（每个key一个信号量控制哦）
	// 说明：它并没有使用JDK的java.util.concurrent.Semaphore，而是自己的实现
	// 信号量的实现非常简单，所以就略喽。
	// 当没开启信号量隔离的时候，该实现类使用的是TryableSemaphoreNoOp
	protected final TryableSemaphore executionSemaphoreOverride;
	// 发生fallabck时的信号量，也是每个key一个。至于fallback都需要信号量隔离，前面有详细说明
	// 它哥俩均只有在隔离策略是SEMAPHORE才有效。它哥俩信号量默认值都是10
	protected final TryableSemaphore executionSemaphoreOverride;
 
 	// 各种状态值（这里也说明command是有状态的：一个实例只能执行一次）
    protected AtomicReference<CommandState> commandState = new AtomicReference<>(CommandState.NOT_STARTED);
    protected AtomicReference<ThreadState> threadState = new AtomicReference<>(ThreadState.NOT_USING_THREAD);
    protected final AtomicReference<TimedOutStatus> isCommandTimedOut = new AtomicReference<>(TimedOutStatus.NOT_EXECUTED);

	// 执行结果
	protected volatile ExecutionResult executionResult = ExecutionResult.EMPTY;
	// 取消时的执行结果
	protected volatile ExecutionResult executionResultAtTimeOfCancellation;
	
	// 表示为：响应是否来自于缓存
	// 若缓存开启了，并且请求时缓存命中了，那此值就会被置为true
	protected volatile boolean isResponseFromCache = false;
	// command开始执行的时刻（并不代表一定会执行到目标方法哦）
	protected volatile long commandStartTimestamp = -1L;

	// 缓存、日志相关
	protected final HystrixRequestCache requestCache;
	protected final HystrixRequestLog currentRequestLog;


	// 缓存默认名称（官方数据，加了这个缓存后效率提升了1-2微秒）
	// 默认的key名使用getSimpleName()，简单类名
	// 但若你没有简单类名（比如内部类），那就使用全类名getName()
	private static ConcurrentHashMap<Class<?>, String> defaultNameCache = new ConcurrentHashMap<>();
	// 缓存该key是否有fallback方法，这样如果木有就避免每次都反射去找了
	protected static ConcurrentHashMap<HystrixCommandKey, Boolean> commandContainsFallback = new ConcurrentHashMap<>();

	// 唯一构造器，完成了所有属性的初始化
	protected AbstractCommand(...){
		// group不能为null，但是key可以为null -> 自动用简单类名
        this.commandGroup = initGroupKey(group);
        this.commandKey = initCommandKey(key, getClass());
        ...
        // 自己没配置，那就使用传入的值。若传入为null，就使用groupKey
        // 大多数情况下让其保持和groupKey一样即可
		this.threadPoolKey = initThreadPoolKey(threadPoolKey, this.commandGroup, this.properties.executionIsolationThreadPoolKeyOverride().get());
		...
		this.threadPool = initThreadPool(threadPool, this.threadPoolKey, threadPoolPropertiesDefaults);
		...
	}
	... // 省略所有的init方法


	// 允许折叠器将此命令实例标记为用于折叠请求以及折叠多少请求
    void markAsCollapsedCommand(HystrixCollapserKey collapserKey, int sizeOfBatch) {
        eventNotifier.markEvent(HystrixEventType.COLLAPSED, this.commandKey);
        executionResult = executionResult.markCollapsed(collapserKey, sizeOfBatch);
    }


	// ========执行方法============
	public Observable<R> observe() { ... }
	public Observable<R> toObservable() { ... }

	// 抽象方法：目标方法以及针对的fallback方法
    protected abstract Observable<R> getExecutionObservable();
    protected abstract Observable<R> getFallbackObservable();
}

AbstractCommand：

	public Observable<R> toObservable() {
		... // 暂时省略非常多的Action和Func们
		// 需要注意的是：这里使用的是defer()，所以里面内容是不会立马执行的，知道有订阅者订阅了就执行
		// 也就是说observable = command.toObservable()是不会执行
		// observable.subscribe(xxx)订阅后，才会开始执行
		// 下面使用defer的效果都一样~~~~~~~~~
		return Observable.defer(() -> {
			
			... // 检验线程状态CommandState，每个command只能被执行一次，否则额抛出HystrixRuntimeException异常
			commandStartTimestamp = System.currentTimeMillis(); // 命令开始执行
			... // 记录日志：不管发生了什么，都要记录这个命令的执行
			// 是否允许请求缓存：properties.requestCacheEnabled().get() && getCacheKey() != null;
			// 虽然properties里默认是true开启的，但是需要你重写getCacheKey()缓存才会生效的呀
			boolean requestCacheEnabled = isRequestCachingEnabled();
			String cacheKey = getCacheKey(); // 默认是null。重写它缓存才会生效
			// 如果开启了缓存，就先从缓存里找结果
			if (requestCacheEnabled) {
				... // 一旦命中缓存，就处理好数据后return（关于请求缓存会后面再聊）
				// 若命中了缓存，设置isResponseFromCache = true;
			}
			
			// =========如果没有获取到缓存，则需要执行命令获得结果。========
			// applyHystrixSemantics函数：执行熔断器 + 目标方法等核心逻辑（最为复杂和关键的一个函数，下有详解）
			// 所以这里返回的hystrixObservable已经是目标命令结果了
			// wrapWithAllOnNextHooks：触发HystrixCommandExecutionHook相关回调
			// 另外，此处也是defer实现哦，所以目标方法并不会立马执行哦~~~（执行时机交给调用者才对嘛）
			Observable<R> hystrixObservable = Observable.defer(applyHystrixSemantics)
														.map(wrapWithAllOnNextHooks);
			// 它是最终的return
			Observable<R> afterCache;
			// 若开启了缓存
			if(requestCacheEnabled){
				... // 把结果hystrixObservable缓存起来并处理后返回
			} else {
				afterCache = hystrixObservable;
			}
			
			// terminateCommandCleanup：执行清理（分为目标方法执行了or没执行）
			// unsubscribeCommandCleanup：取消订阅
			// fireOnCompletedHook：触发executionHook.onSuccess(_cmd)该方法
			return afterCache.doOnTerminate(terminateCommandCleanup)
							.doOnUnsubscribe(unsubscribeCommandCleanup)
							.doOnCompleted(fireOnCompletedHook);
		});
	}

AbstractCommand：

	private Observable<R> applyHystrixSemantics(final AbstractCommand<R> _cmd) {
		executionHook.onStart(_cmd); // 开始执行

		// 若断路器放行（断路器闭合状态，当然喽也可能是半开状态）
		if (circuitBreaker.allowRequest()) {
			// 若你实现的是线程池隔离，那么此处实现就是TryableSemaphoreNoOp
			// 若你使用信号量隔离，它就会生效啦
			TryableSemaphore executionSemaphore = getExecutionSemaphore();
			...
			// 尝试申请信号资源
			if (executionSemaphore.tryAcquire()) {
				executionResult = executionResult.setInvocationStartTime(System.currentTimeMillis());

                return executeCommandAndObserve(_cmd) // =====执行目标方法=====
                        .doOnError(markExceptionThrown) // eventNotifier.markEvent() ... 
                        .doOnTerminate(singleSemaphoreRelease) // 确保释放信号量
                        .doOnUnsubscribe(singleSemaphoreRelease); // 确保释放信号量
			} else { // 若木有信号资源了，进入到信号量的fallabck
				return handleSemaphoreRejectionViaFallback();
			}
		} else { // 断路器打开了禁止你访问，那就直接执行fallabck逻辑
			return handleShortCircuitViaFallback();
		}

	}