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 Consumer<? super T> onNext, @NonNull Consumer<? super Throwable> onError,
@NonNull Action onComplete) {
Objects.requireNonNull(onNext, "onNext is null");
Objects.requireNonNull(onError, "onError is null");
Objects.requireNonNull(onComplete, "onComplete is null");
LambdaSubscriber<T> ls = new LambdaSubscriber<>(onNext, onError, onComplete, FlowableInternalHelper.RequestMax.INSTANCE);
subscribe(ls);
return ls;
}
看上去最重要的就是这两句了
LambdaSubscriber<T> ls = new LambdaSubscriber<>(onNext, onError, onComplete, FlowableInternalHelper.RequestMax.INSTANCE);
subscribe(ls);
先进到subscribe(ls)中,发现这句
subscribeActual(flowableSubscriber)
跳进去发现是个抽象方法,那么实现肯定在子类啦,进到子类FlowableFromArray
@Override
public void subscribeActual(Subscriber<? super T> s) {
if (s instanceof ConditionalSubscriber) {
s.onSubscribe(new ArrayConditionalSubscription<>(
(ConditionalSubscriber<? super 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<String>() {
@Override
public void accept(String s) throws Throwable {
System.out.println("hello " + s + "!");
}
});
是不是很麻烦,饶了一大圈,没关系,我们继续往下看
这里给出一些名词的翻译
下面是常用的操作符列表:
下面我们来看第一个操作符:Create
Observable.create(new Observable.OnSubscribe<Integer>() {
@Override
public void call(Subscriber<? super 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<Integer>() {
@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 <T> Observable<T> create(OnSubscribe<T> f) {
return new Observable<T>(hook.onCreate(f));
}
这里没什么,就是返回创建一个Observable对象,但是要注意里面的参数OnSubscribe
public interface OnSubscribe<T> extends Action1<Subscriber<? super T>> {
// cover for generics insanity
}
public interface Action1<T> extends Action {
void call(T t);
}
这个参数是一个接口,它的父类里有个抽象待实现的方法call,而且call方法被传了Subscriber进去
我们来看Subscriber这个类,原来是个接口,而且它的父类Observer有三个很重要的方法
public interface Observer<T> {
void onCompleted();
void onError(Throwable e);
void onNext(T t);
}
第一个create方法算是完成了,我们可以拆分来看
Observable<Integer> integerObservable = Observable.create(new Observable.OnSubscribe<Integer>() {
@Override
public void call(Subscriber<? super 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<Integer>() {
@Override
public void call(Integer integer) {
System.out.println(integer);
}
});
先看Observable的from方法
public final static <T> Observable<T> from(T[] array) {
return from(Arrays.asList(array));
}
其实就是把数组转成list,但是再点from进去就很重要
public final static <T> Observable<T> from(Iterable<? extends T> iterable) {
return create(new OnSubscribeFromIterable<T>(iterable));
}public OnSubscribeFromIterable(Iterable<? extends 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 Subscriber<? super T> o) {
final Iterator<? extends T> it = is.iterator();
if (!it.hasNext() && !o.isUnsubscribed())
o.onCompleted();
else
o.setProducer(new IterableProducer<T>(o, it));
}
真相大白了,在这里做了迭代。还有一个操作符just,其实底层里面调的就是from,只不过还限制了参数个数,而且参数类型必须相同,感觉用处不大
第三个操作符repeat
Observable.just(1, 2).repeat(4).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
System.out.println(integer);
}
});
repeat点进去是OnSubcribRedo.repeat,紧追着count这个参数,会看到一个RedoFinite类
public static final class RedoFinite implements Func1<Observable<? extends Notification<?>>, Observable<?>> {
private final long count;
public RedoFinite(long count) {
this.count = count;
}
@Override
public Observable<?> call(Observable<? extends Notification<?>> ts) {
return ts.map(new Func1<Notification<?>, 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<=count判断,就知道是怎么重复的了
第4个操作符Map和flapMap
这两个变换操作符可谓非常重要,经常用到,我写了4个例子,请仔细区别,就可以知道它们到底做了什么
Student student1 = new Student("stark", new Course[]{new Course("Chinese"), new Course("English")});
Student student2 = new Student("adam", new Course[]{new Course("Math"), new Course("Physical")});
Student[] students = new Student[]{student1, student2};
Observable.from(students).subscribe(new Action1<Student>() {
@Override
public void call(Student student) {
System.out.println(student.getName());
}
});
System.out.println("-------------");
Observable.from(students).map(new Func1<Student, String>() {
@Override
public String call(Student student) {
return student.getName();
}
}).subscribe(new Action1<String>() {
@Override
public void call(String name) {
System.out.println(name);
}
});
System.out.println("-------------");
Observable.from(students).map(new Func1<Student, Course[]>() {
@Override
public Course[] call(Student student) {
return student.getCourses();
}
}).subscribe(new Action1<Course[]>() {
@Override
public void call(Course[] courses) {
System.out.println(courses[0].getName());
System.out.println(courses[1].getName());
}
});
System.out.println("-------------");
Observable.from(students).flatMap(new Func1<Student, Observable<Course>>() {
@Override
public Observable<Course> call(Student student) {
return Observable.from(student.getCourses());
}
}).subscribe(new Action1<Course>() {
@Override
public void call(Course course) {
System.out.println(course.getName());
}
});
输出:
stark
adam
-------------
stark
adam
-------------
Chinese
English
Math
Physical
-------------
Chinese
English
Math
Physical
如果你仔细看代码,就会发现map就是一对一的转换,flatMap是一对多的转换,转换的前后类型在方法Func1中已经标的很清楚。例子:Func1(Student,String)就代表传参是Student,返回类型是String,具体的实现在call里面student.getName()
map和flatMap可以看作是将我们经常用到的嵌套循环for(i){for(j)...}...给解耦了,看起来更清楚一些,中间可以插入更多的操作
源码里面的实现就是迭代,没什么好说
第5个操作符filter:
Observable.just(1,2,3,4,5).filter(new Func1<Integer, Boolean>() {
@Override
public Boolean call(Integer integer) {
return integer<4;
}
}).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
System.out.println(integer);
}
});
先过滤再循环输出
第6个组合操作符and/then/when
implementation 'io.reactivex:rxjava-joins:0.22.0'
Observable<String> just1 = Observable.just("A", "B");
Observable<Integer> just2 = Observable.just(1, 2, 3);
Pattern2<String, Integer> pattern = JoinObservable.from(just1).and(just2);
Plan0<String> plan = pattern.then(new Func2<String, Integer, String>() {
@Override
public String call(String s, Integer integer) {
return s + integer;
}
});
JoinObservable.when(plan).toObservable().subscribe(new Action1<String>() {
@Override
public void call(String s) {
System.out.println(s);
}
});
输出:
A1
B2
第7个组合操作符merge:
Observable<Integer> odds = Observable.just(1, 3, 5);
Observable<Integer> evens = Observable.just(2, 4, 6);
Observable.merge(odds,evens).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
System.out.println(integer);
}
});
输出:
1
3
5
2
4
6
第8个操作符doOnNext:
Observable.just(1, 2, 3).doOnNext(new Action1<Integer>() {
@Override
public void call(Integer integer) {
if (integer > 1) {
throw new RuntimeException("item exceeds maximum value");
}
}
}).subscribe(new Subscriber<Integer>() {
@Override
public void onCompleted() {
System.out.println("onCompleted");
}
@Override
public void onError(Throwable e) {
System.out.println("onError");
}
@Override
public void onNext(Integer integer) {
System.out.println("next:" + integer);
}
});
输出:
next:1
onError
第9个操作符SubscribeOn(Scheduler):即申明在哪个调度器工作
第10个:android例子:
Observable.from(new String[]{"one", "two", "three", "four", "five"})
.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
System.out.println(s);
}
});