在讲 Hystrix 插件前，我们需要先了解一下 Hystrix 的具体实现，Hystrix 限流框架基于 大名鼎鼎的 RxJava 实现，我们这节先来了解一下 RxJava。
1.8.6.1 响应式编程
响应式编程简称 FRP （Function Reatcive Programing）详细定义大家可以看一下 stackOverFlow 的这位大神解释，简单来说他就是一种基于事件的模型，我们需要知道某个事件是否发生有两种方式，一种就是主动轮训，我们把它称为 Proactive 方式。另一种就是被动接收反馈，我们称为 Reactive。
响应式编程是未来的一种趋势，对于后端java 程序员关于响应式编程第一个应该会想到 Spring 5 引入的 WebFlux ，而在 Android 中，响应式编程发展却快得多，最主要原因是使用场景，Android 客户端中存在大量的和服务器交互的程序，所以存在大量的异步回掉过程，一般这种请看就会使用到 RxJava ，它可以更加简化代码的逻辑。
1.8.6.2 Rxjava 使用入门
我们先来看一下一个RxJava 调用的实例：
//创建被观察者
Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("111");
subscriber.onNext("222");
subscriber.onNext("333");
subscriber.onCompleted();
}
});
//创建观察者
Observer<String> observer = new Observer<String>() {
@Override
public void onCompleted() {
System.out.println("Complete");
}
@Override
public void onError(Throwable e) {
e.printStackTrace();
}
@Override
public void onNext(String s) {
System.out.println(s);
}
};
//注册，将观察者和被观察者关联，将会触发OnSubscribe.call方法
observable.subscribe(observer);
我们先来介绍一下 RxJava 最重要的两个成员：
-
Observable可以理解为被观察者，RxJava中使用了观察者模式。 -
Observer可以理解为观察者。
Observable 方法 -
call即被观察者的所有活动，这里可以调用subscriber.onNext向订阅者发送消息。 -
subscribe这个是所有方法的启动源头，我将其主要逻辑整理如下，首先他会调用Observer的start方法，告诉它开始了，然后调用自己的call方法，假如发生异常，就调用subscriber的onError方法，然后unsubscribed, 也就是将被观察者注册的观察者清空。
static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {
subscriber.onStart();
try {
RxJavaHooks.onObservableStart(observable, observable.onSubscribe).call(subscriber);
return RxJavaHooks.onObservableReturn(subscriber);
} catch (Throwable e) {
subscriber.onError(RxJavaHooks.onObservableError(e));
return Subscriptions.unsubscribed();
}
}
接下来我们说一下 observer 的三个方法：
-
onCompleted当Observable的 call 完成后调用，它后onError只有一个会被调用 -
onError我们可以参考上面的源码，在 调用call异常时调用 -
onNextObservable每个发生时间都会调用。
1.8.6.3 Rxjava 各种操作
-
Schedule调度线程
我们将我们的代码稍微改一下
final CountDownLatch latch = new CountDownLatch(1);
//创建被观察者
Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("111");
subscriber.onNext("222");
subscriber.onNext("333");
subscriber.onCompleted();
System.out.println(Thread.currentThread().getName());
}
});
//创建观察者
Observer<String> observer = new Observer<String>() {
@Override
public void onCompleted() {
System.out.println(Thread.currentThread().getName() + "-Complete");
latch.countDown();
}
@Override
public void onError(Throwable e) {
e.printStackTrace();
}
@Override
public void onNext(String s) {
System.out.println(Thread.currentThread().getName() + "-" +s);
}
};
//注册，将观察者和被观察者关联，将会触发OnSubscribe.call方法
observable.subscribeOn(Schedulers.newThread()) // 指定 subscribe() 发生在新的线程
.observeOn(Schedulers.newThread()).subscribe(observer);
latch.await();
得到结果入下：
RxNewThreadScheduler-2
RxNewThreadScheduler-1-111
RxNewThreadScheduler-1-222
RxNewThreadScheduler-1-333
RxNewThreadScheduler-1-Complete
这对于我们来说确实是很酷的事情，这里subscribeOn 和 observeOn 分别指定了我们生产者在调用 call 的线程和消费者的线程。
- map 转换
map
我们可以参考如上面的图和下面的代码，假如observable发射的是 Integer 类型，而Obsever需要接收 String 那么我们就可以通过 map 进行转换。
observable.subscribeOn(Schedulers.newThread())
.observeOn(Schedulers.newThread()).map(i -> i +"").subscribe(observer);
- flatmap
flatmap 对于很多新手来说是最难理解的，flat 可以理解为平铺，我们需要将Observable发射的时间平铺开来的场景，如我们需要将 Stirng 使用空格分割成一个个单词进行消费，如下：
observable.subscribeOn(Schedulers.newThread())
.observeOn(Schedulers.newThread()).flatMap(s -> Observable.from(s.split(" "))).subscribe(observer);
我们也可以参考图片，我们这里将一个OnNext 事件转为一个 Observable 然后将这些 Observable 汇总成一个 Observable 给 Obsever 进行消费。
flatmpa
-
from如上，我们可以通过from 将一个数组快速转为Observable。 -
filter顾名思义，就是将一部分事件过滤掉，这里很多操作和 stream 是类似的。 -
distinct去重。
1.8.6.4 总结
我们这次只是简单地学习了一下 RxJava 的原理，RxJava 还有很多好玩的地方等着我们去探索，有兴趣的小伙伴可以深入学习一下，非常适合入门响应式编程，我们下一节将接收 Hystrix 如何通过 RxJava 实现降级和熔断。
网友评论