[享学Netflix] 三十六、Hystrix请求命令：HystrixCommand和HystrixObservableCommand

// 在父类的基础上，多实现了HystrixExecutable接口，从而新增提供了execute()/queue()方法
public abstract class HystrixCommand<R> extends AbstractCommand<R> implements HystrixExecutable<R> {

	... // 省略其所有构造器
	... // 省略静态配置内部类Setter

	// 执行线程Thread
	private final AtomicReference<Thread> executionThread = new AtomicReference<>();
	// 用于接收executionIsolationThreadInterruptOnFutureCancel这个属性值
	// 用于：Future.cancel(interruptOnFutureCancel )
	private final AtomicBoolean interruptOnFutureCancel = new AtomicBoolean(false);


	// 最为重要的一个抽象方法：你的目标逻辑写在这里面
	// execute()和queue()会调用方法
	// 小细节：此方法特意设计为protected，警告你执行时候不要直接调用此方法哦~
	protected abstract R run() throws Exception;
	// 详细你对`No fallback available.`这几个字无比熟悉吧，在项目中经常碰见。
	// 也就是当execute()任何情况失败了，都会尝试来调用这个方法给你个回滚的机会
	// 说明：这里应该做的工作是：不需要网络产生，也就是JVM内的调用
	// 换句话说：这个返回最好是一个常量值，或者是缓存值是最好的，不要耗时
	// 默认是抛出异常，建议你实现此方法，当然喽，具体需要具体分析~~~~~比较有些方法不能回退...
    protected R getFallback() {
        throw new UnsupportedOperationException("No fallback available.");
    }

	// ============实现父类抽象方法============
    @Deprecated // 这个方法过期了请一定不要用，不管它就成
    @Override
    protected String getFallbackMethodName() {
        return "getFallback";
    }
    // 父类方法：判断你是否提供了fallabck方法（看你是否复写了getFallback方法）
    @Override
    protected boolean isFallbackUserDefined() {
    	... // 查缓存
    	// 这里使用的是getDeclaredMethod，所以只有你实现类本类里有名为`getFallback`的方法才算的哦
    	getClass().getDeclaredMethod("getFallback");
    	... // 放缓存
    }
	
	// command是否是标量
	// 当commandIsScalar为true时，在markEmits就会circuitBreaker.markSuccess()；
	// commandIsScalar为false时，在markOnCompleted动作时调用circuitBreaker.markSuccess()
	// 本处为true，另一实现HystrixObservableCommand为false
    @Override
    protected boolean commandIsScalar() {
        return true;
    }
	
	// ============实现接口方法============

	// 可以看到execute()的实现其实就是queue的实现，只是立马get了而已
    public R execute() {
        try {
            return queue().get();
        } catch (Exception e) {
            throw Exceptions.sneakyThrow(decomposeException(e));
        }
    }
    // 可以看到，Futrue的实现实际上还是Observable
    // 只不过它在其基础上对cancle方法特殊处理了一下：中断线程与否
    public Future<R> queue() {
		Future<R> delegate = toObservable().toBlocking().toFuture();
		Future<R> f = new Future<R>() { ... };
		
		// 对立即抛出的错误状态的特殊处理
		// 我在想，什么情况下这么快就isDone了呢？delegate Futrue刚创建就抛错了？
        if (f.isDone()) {
            try {
                f.get();
                return f;
            } catch (Exception e) { ... }
		}
		return f;
    }
}

@Test
public void fun1(){
    MyCommand command = new MyCommand();
    Integer result = command.execute();
    System.out.println(result);
}


private static class MyCommand extends HystrixCommand<Integer>{

    protected MyCommand() {
        super(HystrixCommandGroupKey.Factory.asKey("MyCommandGroup"));
    }

    // 目标方法
    @Override
    protected Integer run() throws Exception {
        return 1;
    }

    // fallback方法逻辑，并不是必须提供的哦
    @Override
    protected Integer getFallback() {
        return -1;
    }
}

1

// 它比HystrixCommand少实现了一个`HystrixExecutable`，所以它仅有异步的能力
public abstract class HystrixObservableCommand<R> extends AbstractCommand<R> implements HystrixObservable<R> {
	
	... // 省略所有构造器
	... // 省略Setter

	// ============实现父类抽象方法============
	// 如果有下一个事件应该报告
    @Override
    protected boolean shouldOutputOnNextEvents() {
        return true;
    }
    @Override // 标记为过期了，禁用
    protected String getFallbackMethodName() {
        return "resumeWithFallback";
    }
    // 它的方法名改为了resumeWithFallback，请务必注意哦
    @Override
    protected boolean isFallbackUserDefined() {
    	... // 缓存
    	getClass().getDeclaredMethod("resumeWithFallback");
    	... // 缓存
    }
    // 若你想fallback，请复写此方法
    protected Observable<R> resumeWithFallback() {
        return Observable.error(new UnsupportedOperationException("No fallback available."));
    }

    @Override
    protected boolean commandIsScalar() {
        return false;
    }


    @Override
    final protected Observable<R> getExecutionObservable() {
        return construct();
    }
    // 目标方法。你的逻辑写在这里面
    // HystrixCommand的叫run方法嘛，这个叫construct方法~~~~
	protected abstract Observable<R> construct();
	// ============实现接口方法============
	... // 父类均已实现
}

@Test
public void fun2() {
    MyObservableCommand command = new MyObservableCommand();
    Observable<Integer> observe = command.observe();
    // Observable<Integer> observe = command.toObservable();

    System.out.println("==========订阅者之前的一句话==========");
    observe.subscribe(d -> System.out.println(d));
    System.out.println("==========订阅者之后的一句话==========");
}


private static class MyObservableCommand extends HystrixObservableCommand<Integer> {

    protected MyObservableCommand() {
        super(HystrixCommandGroupKey.Factory.asKey("MyCommandGroup"));
    }

    // 一次性发送5个返回值，持续不断
    @Override
    protected Observable<Integer> construct() {
        // return Observable.just(1, 2, 3, 4);
        return Observable.create(subscriber -> {
            subscriber.onStart();
            for (int i = 1; i <= 4; i++) {
                System.out.println("开始发射数据：" + i);
                subscriber.onNext(i);
            }
            subscriber.onCompleted();
        });
    }

    @Override
    protected Observable<Integer> resumeWithFallback() {
        return Observable.just(-1);
    }
}

// 使用observe()方法结果：发射数据动作立马执行
开始发射数据：1
开始发射数据：2
开始发射数据：3
开始发射数据：4
==========订阅者之前的一句话==========
1
2
3
4
==========订阅者之后的一句话==========

// 使用toObservable()方法结果：发射数据动作明显是木有立马开始的
==========订阅者之前的一句话==========
开始发射数据：1
1
开始发射数据：2
2
开始发射数据：3
3
开始发射数据：4
4
==========订阅者之后的一句话==========