hystrix 命令执行

Posted 2021-01-23 zhangwanhua

　　上一节中讲到了HystrixCommand有四种执行方法，这一节就来讲一下这四种方法直接的关系以及他们的实现。

　　execute方法使用同步方式获取结果，本质是调用了queue方法获取了一个Future，然后通过该Future获取返回结果。

public R execute() {
        try {
            return queue().get();
        } catch (Exception e) {
            throw Exceptions.sneakyThrow(decomposeException(e));
        }
    }

　　queue方法通过异步方式活动结果，本质是通过toObservable方法获取了一个Observable，然后通过该Observable获得一个Future

public Future<R> queue() {
      ...
        final Future<R> delegate = toObservable().toBlocking().toFuture();
     ...    
};

　　observe方法通过Observable实现异步，并且该Observable无论在任何时候监听，都可以接收到所有消息。本质是使用了一个ReplaySubject监听toObservable返回的Observable，然后再返回该ReplaySubject。

public Observable<R> observe() {
        ReplaySubject<R> subject = ReplaySubject.create();
        final Subscription sourceSubscription = toObservable().subscribe(subject);
       ...
        return subject.doOnUnsubscribe(new Action0() {
            @Override
            public void call() {
                sourceSubscription.unsubscribe();
            }
        });
    }

　　toObservable方法Observable实现异步，在Observable被监听时执行。本质是创建一个Observable，当这个Observable被监听时，执行run命令，并返回消息。

public Observable<R> toObservable() {
       ...
        return Observable.defer(new Func0<Observable<R>>() {
            @Override
            public Observable<R> call() {
               ...
                Observable<R> hystrixObservable =
                        Observable.defer(applyHystrixSemantics)  //会调用run方法
                                .map(wrapWithAllOnNextHooks);
               ...
            }
        });
    }