[享学Netflix] 三十四、Hystrix目标方法执行逻辑源码解读：executeCommandAndObserve

AbstractCommand：

	private Observable<R> executeCommandAndObserve(final AbstractCommand<R> _cmd) {
		// 执行上下文。保证线程池内亦能获取到主线程里的参数
		HystrixRequestContext currentRequestContext = HystrixRequestContext.getContextForCurrentThread();
		... // 暂时忽略一些内置的function们
		
		Observable<R> execution;
		// 二者的唯一却别是：若开启了超时支持的话，就只可观察对象的结果处
		// lift一个HystrixObservableTimeoutOperator实例，以监控超时情况
		// 关于Hystrix是如何实现超时的，这在后面专文讲解~~是个较大的，也较难的话题
        if (properties.executionTimeoutEnabled().get()) {
            execution = executeCommandWithSpecifiedIsolation(_cmd)
                    .lift(new HystrixObservableTimeoutOperator<R>(_cmd));
        } else {
            execution = executeCommandWithSpecifiedIsolation(_cmd);
        }

		// 得到execution后，开始注册一些基本的事件、观察者
        return execution.doOnNext(markEmits)
                .doOnCompleted(markOnCompleted)
                .onErrorResumeNext(handleFallback)
                .doOnEach(setRequestContext);
	}

markEmits函数内容：

	markEmits = r -> {
		// 是否应该在onNext这步报告数据
		// HystrixCommand -> false
		// HystrixObservableCommand -> true
	    if (shouldOutputOnNextEvents()) {
	        executionResult = executionResult.addEvent(HystrixEventType.EMIT);
	        eventNotifier.markEvent(HystrixEventType.EMIT, commandKey);
	    }

		// 命令是否是标量
		// HystrixCommand -> true
		// HystrixObservableCommand -> false
	    if (commandIsScalar()) {
	        long latency = System.currentTimeMillis() - executionResult.getStartTimestamp();
	        eventNotifier.markCommandExecution(getCommandKey(), properties.executionIsolationStrategy().get(), (int) latency, executionResult.getOrderedList());

			// 这几句代码是重点：
			// 记录结果为SUCCESS成功
			// 并且，并且，并且circuitBreaker.markSuccess();（若断路器是打开的，此处就关闭了）
	        eventNotifier.markEvent(HystrixEventType.SUCCESS, commandKey);
	        executionResult = executionResult.addEvent((int) latency, HystrixEventType.SUCCESS);
	        circuitBreaker.markSuccess();
	    }
	}

markOnCompleted函数内容：

	markOnCompleted = () -> {
		// 注意这个！符号
		if (!commandIsScalar()) {
			... // 逻辑完全同上面的Scalar部分
		}
	}

handleFallback函数内容：

	handleFallback = (Throwable t) -> {
		// 把Throwable t强转为Exception e（若不是Exception类型就包装为Exception类型）
		// 比如若t是NPE异常，那么t和e是完全一样的。
		// 只有当t是error类时，t才和e不相等
        Exception e = getExceptionFromThrowable(t);
        // 既然发生错误了，那就记录执行时候的异常e
        executionResult = executionResult.setExecutionException(e);
        
		// 若异常类型是RejectedExecutionException：线程池拒绝
		// 若异常类型是HystrixTimeoutException：目标方法执行超时
		// 若异常类型是HystrixBadRequestException：下文详细分解
        if (e instanceof RejectedExecutionException) {
            return handleThreadPoolRejectionViaFallback(e);
        } else if (t instanceof HystrixTimeoutException) {
            return handleTimeoutViaFallback();
        } else if (t instanceof HystrixBadRequestException) {
            return handleBadRequestByEmittingError(e);
        } else {
            // 什么时候会进入到这里？只有当子类复写了getExceptionFromThrowable()方法的时候才有可能进入到这里
            // 这里算是一种兜底：保证不管咋样HystrixBadRequestException都不会触发熔断
            // 其实我倒觉得，不让getExceptionFromThrowable这个方法被复写也行的
            if (e instanceof HystrixBadRequestException) {
                eventNotifier.markEvent(HystrixEventType.BAD_REQUEST, commandKey);
                return Observable.error(e);
            }

            return handleFailureViaFallback(e);
        }
	}

	setRequestContext = (Notification<? super R> rNotification) -> {
		if (!HystrixRequestContext.isCurrentThreadInitialized()) {
			HystrixRequestContext.setContextOnCurrentThread(currentRequestContext);
		}
	}

AbstractCommand：

	private Observable<R> executeCommandWithSpecifiedIsolation(AbstractCommand<R> _cmd) {
		// 标记我们正在一个线程中执行(即使我们最终被拒绝
		// 我们仍然是一个线程执行，而不是信号量)
		if (properties.executionIsolationStrategy().get() == ExecutionIsolationStrategy.THREAD) {
			// 默认是一个defer延迟执行的可观察对象
			// 注意：进到这个回调里面来后，就是使用的线程池的资源去执行了（获取到了线程池资源）
			// 比如此处线程号就是：hystrix-fallbackDemoGroup-1
			return Observable.defer(() -> {
				// 记录：目标方法已经执行（不管出异常与否，反正就是执行了）
				// 应为如果被熔断了，或者线程池拒绝了它是不会被执行的
				executionResult = executionResult.setExecutionOccurred();
				// 线程状态必须是OBSERVABLE_CHAIN_CREATED时才让执行
				// 而此状态是由toObservable()方法设置过来的
                if (!commandState.compareAndSet(CommandState.OBSERVABLE_CHAIN_CREATED, CommandState.USER_CODE_EXECUTED)) {
                    return Observable.error(new IllegalStateException("execution attempted while in state : " + commandState.get().name()));
                }
                // 收集指标信息：开始执行
				metrics.markCommandStart(commandKey, threadPoolKey, ExecutionIsolationStrategy.THREAD);

				// 这个判断非常的有意思：如果在run方法还没执行之前
				// 也就是在线程切换之间就超时了，那就直接返回一个错误
				if (isCommandTimedOut.get() == TimedOutStatus.TIMED_OUT) {
					return Observable.error(new RuntimeException("timed out before executing run()"));
				}

				// 把执行的线程状态标记为STARTED：启动
				if (threadState.compareAndSet(ThreadState.NOT_USING_THREAD, ThreadState.STARTED)) {
					// 全局计数器+1
					// 此处是唯一调用处。信号量里是木有此调用的哦
					HystrixCounters.incrementGlobalConcurrentThreads();
					// 标记线程池已经开始准备执行了
					threadPool.markThreadExecution(); 

                    // store the command that is being run
                    // 这个保存使用的ThreadLocal<ConcurrentStack<HystrixCommandKey>>和当前线程绑定
                    // 这样确保了命令在执行时的线程安全~~~~~~~
                    endCurrentThreadExecutingCommand = Hystrix.startCurrentThreadExecutingCommand(getCommandKey());
                    executionResult = executionResult.setExecutedInThread();

					// 执行钩子程序，以及执行目标run方法程序
					// getUserExecutionObservable：getExecutionObservable()抽象方法获取到目标方法
					// 本处也是该抽象方法的唯一调用处哦
					// 若这里面任何一个方法抛出异常（哪怕是hook方法），就原样抛出
                    try {
                        executionHook.onThreadStart(_cmd);
                        executionHook.onRunStart(_cmd);
                        executionHook.onExecutionStart(_cmd);
                        return getUserExecutionObservable(_cmd);
                    } catch (Throwable ex) {
                        return Observable.error(ex);
                    }
				} else { // 说明已经unsubscribed了，就抛错
					return Observable.error(new RuntimeException("unsubscribed before executing run()"));
				}
			});
		}
	}

AbstractCommand：

		// 核心是threadPool.getScheduler()获取到一个Scheduler
		.subscribeOn(threadPool.getScheduler(() -> {
			properties.executionIsolationThreadInterruptOnTimeout().get() && _cmd.isCommandTimedOut.get() == TimedOutStatus.TIMED_OUT
		}));

AbstractCommand：

	return Observable.defer(() -> {
		// 检测线程状态。记录线程执行（不管成功与否）
        executionResult = executionResult.setExecutionOccurred();
        if (!commandState.compareAndSet(CommandState.OBSERVABLE_CHAIN_CREATED, CommandState.USER_CODE_EXECUTED)) {
            return Observable.error(new IllegalStateException("execution attempted while in state : " + commandState.get().name()));
        }
	
		// 一样的，标记线程开始执行了。隔离模式是：SEMAPHORE哦
		metrics.markCommandStart(commandKey, threadPoolKey, ExecutionIsolationStrategy.SEMAPHORE);
		
		// 不解释
		endCurrentThreadExecutingCommand = Hystrix.startCurrentThreadExecutingCommand(getCommandKey());
        try {
            executionHook.onRunStart(_cmd);
            executionHook.onExecutionStart(_cmd);
            return getUserExecutionObservable(_cmd);
        } catch (Throwable ex) {
            return Observable.error(ex);
        }
	});