说到区别,可能有的小伙伴会问,我没看过rxjava1。可以直接看rxjava2么。个人觉得不必要,因为 rxjava2.x 是按照 Reactive-Streams specification 规范完全的重写的,完全独立于 rxjava1.x 而存在,它改变了以往 rxjava1的用法。换句话说,我学java需不需要先学C语言一样。
那么两者的区别体现在哪呢?主要是如下几个方面:
-
空指针问题这应该是一个很大的变化,用过rxjava1的人都知道,我们可以在发射事件的时候传入NULL。但这在rxjava2中是不存在的。不信你试试?分分钟给你来一个NullPointerExpection。
-
Function相关的在rxjava1中,我们有各种Func1,Func2…,但在rxjava2中只有Function了。依旧记得看凯哥的文章的时候把我整蒙了。愣是没发现,后来才注意到被替换了。并且,他们都增加了throw exception。
-
背压—backpressure 关于backpressure,这个就厉害了。厉害到我都不懂了。好了,开个玩笑,我们继续说。我们知道在Rxjava1中Observable对backpressure是支持的。但在Rxjava2中Observable取消了对backpressure的支持。并且引进了一个叫做Flowable的来支持backpressure。
那么什么是背压: 听不懂的含义:上游的生产速度大于下游的处理速度,导致下游处理不急,这种操作被称为backpressure。
这种情况看似很常见,但实际上,这种情况并不常见,或者可以说成非常罕见。那么遇到了怎么办?如果它出现了,直接丢弃。what the fuck?你tm在逗我?但事实就是这样,如果我们在开发过程中,遇到了backpressure,我们就应该丢弃它。
听得懂的含义:对于可丢弃的事件,上游生产速度过快导致事件堆积,当堆积到超出buffer上限的时候,就叫做backpressure。
处理方案是什么: 1、丢弃新事件;2、不丢弃,继续堆积。(忽略了backpressure,相当于Observable)。
适合backpressure的情况: 在线直播流:比如说,正在直播的时候,突然网络出现了卡顿,页面卡住了。那么当网络好了之后肯定不会是在接着之前的页面继续的,就相当于,你网络卡了多久,他就丢弃了多长时间的数据。
backpressure的关键点是什么:不可控,可丢弃。
基本使用
讲了一大堆理念知识,接下来就是开工干活了。那么关于Rxjava2的基本实现主要是三点:创建Observable,创建Observer,进行绑定。那么我们一个个的看。
创建Observable
Observable是什么?观察者还是被观察者?我又忘了。哈哈。开个玩笑,当然是后者了。为什么是先创建Observable而不是Observer?当然了,先后顺序的无所谓的。但是考虑到后面的链式调用。所以我这边就先写了先创建Observable了。
Observable observable = Observable.create(new ObservableOnSubscribe() {
@Override
public void subscribe(ObservableEmitter emitter) throws Exception {
emitter.onNext(“Hello”);
emitter.onNext(“Rxjava2”);
emitter.onNext(“My name is Silence”);
emitter.onNext(“What’s your name”);
//一旦调用onComplete,下面将不在接受事件
emitter.onComplete();
}
});
现在我来解释一下上面的ObservableEmitter到底是什么。字面意思是可观察的发射器。没错,这个就是被观察者用来发送事件的。它可以发出三种类型的事件,通过调用emitter的onNext(T value)、onError(Throwable error)和onComplete()就可以分别发出next事件、error事件和complete事件。至于这三个事件到底什么意思。不急,我们后面说。
创建Observer
现在我们来创建一个观察者,它决定了在观察中到底应该有着什么样的行为操作。
Observer observer = new Observer() {
@Override
public void onSubscribe(Disposable d) {
Log.i(TAG, "onSubscribe: " + d);
result += "onSubscribe: " + d + “\n”;
}
@Override
public void onNext(String string) {
Log.i(TAG, "onNext: " + string);
result += "onNext: " + string + “\n”;
}
@Override
public void onError(Throwable e) {
Log.i(TAG, "onError: " + e);
result += "onError: " + e + “\n”;
}
@Override
public void onComplete() {
Log.i(TAG, "onComplete: ");
result += "onComplete: " + “\n”;
}
};
其中onSubscribe、onNext、onError和onComplete是必要的实现方法,其含义如下:
-
onSubscribe:它会在事件还未发送之前被调用,可以用来做一些准备操作。而里面的Disposable则是用来切断上下游的关系的。
-
onNext:普通的事件。将要处理的事件添加到队列中。
-
onError:事件队列异常,在事件处理过程中出现异常情况时,此方法会被调用。同时队列将会终止,也就是不允许在有事件发出。
-
onComplete:事件队列完成。rxjava不仅把每个事件单独处理。而且会把他们当成一个队列。当不再有onNext事件发出时,需要触发onComplete方法作为完成标识。
进行Subscribe
订阅其实只需要一行代码就够了:
observerable.subscribe(Observer);
运行一个看看效果先:
和之前介绍的一样,先调用onSubscribe,然后走了onNext,最后以onComplete收尾。
神奇的操作符
对于rxjava来说,有一句话,我觉得说的很对,叫做:如果你每天研究一个操作符,最少一个半月,如果你想理解原理。最少半年。换句话说,有关rxjava的知识完全可以写一本书。那么本文肯定不会讲那么细。在这边我会给你们介绍一些常用的操作符。保证日常开发的流程足矣。
创建操作符
一般创建操作符是指,刚开始创建观察者的时候调用的。在基本使用中我已经介绍了create操作符,那么这边我们就要说到just,fromarray和interval了。
just
此操作符是将传入的参数依次发出来。
Observable observable = Observable.just(“Hello”, “Rxjava2”, “My name is Silence”,“What’s your name”);
// 将会依次调用:
// onNext(“Hello”);
// onNext(“Rxjava2”);
// onNext(“My name is Silence”);
// onNext(“What’s your name”);
// onCompleted();
fromarray
将传入的数组通过坐标一次发送出去。
String[] words = {“Hello”, “Rxjava2”, “My name is Silence”,“What’s your name”};
Observable observable = Observable.from(words);
// 将会依次调用:
// onNext(“Hello”);
// onNext(“Rxjava2”);
// onNext(“My name is Silence”);
// onNext(“What’s your name”);
// onCompleted();
interval
这个其实就是定时器,用了它你可以抛弃CountDownTimer了。现在我们看看怎么用:
Observable.interval(2, TimeUnit.SECONDS).subscribe(
new Consumer() {
@Override
public void accept(Long aLong) throws Exception {
Log.i(TAG, "accept: "+aLong.intValue());
}
}
);
我们看看结果:
上面就是我们每隔2s打印一次long的值。
变换操作符
变换操作符的作用是对Observable发射的数据按照一定规则做一些变换操作,然后讲变换后的数据发射出去。变换操作符有map,flatMap,concatMap,switchMap,buffer,groupBy等等。这里我们会讲解最常用的map,flatMap、concatMap以及compose。
map
map操作符通过指定一个Function对象,将Observable转换为一个新的Observable对象并发射,观察者将收到新的Observable处理。直接上代码:
Observable.create(new ObservableOnSubscribe() {
@Override
public void subscribe(ObservableEmitter emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onNe
xt(4);
}
}).map(new Function<Integer, String>() {
@Override
public String apply(Integer integer) throws Exception {
return “This is result " + integer + “\n”;
}
}).subscribe(new Consumer() {
@Override
public void accept(String str) throws Exception {
Log.i(”—>", "accept: "+str);
string += str;
}
});
tv_first.setText(string);
输入结果如下:
仔细看,map()方法中,我们把一个integer对象转换成了一个String对象。然后当map()调用结束时,事件的参数类型也从integer转换成了String。这就是最常见的变换操作。
flatMap
flatmap的操作符是将Observable发射的数据集合变成一个Observable集合。也就是说它可以讲一个观察对象变换成多个观察对象,但是并不能保证事件的顺序。想保证事件的顺序?那你过会看下面降到的concatMap。
那么什么叫作数据集合变成一个Observable集合呢?还是用上面的例子,我有一组integer集合。我想转换成string集合怎么办?那就继续看代码:
Observable.create(new ObservableOnSubscribe() {
@Override
public void subscribe(ObservableEmitter emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
}
}).flatMap(new Function<Integer, ObservableSource>() {
@Override
public ObservableSource apply(Integer integer) throws Exception {
final List list = new ArrayList<>();
for (int i = 0; i < 3; i++) {
list.add(“I am value " + integer + “\n”);
}
return Observable.fromIterable(list);
}
}).subscribe(new Consumer() {
@Override
public void accept(String s) throws Exception {
Log.i(”—>", "accept: "+s);
string += s;
}
});
tv_first.setText(string);
我们来看结果:
打住打住,是不是有问题?WTF?有啥问题?还记不记得我上面说过flatMap不能保证事件执行顺序。那么这边事件为什么都是按顺序执行的?不急,我们在发射事件的时候给他加一个延迟在看看结果:
Observable.create(new ObservableOnSubscribe() {
@Override
public void subscribe(ObservableEmitter emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
}
}).flatMap(new Function<Integer, ObservableSource>() {
@Override
public ObservableSource apply(Integer integer) throws Exception {
final List list = new ArrayList<>();
for (int i = 0; i < 3; i++) {
list.add(“I am value " + integer + “\n”);
}
return Observable.fromIterable(list).delay(100,TimeUnit.MILLISECONDS);
}
}).subscribe(new Consumer() {
@Override
public void accept(String s) throws Exception {
Log.i(”—>", "accept: "+s);
string += s;
}
});
tv_first.setText(string);
我们在当他发射事件的时候给他加一个100ms的延迟看看结果:
看到没有,我说啥的?不能保证执行顺序。所以万事容我慢慢道来。先喝杯茶压压惊。我们在接着往下讲。
concatMap
上面我也介绍了concatMap。除了保证了执行顺序,其他都和concatMap一毛一样。你说保证就保证啊。您先喝杯茶,接着往下看:
Observable.create(new ObservableOnSubscribe() {
@Override
public void subscribe(ObservableEmitter emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
}
}).concatMap(new Function<Integer, ObservableSource>() {
@Override
public ObservableSource apply(Integer integer) throws Exception {
final List list = new ArrayList<>();
for (int i = 0; i < 3; i++) {
list.add(“I am value " + integer + “\n”);
}
return Observable.fromIterable(list).delay(1000,TimeUnit.MILLISECONDS);
// return Observable.fromIterable(list);
}
}).subscribe(new Consumer() {
@Override
public void accept(String s) throws Exception {
Log.i(”—>", "accept: "+s);
string += s;
}
});
tv_first.setText(string);
为了我们能看的更明显一点,我这边直接设置了一秒钟的延迟。下面我们来看效果图:
可以从执行顺序和打印时间看出,的的确确是延迟了一秒钟。
compose
这个操作符就很厉害了。他的变换是怎么做的呢?我们知道rxjava是通过建造者的模式通过链式来调用起来的。那么多个链式就需要多个Observable。而这个操作符就是把多个Observable转化成一个Observable。听起来是不是很厉害~。具体如何操作,我们接着看:
public ObservableTransformer<T, T> applyObservableAsync() {
return new ObservableTransformer<T, T>() {
@Override
public ObservableSource apply(Observable upstream) {
return upstream.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread());
}
};
}
上面代码可以看出,我把子线程和主线程进行了一个封装,然后返回了一个ObservableTransformer对象。那么我们只要这边做就可以了:
Observable.just(1, 2, 3, 4, 5, 6)
.compose(this.applyObservableAsync())
.subscribe(new Consumer() {
@Override
public void accept(Integer strings) throws Exception {
Log.i(“–>”, "accept: " + strings);
string += strings;
}
});
tv_first.setText(string);
过滤操作符
过滤操作符用于过滤和选择Observable发射的数据序列。让Observable只返回满足我们条件的数据。过滤操作符有buffer,filter,skip,take,skipLast,takeLast等等,这边我会介绍到filter,buffer,skip,take,distinct。
filter
filter操作符是对源Observable产生的结果进行有规则的过滤。只有满足规则的结果才会提交到观察者手中。例如:
Observable.just(1,2,3).filter(new Predicate() {
@Override
public boolean test(Integer integer) throws Exception {
return integer < 3;
}
}).subscribe(new Consumer() {
@Override
public void accept(Integer s) throws Exception {
Log.i(“—>”, "accept: " + s);
string += s;
}
});
tv_first.setText(string);
}
代码很简单,我们发送1,2,3;但是我们加上一个filter操作符,让它只返回小于3的的内容。那么我们来看一下结果:
distinct
这个操作符其实就更简单了。比如说,我要在一组数据中去掉重复的内容,就要用到它。也就是去重。它只允许还没有发射的数据项通过。发射过的数据项直接pass。
Observable.just(1,2,3,4,2,3,5,6,1,3)
.distinct().subscribe(new Consumer() {
@Override
public void accept(Integer s) throws Exception {
Log.i(“—>”, "accept: " + s);
string += s;
}
});
tv_first.setText(string);
那么输出结果就很简单了:
buffer
这个其实也不难,主要是缓存,把源Observable转换成一个新的Observable。这个新的Observable每次发射的是一组List,而不是单独的一个个的发送数据源。
Observable.just(1,2,3,4,5,6)
.buffer(2).subscribe(new Consumer<List>() {
@Override
public void accept(List strings) throws Exception {
for (Integer integer : strings) {
Log.i(“–>”, “accept: “+integer);
string+=strings;
}
Log.i(”–>”, “accept: ----------------------->”);
}
});
tv_first.setText(string);
我们让他每次缓存2个,下面我们来看结果:
skip 、take
skip操作符将源Observable发射过的数据过滤掉前n项,而take操作则只取前n项;另外还有skipLast和takeLast则是从后往前进行过滤。先来看看skip操作符。
Observable.just(1, 2, 3, 4, 5, 6)
.skip(2).subscribe(new Consumer() {
@Override
public void accept(Integer strings) throws Exception {
Log.i(“–>”, "accept: " + strings);
string += strings;
}
});
tv_first.setText(string);
结果如下:
接下来我们把skip换成take看看。
Observable.just(1, 2, 3, 4, 5, 6)
.take(3).subscribe(new Consumer() {
@Override
public void accept(Integer strings) throws Exception {
Log.i(“–>”, "accept: " + strings);
string += strings;
}
});
tv_first.setText(string);
结果如下:
组合操作符
merge
merge是将多个操作符合并到一个Observable中进行发射,merge可能让合并到Observable的数据发生错乱。(并行无序)
Observable observable1=Observable.just(1,2,3);
Observable observable2=Observable.just(1,2,3);
Observable.merge(observable1,observable2).subscribe(new Consumer() {
@Override
public void accept(Integer integer) throws Exception {
Log.i(TAG, "accept: "+integer);
}
});
结果如下:
concat
将多个Observable发射的数据进行合并并且发射,和merge不同的是,merge是无序的,而concat是有序的。(串行有序)没有发射完前一个它一定不会发送后一个。
Observable observable1=Observable.just(1,2,3);
Observable observable2=Observable.just(4,5,6);
Observable.concat(observable1,observable2).subscribe(new Consumer() {
@Override
public void accept(Integer integer) throws Exception {
Log.i(TAG, "accept: "+integer);
}
});
结果如下:
zip
此操作符和合并多个Observable发送的数据项,根据他们的类型就行重新变换,并发射一个新的值。
Observable observable1=Observable.just(1,2,3);
Observable observable2=Observable.just(“a”,“b”,“c”);
Observable.zip(observable1, observable2, new BiFunction<Integer, String, String>() {
@Override
public String apply(Integer integer, String s) throws Exception {
return integer+s;
}
}).subscribe(new Consumer() {
@Override
public void accept(String s) throws Exception {
Log.i(TAG, "apply: "+s);
}
});
结果如下:
concatEager
前面说道串行有序,而concatEager则是并行且有序。我们来看看如果修改:
Observable observable1=Observable.just(1,2,3);
Observable observable2=Observable.just(“a”,“b”,“c”);
Observable.concatEager(Observable.fromArray(observable1,observable2)).subscribe(new Consumer() {
@Override
public void accept(Serializable serializable) throws Exception {
Log.i(TAG, "accept: "+serializable);
}
});
最后
小编这些年深知大多数初中级Android工程师,想要提升自己,往往是自己摸索成长,自己不成体系的自学效果低效漫长且无助。
因此我收集整理了一份《2024年Android移动开发全套学习资料》,初衷也很简单,就是希望能够帮助到想自学提升又不知道该从何学起的朋友。
一个人可以走的很快,但一群人才能走的更远!不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人
都欢迎加入我们的的圈子(技术交流、学习资源、职场吐槽、大厂内推、面试辅导),让我们一起学习成长!
资料⬅专栏获取
servable.concatEager(Observable.fromArray(observable1,observable2)).subscribe(new Consumer() {
@Override
public void accept(Serializable serializable) throws Exception {
Log.i(TAG, "accept: "+serializable);
}
});
最后
小编这些年深知大多数初中级Android工程师,想要提升自己,往往是自己摸索成长,自己不成体系的自学效果低效漫长且无助。
因此我收集整理了一份《2024年Android移动开发全套学习资料》,初衷也很简单,就是希望能够帮助到想自学提升又不知道该从何学起的朋友。
[外链图片转存中…(img-drtOPxRx-1719092632705)]一个人可以走的很快,但一群人才能走的更远!不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人
都欢迎加入我们的的圈子(技术交流、学习资源、职场吐槽、大厂内推、面试辅导),让我们一起学习成长!
资料⬅专栏获取