rxjava文档地址https://mcxiaoke.gitbooks.io/rxdocs/content/ 这个是中文版的
android studio 添加依赖 implementation ‘io.reactivex.rxjava3:rxjava:3.0.4’
首先，打印helloworld：
public void hello(String args){
Flowable.fromArray(args).subscribe(s -> System.out.println("hello " + s + "!"));
}
跟以前其他语言不大一样，看上去很麻烦，我们一步步来看
Flowable.fromArray(args)
这个方法最重要的就是里面的最后一句
new FlowableFromArray<>(items)
果然FlowableFromArray是Flowable的子类，所以真正的实现在子类里面
Flowable.fromArray(args).subscribe
subscribe进到里面的是
public final Disposable subscribe(@NonNull Consumersuper T> onNext, @NonNull Consumersuper Throwable> onError,
@NonNull Action onComplete) {
Objects.requireNonNull(onNext, "onNext is null");
Objects.requireNonNull(onError, "onError is null");
Objects.requireNonNull(onComplete, "onComplete is null");
LambdaSubscriber ls = new LambdaSubscriber<>(onNext, onError, onComplete, FlowableInternalHelper.RequestMax.INSTANCE);
subscribe(ls);
return ls;
}
看上去最重要的就是这两句了
LambdaSubscriber ls = new LambdaSubscriber<>(onNext, onError, onComplete, FlowableInternalHelper.RequestMax.INSTANCE);
subscribe(ls);
先进到subscribe(ls)中，发现这句
subscribeActual(flowableSubscriber)
跳进去发现是个抽象方法，那么实现肯定在子类啦，进到子类FlowableFromArray
@Override
public void subscribeActual(Subscribersuper T> s) {
if (s instanceof ConditionalSubscriber) {
s.onSubscribe(new ArrayConditionalSubscription<>(
(ConditionalSubscribersuper T>)s, array));
} else {
s.onSubscribe(new ArraySubscription<>(s, array));
}
}
跳进去又发现onSubscribe是个抽象方法，那么实现方法在哪呢，对啦，就是之前看到的LambdaSubscriber
public void onSubscribe(Subscription s) {
if (SubscriptionHelper.setOnce(this, s)) {
try {
onSubscribe.accept(this);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
s.cancel();
onError(ex);
}
}
}
这个onSubscribe.accept(this)跳过去就是接口Consumer的accept方法了
所以一开始的helloworld代码也可以改成
FlowableFromArray flowableFromArray = new FlowableFromArray(new String[]{args});
flowableFromArray.subscribe(new Consumer() {
@Override
public void accept(String s) throws Throwable {
System.out.println("hello " + s + "!");
}
});
是不是很麻烦，饶了一大圈，没关系，我们继续往下看
这里给出一些名词的翻译
- Reactive 直译为反应性的，有活性的，根据上下文一般翻译为反应式、响应式
- Iterable 可迭代对象，支持以迭代器的形式遍历，许多语言中都存在这个概念
- Observable 可观察对象，在Rx中定义为更强大的Iterable，在观察者模式中是被观察的对象，一旦数据产生或发生变化，会通过某种方式通知观察者或订阅者
- Observer 观察者对象，监听Observable发射的数据并做出响应，Subscriber是它的一个特殊实现
- emit 直译为发射，发布，发出，含义是Observable在数据产生或变化时发送通知给Observer，调用Observer对应的方法，文章里一律译为发射
- items 直译为项目，条目，在Rx里是指Observable发射的数据项，文章里一律译为数据，数据项
下面是常用的操作符列表：
- 创建操作 Create, Defer, Empty/Never/Throw, From, Interval, Just, Range, Repeat, Start, Timer
- 变换操作 Buffer, FlatMap, GroupBy, Map, Scan和Window
- 过滤操作 Debounce, Distinct, ElementAt, Filter, First, IgnoreElements, Last, Sample, Skip, SkipLast, Take, TakeLast
- 组合操作 And/Then/When, CombineLatest, Join, Merge, StartWith, Switch, Zip
- 错误处理 Catch和Retry
- 辅助操作 Delay, Do, Materialize/Dematerialize, ObserveOn, Serialize, Subscribe, SubscribeOn, TimeInterval, Timeout, Timestamp, Using
- 条件和布尔操作 All, Amb, Contains, DefaultIfEmpty, SequenceEqual, SkipUntil, SkipWhile, TakeUntil, TakeWhile
- 算术和集合操作 Average, Concat, Count, Max, Min, Reduce, Sum
- 转换操作 To
- 连接操作 Connect, Publish, RefCount, Replay
- 反压操作，用于增加特殊的流程控制策略的操作符
下面我们来看第一个操作符：Create
Observable.create(new Observable.OnSubscribe() {
@Override
public void call(Subscribersuper Integer> observer) {
try {
if (!observer.isUnsubscribed()) {
for (int i = 1; i < 5; i++) {
observer.onNext(i);
}
observer.onCompleted();
}
} catch (Exception e) {
observer.onError(e);
}
}
} ).subscribe(new Subscriber() {
@Override
public void onNext(Integer item) {
System.out.println("Next: " + item);
}
@Override
public void onError(Throwable error) {
System.err.println("Error: " + error.getMessage());
}
@Override
public void onCompleted() {
System.out.println("Sequence complete.");
}
});
我们一起来看源码
首先是Observable的create方法
public final static Observable create(OnSubscribe f) {
return new Observable(hook.onCreate(f));
}
这里没什么，就是返回创建一个Observable对象，但是要注意里面的参数OnSubscribe
public interface OnSubscribe extends Action1super T>> {
// cover for generics insanity
}
public interface Action1 extends Action {
void call(T t);
}
这个参数是一个接口，它的父类里有个抽象待实现的方法call，而且call方法被传了Subscriber进去
我们来看Subscriber这个类，原来是个接口，而且它的父类Observer有三个很重要的方法
public interface Observer {
void onCompleted();
void onError(Throwable e);
void onNext(T t);
}
第一个create方法算是完成了，我们可以拆分来看
Observable integerObservable = Observable.create(new Observable.OnSubscribe() {
@Override
public void call(Subscribersuper Integer> observer) {
try {
if (!observer.isUnsubscribed()) {
for (int i = 0; i < 5; i++) {
observer.onNext(i);
}
observer.onCompleted();
}
} catch (Exception e) {
observer.onError(e);
}
}
});
第二个方法subscribe，它的参数也是Subscriber,即intergerObservable.subscribe(Subscriber)
所以我们就看出来了，Observable这个被观察者先是通过call增加一系列的监听，然后通过subscribe订阅监听。这样，当call里的内容开始执行后，触发监听回调
下面我要放大招了，我把源码简化了一下
public interface MyOnSubscribe {
void call(MySubscriber subscriber);
}
public interface MySubscriber {
void onNext();
void onCompleted();
void onError();
}
public class MyObservable {
MyOnSubscribe onSubscribe;
public MyObservable(MyOnSubscribe onSubscribe) {
this.onSubscribe = onSubscribe;
}
public final static MyObservable create(MyOnSubscribe onSubscribe) {
return new MyObservable(onSubscribe);
}
public final void subscribe(MySubscriber subscriber) {
onSubscribe.call(subscriber);
}
}
测试代码
public void hello() {
MyObservable.create(new MyOnSubscribe() {
@Override
public void call(MySubscriber subscriber) {
try {
for (int i = 0; i < 5; i++) {
subscriber.onNext();
}
subscriber.onCompleted();
} catch (Exception e) {
subscriber.onError();
}
}
}).subscribe(new MySubscriber() {
@Override
public void onNext() {
System.out.println(1);
}
@Override
public void onCompleted() {
System.out.println("onCompleted");
}
@Override
public void onError() {
System.out.println("onError");
}
});
}
得到的结果是一样的。所以说，代码万变不离其中，只要灵活运用接口，接口就是用来监听的
第二个操作符from
Integer[] items = {0, 1, 2, 3, 4, 5};
Observable.from(items).subscribe(new Action1() {
@Override
public void call(Integer integer) {
System.out.println(integer);
}
});
先看Observable的from方法
public final static Observable from(T[] array) {
return from(Arrays.asList(array));
}
其实就是把数组转成list，但是再点from进去就很重要
public final static Observable from(Iterableextends T> iterable) {
return create(new OnSubscribeFromIterable(iterable));
}
public OnSubscribeFromIterable(Iterableextends T> iterable) {
if (iterable == null) {
throw new NullPointerException("iterable must not be null");
}
this.is = iterable;
}
OnSubscribeFromIterable是继承自OnSubscribe的，所以后面调的call方法，实际上是调的OnSubscribeFromIterable里的call方法，我们来看一下
@Override
public void call(final Subscribersuper T> o) {
final Iteratorextends T> it = is.iterator();
if (!it.hasNext() && !o.isUnsubscribed())
o.onCompleted();
else
o.setProducer(new IterableProducer(o, it));
}
真相大白了，在这里做了迭代。还有一个操作符just，其实底层里面调的就是from，只不过还限制了参数个数，而且参数类型必须相同，感觉用处不大
第三个操作符repeat
Observable.just(1, 2).repeat(4).subscribe(new Action1() {
@Override
public void call(Integer integer) {
System.out.println(integer);
}
});
repeat点进去是OnSubcribRedo.repeat，紧追着count这个参数，会看到一个RedoFinite类
public static final class RedoFinite implements Func1extends Notification>, Observable> {
private final long count;
public RedoFinite(long count) {
this.count = count;
}
@Override
public Observable call(Observableextends Notification> ts) {
return ts.map(new Func1, Notification>() {
int num=0;
@Override
public Notification call(Notification terminalNotification) {
if(count == 0) {
return terminalNotification;
}
num++;
if(num count) {
return Notification.createOnNext(num);
} else {
return terminalNotification;
}
}
}).dematerialize();
}
}
这里就看到了，有个num++和num
Original: https://www.cnblogs.com/anni-qianqian/p/13959959.html
Author: 嘉禾世兴
Title: RxJava简析
原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/538945/
转载文章受原作者版权保护。转载请注明原作者出处！