Hystrix拾遗

HystrixCommand 与 HystrixObservableCommand 区别

HystrixCommand 的命令逻辑写在run() HystrixObservableCommand的命令逻辑写在construct()

HystrixCommand的run()是由新建立的线程执行(非阻塞) HystrixObservableCommand的construct()是由调用程序线程执行(阻塞)

HystrixCommand一个实例只能向调用程序发送单条数据 HystrixObservableCommand一个实例能够顺序发送多条数据

重要方法

execute 与 queue

这2个方法是只有HystrixCommand才有的，execute是阻塞的，queue是非阻塞的。

observe 与 toObservable

这2个方法是HystrixCommand 与 HystrixObservableCommand都有的。

不过对于HystrixCommand 与 HystrixObservableCommand这2个方法的表现是不太同样的。

咱们知道HystrixCommand逻辑在run中，HystrixObservableCommand的逻辑在construct中，observe方法触发run是非阻塞的方式，就是新的线程执行run，而触发construct则是阻塞方式，就是调用线程执行construct。

使用toObservable和observe方法不太同样，toObserve方法自己不会触发run或者construct方法，而是要在subscribe的时候触发这run或者construct方法。触发的方式和observe同样，对于run使用新线程非阻塞的方式，对于construct使用调用线程阻塞的方式。

GitHub上对于observe和toObservable区别介绍的确有点绕，其实就是，observe无论有没有订阅者都会执行run或者construct，toObserve只有有订阅者的时候才会执行run或者construct的方法。

observe和toObserve用的很少，只有有多条结果不是一块儿返回的时候才会用到。

HystrixCommand同步与异步测试

import com.netflix.hystrix.HystrixCommand;
import com.netflix.hystrix.HystrixCommandGroupKey;
import com.netflix.hystrix.HystrixCommandProperties;

import java.util.concurrent.TimeUnit;

public class ASHystrixCommand extends HystrixCommand<String>{

    public ASHystrixCommand() {
        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("as")).
                andCommandPropertiesDefaults(HystrixCommandProperties.Setter().withExecutionTimeoutInMilliseconds(6000)));
    }

    @Override
    protected String run() throws Exception {
        TimeUnit.SECONDS.sleep(5);
        return "done";
    }

    @Override
    protected String getFallback() {
        return "as failure";
    }

}

上面我把Hystrix的执行超时时间设置为6秒，主要是为了在run中模拟长时间运行，异步调用给调用线程执行机会。

下面是测试代码：

import org.junit.Test;

import java.util.concurrent.Future;

public class ASHystrixCommandTest {

    private ASHystrixCommand command = new ASHystrixCommand();

    @Test
    public void Syn() throws Exception {
        System.out.println(command.execute());
        System.out.println("do other thing");
    }

    @Test
    public void Asyn() throws Exception {
        Future<String> queue = command.queue();
        System.out.println("do other thing");
        System.out.println(queue.get());
    }

}

咱们能够看到Syn测试的输出是：

done do other thing

说明execute是同步执行的。

而Asyn测试的输出结果是：

do other thing done

说明Asyn是异步执行的，queue执行命令是非阻塞的，Future.get才会阻塞

HystrixObservableCommand 测试

HystrixObservableCommand依赖RxJava，想要使用好的能够先学习一下RxJava，不过不了解也是没有关系的，这里简单介绍几个名词： Observable： 被观察者 Observer：观察者 Flowable ：被观察者 Subscriber ：观察者

import com.netflix.hystrix.HystrixCommandGroupKey;
import com.netflix.hystrix.HystrixCommandProperties;
import com.netflix.hystrix.HystrixObservableCommand;
import rx.Observable;
import rx.Subscriber;
import rx.schedulers.Schedulers;

public class ObservableCommand extends HystrixObservableCommand<String>{

    private String name;

    public ObservableCommand(String name) {
        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("obs")).
                andCommandPropertiesDefaults(HystrixCommandProperties.Setter().withExecutionTimeoutInMilliseconds(6000)));
        this.name = name;
    }

    public Observable<String> resumeWithFallback(){
        return null;
    }

    @Override
    protected Observable<String> construct() {
        System.out.println("construct");
        return Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> observer) {
                try {
                    if (!observer.isUnsubscribed()) {
                        observer.onNext("Hello");
                        observer.onNext(name + " !");
                        observer.onCompleted();
                    }
                } catch (Exception e) {
                    observer.onError(e);
                }
            }
        } ).subscribeOn(Schedulers.io());
    }
}

HystrixObservableCommand的逻辑是在construct中，失败函数也和HystrixCommand的getFailBack不一样，而是使用的resumeWithFallback。

测试类：

import org.junit.Test;
import rx.Observable;
import rx.Observer;
import rx.functions.Action1;
import rx.observables.BlockingObservable;

public class ObservableCommandTest {

    @Test
    public void obs(){
        ObservableCommand observableCommand = new ObservableCommand("allen");
        Observable<String> observe = observableCommand.observe();
//        BlockingObservable<String> stringBlockingObservable = observe.toBlocking();
//        Iterator<String> iterator = stringBlockingObservable.getIterator();
//        while(iterator.hasNext()) {
//            System.out.println(iterator.next());
//        }

        observe.subscribe(new Observer<String>() {
             @Override
            public void onCompleted() {
                System.out.println("obs completed");
            }

             @Override
            public void onError(Throwable e) {
                System.out.println("obs error");
                e.printStackTrace();
            }

             @Override
            public void onNext(String v) {
                System.out.println("obs onNext: " + v);
            }

        });

    }

    @Test
    public void obsWithoutSub(){
        ObservableCommand observableCommand = new ObservableCommand("allen");
        Observable<String> observe = observableCommand.observe();
    }

    @Test
    public void tobs(){
        ObservableCommand observableCommand = new ObservableCommand("allen");
        Observable<String> observe = observableCommand.toObservable();
//        BlockingObservable<String> stringBlockingObservable = observe.toBlocking();
//        Iterator<String> iterator = stringBlockingObservable.getIterator();
//        while(iterator.hasNext()) {
//            System.out.println(iterator.next());
//        }

        observe.subscribe(new Observer<String>() {
            @Override
            public void onCompleted() {
                System.out.println("obs completed");
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("obs error");
                e.printStackTrace();
            }

            @Override
            public void onNext(String v) {
                System.out.println("obs onNext: " + v);
            }

        });
    }

    @Test
    public void tobsWithoutSub(){
        ObservableCommand observableCommand = new ObservableCommand("allen");
        Observable<String> observe = observableCommand.toObservable();
    }

    @Test
    public void action(){
        ObservableCommand observableCommand = new ObservableCommand("allen");
        Observable<String> observe = observableCommand.toObservable();

        observe.subscribe(new Action1<String>() {
            @Override
            public void call(String v) {
                System.out.println("action1 call: " + v);
            }

        });
    }

    @Test
    public void blocking() {
        ObservableCommand observableCommand = new ObservableCommand("allen");
        Observable<String> observe = observableCommand.observe();
        BlockingObservable<String> stringBlockingObservable = observe.toBlocking();
        System.out.println("before");
        System.out.println(stringBlockingObservable.single());
    }

}

上面主要测试了observe和toObserve没有订阅者的时候的执行状况： 从输出咱们能够看到，当toObserve没有订阅者的时候是不会执行construct的，而observe无论有没有订阅者都会执行construct。

Observable的subscribe能够接受的类型有Observer，这个监听了出现异常和完成事件，若是不关心这2个事件，也能够使用Action1。

参考

Hystrix 各个配置

Hystrix实例与高并发测试