RxJava2.0的使用详解
1,初识RxJava
RxJava就是一种用Java语言实现的响应式编程，来创建基于事件的异步程序
RxJava是一个基于事件订阅的异步执行的一个类库,目前比较火的一些技术框架！
参考资料:
Github上RxJava的项目地址:
https://github.com/ReactiveX/RxJava
技术文档Api:
http://reactivex.io/RxJava/javadoc/
RxAndroid，用于 Android 开发:
https://github.com/ReactiveX/RxAndroid
简书博客推荐:
http://www.jianshu.com/p/ba61c047c230
1.1使用前所添加的依赖(build.gradle):
compile 'io.reactivex.rxjava2:rxjava:2.1.3'
compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
1.2作用:
RxJava的目的就是异步。
RxJava的特点就是可以非常简便的实现异步调用，可以在逻辑复杂的代码逻辑中以比较轻易的方式实现异步调用。随着逻辑的复杂，需求的更改，代码可依然能保持极强的阅读性
1.3概念:
RxJava是利用观察者模式来实现一些列的操作，所以对于观察者模式中的观察者，被观察者，以及订阅、事件需要有一个了解.
Observable：在观察者模式中称为“被观察者”；
Observer：观察者模式中的“观察者”，可接收Observable发送的数据；
subscribe：订阅，观察者与被观察者，通过Observable的subscribe()方法进行订阅；
Subscriber：也是一种观察者，在2.0中 它与Observer没什么实质的区别，不同的是 Subscriber要与Flowable(也是一种被观察者)联合使用，该部分 内容是2.0新增的，后续文章再介绍。Obsesrver用于订阅Observable，而Subscriber用于订阅Flowable.
1.4观察者模式的理解:
A 对象（观察者）对 B 对象（被观察者）的某种变化高度敏感,需要在 B 变化的一瞬间做出反应.
在程序的观察者模式，观察者不需要时刻盯着被观察者,而是采用注册或者称为订阅的方式，告诉被观察者：我需要你的某某状态，你要在它变化的时候通知我!
RxJava 有四个基本概念：
Observable (被观察者)、
Observer (观察者)、
subscribe (订阅)
Observable 和 Observer 通过 subscribe() 方法实现订阅关系，从而 Observable 可以在完成某些操作，获得一些结果后，回调触发事件，即 发出事件来通知 Observer。
关于回调，如果理解则可以跳过这一段，如果不理解，在RxJava中可以简单的理解为：为了方便Observable和Observer交互，在Observable中，将 Observer对象传入，在完成某些操作后调用Observer对象的方法，此时将触发Observer中具体实现的对应方法。
注意：Observer是个接口，Observable是个类。
RxJava中定义的事件方法：
onNext(),普通事件,按照队列依次进行处理.
onComplete(),事件队列完结时调用该方法
onError(),事件处理过程中出现异常时，onError()触发，同时队列终止,不再有事件发出.
onSubscribe(),RxJava 2.0 中新增的，传递参数为Disposable,可用于切断接收事件
让Observable (被观察者)开启子线程执行耗操作，完成耗时操作后，触发回调，通知Observer (观察者)进行主线程UI更新
2,简单使用步骤:
步骤:
创建数据发射源,上游Observable
创建数据接收处,下游Observer
数据源关联接收处,上游衔接下游!
3,Observable
数据发射源,可观察的,被观察的,
Observable有两种形式启动形式:
1热启动Observable任何时候都会发送消息，即使没有任何观察者监听它。
2冷启动Observable只有在至少有一个订阅者的时候才会发送消息
Observable的几种创建方式:
01,just()方式
使用just( )，将创建一个Observable并自动调用onNext( )发射数据。
也就是通过just( )方式 直接触发onNext()，just中传递的参数将直接在Observer的onNext()方法中接收到。
02,fromIterable()方式
使用fromIterable()，遍历集合，发送每个item.多次自动调用onNext()方法，每次传入一个item.
注意：Collection接口是Iterable接口的子接口，所以所有Collection接口的实现类都可以作为Iterable对象直接传入fromIterable() 方法。
03,defer()方式
当观察者订阅时,才创建Observable，并且针对每个观察者创建都是一个新的Observable.
通过Callable中的回调方法call(),决定使用以何种方式来创建这个Observable对象,当订阅后，发送事件.
04,interval( )方式
创建一个按固定时间间隔发射整数序列的Observable，可用作定时器。按照固定时间间隔来调用onNext()方法。
05,timer( )方式
通过此种方式创建一个Observable,它在一个给定的延迟后发射一个特殊的值，即表示延迟指定时间后，调用onNext()方法。
06,range( )方式,range(x,y)
创建一个发射特定整数序列的Observable，第一个参数x为起始值，第二个y为发送的个数，如果y为0则不发送，y为负数则抛异常。
range(1,5)
上述表示发射1到5的数。即调用5次Next()方法，依次传入1-5数字。
07,repeat( )方式
创建一个Observable，该Observable的事件可以重复调用。
部分方法介绍:
表示下游不关心任何事件,你上游尽管发你的数据
Disposable subscribe()
表示下游只关心onNext事件,其他不管
Disposable subscribe(Consumer<? super T> onNext)
表示下游只关心onNext事件,onError事件
Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError)
表示只关心onNext事件,onError事件,onComplete事件
Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,Action onComplete)
表示处理所有事件
subscribe(Observer<? super T> observer)
4,ObservableEmitter
Emitter是发射器的意思,就是用来发出事件的，它可以发出三种类型的事件
通过调用onNext(T value),发出next事件
通过调用onComplete(),发出complete事件
通过调用onError(Throwable error),发出error事件
注意事项:
onComplete和onError唯一并且互斥
发送多个onComplete, 第一个onComplete接收到,就不再接收了.
发送多个onError, 则收到第二个onError事件会导致程序会崩溃.
不可以随意乱七八糟发射事件，需要满足一定的规则：
上游可以发送无限个onNext, 下游也可以接收无限个onNext.
当上游发送了一个onComplete后, 上游onComplete之后的事件将会继续发送, 而下游收到onComplete事件之后将不再继续接收事件.
上游发送了一个onError后, 上游onError之后的事件将继续发送, 而下游收到onError事件之后将不再继续接收事件.
上游可以不发送onComplete或onError.
最为关键的是onComplete和onError必须唯一并且互斥, 即不能发多个onComplete, 也不能发多个onError, 也不能先发一个onComplete, 然后再发一个onError
5,Disposable
一次性,它理解成两根管道之间的一个机关, 当调用它的dispose()方法时, 它就会将两根管道切断, 从而导致下游收不到事件.
在RxJava中,用它来切断Observer(观察者)与Observable(被观察者)之间的连接，当调用它的dispose()方法时, 它就会将Observer(观察者)与Observable(被观察者)之间的连接切断, 从而导致Observer(观察者)收不到事件。
注意: 调用dispose()并不会导致上游不再继续发送事件, 上游会继续发送剩余的事件
我们让上游依次发送1,2,3,complete,4,在下游收到第二个事件之后, 切断水管, 看看运行结果
Disposable的对象通过观察者获得，具体分为两种方式
1,Observer接口
Observer<String> observer = new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
//此方法接收到Disposable的实例!
}
@Override
public void onNext(String s) {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
};
通过创建Observer(观察者)接口,重写onSubscribe方法,当订阅后，建立与Observable(被观察者)的联系后，在onSubscribe(Disposable d)方法中便可以获得Disposable对象。
2.Consumer等其他函数式接口
Disposable disposable = Observable.just("你好").subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
}
});
当调用Observable的subscribe()方法后直接返回一个Disposable 对象
6,线程控制——Scheduler
在不指定线程的情况下， RxJava 遵循的是线程不变的原则，即：在哪个线程调用 subscribe()，就在哪个线程生产事件；在哪个线程生产事件，就在哪个线程消费事件。如果需要切换线程，就需要用到 Scheduler （调度器）。
Schedulers.immediate():
直接在当前线程运行，相当于不指定线程。这是默认的Scheduler。
Schedulers.newThread():
总是启用新线程，并在新线程执行操作。
Schedulers.io(): I/O
操作（读写文件、读写数据库、网络信息交互等）所使用的Scheduler。行为模式和newThread()差不多，区别在于io()的内部实现是是用一个无数量上限的线程池，可以重用空闲的线程，因此多数情况下io()比newThread()更有效率。不要把计算工作放在io()中，可以避免创建不必要的线程。
Schedulers.computation():
计算所使用的Scheduler。这个计算指的是 CPU 密集型计算，即不会被 I/O 等操作限制性能的操作，例如图形的计算。这个Scheduler使用的固定的线程池，大小为 CPU 核数。不要把 I/O 操作放在computation()中，否则 I/O 操作的等待时间会浪费 CPU。
AndroidSchedulers.mainThread()，
Android专用线程，指定操作在主线程运行。
如何切换线程呢？RxJava中提供了两个方法：
subscribeOn() 和 observeOn() ，
两者的不同点在于：
subscribeOn(): 指定subscribe()订阅所发生的线程，或者叫做事件产生的线程。
observeOn(): 指定Observer所运行在的线程，即onNext()执行的线程。或者叫做事件消费的线程。
7,以Consumer为例，我们可以实现简便式的观察者模式
Observable.just("hello").subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
System.out.println(s);
}
});
其中Consumer中的accept()方法接收一个来自Observable的单个值。Consumer就是一个观察者。其他函数式接口可以类似应用
8,RxJava中的操作符
01,操作符就是用于在Observable和最终的Observer之间，通过转换Observable为其他观察者对象的过程，修改发出的事件，
最终将最简洁的数据传递给Observer对象.
每次调用一次操作符，就进行一次观察者对象的改变，同时将需要传递的数据进行转变，最终Observer对象获得想要的数据。
以网络加载为例，我们通过Observable开启子线程，进行一些网络请求获取数据的操作，获得到网络数据后，然后通过操作符进行转换，获得我们想要的形式的数据，然后传递给Observer对象
02,比较常用的操作符:
map()操作符
map()操作符，就是把原来的Observable对象转换成另一个Observable对象，同时将传输的数据进行一些灵活的操作，方便Observer获得想要的数据形式。
举例:
Observable<Integer> observable = Observable
.just("hello")
.map(new Function<String, Integer>() {
@Override
public Integer apply(String s) throws Exception {
return s.length();
}
});
flatMap()操作符
flatMap()对于数据的转换比map()更加彻底，如果发送的数据是集合，flatmap()重新生成一个Observable对象，并把数据转换成Observer想 要的数据形式。它可以返回任何它想返回的Observable对象。
举例:
Observable.just(list)
.flatMap(new Function<List<String>, ObservableSource<?>>() {
@Override
public ObservableSource<?> apply(List<String> strings) throws Exception {
return Observable.fromIterable(strings);
}
});
filter()操作符
filter()操作符根据它的test()方法中，根据自己想过滤的数据加入相应的逻辑判断，返回true则表示数据满足条件，返回false则表示数据需要被过滤。
最后过滤出的数据将加入到新的Observable对象中，方便传递给Observer想要的数据形式。
举例:
Observable
.just(list)
.flatMap(new Function<List<String>, ObservableSource<?>>() {
@Override
public ObservableSource<?> apply(List<String> strings) throws Exception {
return Observable.fromIterable(strings);
}
}).filter(new Predicate<Object>() {
@Override
public boolean test(Object s) throws Exception {
String newStr = (String) s;
if (newStr.charAt(5) - '0' > 5) {
return true;
}
return false;
}
}).subscribe(new Consumer<Object>() {
@Override
public void accept(Object o) throws Exception {
System.out.println((String)o);
}
});
take()操作符
输出最多指定数量的结果.(接收指定数量的结果)
举例:
Observable.just(new ArrayList<String>(){
{
for (int i = 0; i < 8; i++) {
add("data"+i);
}
}
}).flatMap(new Function<List<String>, ObservableSource<?>>() {
@Override
public ObservableSource<?> apply(List<String> strings) throws Exception {
return Observable.fromIterable(strings);
}
}).take(5).subscribe(new Consumer<Object>() {
@Override
public void accept(Object s) throws Exception {
DemonstrateUtil.showLogResult(s.toString());
}
});
doOnNext()
允许我们在每次输出一个元素之前做一些额外的事情
举例:
Observable.just(new ArrayList<String>(){
{
for (int i = 0; i < 6; i++) {
add("data"+i);
}
}
}).flatMap(new Function<List<String>, ObservableSource<?>>() {
@Override
public ObservableSource<?> apply(List<String> strings) throws Exception {
return Observable.fromIterable(strings);
}
}).take(5).doOnNext(new Consumer<Object>() {
@Override
public void accept(Object o) throws Exception {
DemonstrateUtil.showLogResult("额外的准备工作!");
}
}).subscribe(new Consumer<Object>() {
@Override
public void accept(Object s) throws Exception {
DemonstrateUtil.showLogResult(s.toString());
}
});
8,Flowable的理解
Flowable是一个被观察者，与Subscriber(观察者)配合使用，解决Backpressure问题
Backpressure(背压)。所谓背压，即生产者的速度大于消费者的速度带来的问题。
什么情况下才会产生Backpressure问题？
1.如果生产者和消费者在一个线程的情况下，无论生产者的生产速度有多快，每生产一个事件都会通知消费者，等待消费者消费完毕，再生产下一个事件。
所以在这种情况下，根本不存在Backpressure问题。即同步情况下，Backpressure问题不存在。
2.如果生产者和消费者不在同一线程的情况下，如果生产者的速度大于消费者的速度，就会产生Backpressure问题。
即异步情况下，Backpressure问题才会存在。
现象演示说明:
被观察者是事件的生产者,观察者是事件的消费者.假如生产者无限生成事件,而消费者以很缓慢的节奏来消费事件,会造成事件无限堆积,形成背压,最后造成OOM!
Flowable悠然而生，专门用来处理这类问题。
Flowable是为了应对Backpressure而产生的。Flowable是一个被观察者，
与Subscriber(观察者)配合使用，解决Backpressure问题。
注意：处理Backpressure的策略仅仅是处理Subscriber接收事件的方式，并不影响Flowable发送事件的方法。
即使采用了处理Backpressure的策略，Flowable原来以什么样的速度产生事件，现在还是什么样的速度不会变化，主要处理的是Subscriber接收事件的方式。
处理Backpressure问题的策略,或者来解决Backpressure问题
BackpressureStrategy.ERROR
如果缓存池溢出,就会立刻抛出MissingBackpressureException异常
request()用来向生产者申请可以消费的事件数量,这样我们便可以根据本身的消费能力进行消费事件.
虽然并不限制向request()方法中传入任意数字，但是如果消费者并没有这么多的消费能力，依旧会造成资源浪费，最后产生OOM
at java.lang.OutOfMemoryError.<init>(OutOfMemoryError.java:33)
在异步调用时，RxJava中有个缓存池，用来缓存消费者处理不了暂时缓存下来的数据，缓存池的默认大小为128，即只能缓存128个事件。
无论request()中传入的数字比128大或小，缓存池中在刚开始都会存入128个事件。
当然如果本身并没有这么多事件需要发送，则不会存128个事件。
应用举例:
BackpressureStrategy.BUFFER
是把RxJava中默认的只能存128个事件的缓存池换成一个大的缓存池,支持存更多的数据.
消费者通过request()即使传入一个很大的数字，生产者也会生产事件,并将处理不了的事件缓存.
注意:
这种方式任然比较消耗内存，除非是我们比较了解消费者的消费能力，能够把握具体情况，不会产生OOM。
BUFFER要慎用
BackpressureStrategy.DROP
顾名思义,当消费者处理不了事件，就丢弃!
例如,当数据源创建了200个事件,先不进行消费临时进行缓存实际缓存128个,我们第一次申请消费了100个,再次申请消费100个,
那么实际只消费了128个,而其余的72个被丢弃了!
BackpressureStrategy.LATEST
LATEST与DROP功能基本一致,当消费者处理不了事件，就丢弃!
唯一的区别就是LATEST总能使消费者能够接收到生产者产生的最后一个事件。
例如,当数据源创建了200个事件,先不进行消费临时进行缓存,我们第一次申请消费了100个,再次申请消费100个,
那么实际只消费了129个,而其余的71个被丢弃了,但是第200个(最后一个)会被消费.
BackpressureStrategy.MISSING
生产的事件没有进行缓存和丢弃,下游接收到的事件必须进行消费或者处理!
在RxJava中会经常遇到一种情况就是被观察者发送消息十分迅速以至于观察者不能及时的响应这些消息
举例:
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
while (true){
e.onNext(1);
}
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Thread.sleep(2000);
System.out.println(integer);
}
});
被观察者是事件的生产者，观察者是事件的消费者。上述例子中可以看出生产者无限生成事件，而消费者每2秒才能消费一个事件，这会造成事件无限堆积，最后造成OOM。
Flowable就是由此产生，专门用来处理这类问题
代码实现：
public class RxJavaDemo1Activity extends AppCompatActivity implements View.OnClickListener {
protected Button btnSend1;
protected Button btnSend2;
protected Button btnSend3;
protected Button btnSend4;
protected Button btnSend5;
protected Button btnSend6;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
super.setContentView(R.layout.activity_rx_java_demo1);
initView();
}
@Override
public void onClick(View view) {
if (view.getId() == R.id.btn_send1) {
test1();//普通使用
} else if (view.getId() == R.id.btn_send2) {
test2();//链式调用
} else if (view.getId() == R.id.btn_send3) {
test3();//发送中,中断.
} else if (view.getId() == R.id.btn_send4) {
test4();//只关心onnext事件的操作
} else if (view.getId() == R.id.btn_send5) {
test5();//几种被观察者的创建方式
} else if (view.getId() == R.id.btn_send6) {
test6();//常用的操作符
}
}
private void test6() {
DialogUtil.showListDialog(this, "rxjava的操作符号使用", new String[]{
"0map()操作符",
"1flatMap()操作符",
"2filter()操作符",
"3take()操作符",
"4doOnNext()操作符",
}, new DialogInterface.OnClickListener() {
@Override
public void onClick(DialogInterface dialog, int which) {
switch (which) {
case 0:
map0();
break;
case 1:
map1();
break;
case 2:
map2();
break;
case 3:
map3();
break;
case 4:
map4();
break;
}
}
});
}
private void map4() {
Observable.just(new ArrayList<String>(){
{
for (int i = 0; i < 6; i++) {
add("data"+i);
}
}
}).flatMap(new Function<List<String>, ObservableSource<?>>() {
@Override
public ObservableSource<?> apply(List<String> strings) throws Exception {
return Observable.fromIterable(strings);
}
}).take(5).doOnNext(new Consumer<Object>() {
@Override
public void accept(Object o) throws Exception {
DemonstrateUtil.showLogResult("额外的准备工作!");
}
}).subscribe(new Consumer<Object>() {
@Override
public void accept(Object s) throws Exception {
DemonstrateUtil.showLogResult(s.toString());
}
});
}
private void map3() {
Observable.just(new ArrayList<String>(){
{
for (int i = 0; i < 8; i++) {
add("data"+i);
}
}
}).flatMap(new Function<List<String>, ObservableSource<?>>() {
@Override
public ObservableSource<?> apply(List<String> strings) throws Exception {
return Observable.fromIterable(strings);
}
}).take(10).subscribe(new Consumer<Object>() {
@Override
public void accept(Object s) throws Exception {
DemonstrateUtil.showLogResult(s.toString());
}
});
}
private void map2() {
Observable
.just(new ArrayList<String>(){
{
for (int i = 0; i < 5; i++) {
add("data"+i);
}
}
})
.flatMap(new Function<List<String>, ObservableSource<?>>() {
@Override
public ObservableSource<?> apply(List<String> strings) throws Exception {
return Observable.fromIterable(strings);
}
}).filter(new Predicate<Object>() {
@Override
public boolean test(Object s) throws Exception {
String newStr = (String) s;
if (newStr.contains("3")){
return true;
}
return false;
}
}).subscribe(new Consumer<Object>() {
@Override
public void accept(Object o) throws Exception {
DemonstrateUtil.showLogResult((String)o);
}
});
}
private void map1() {
Observable.just(new ArrayList<String>(){
{
for (int i = 0; i < 3; i++) {
add("data"+i);
}
}
}).flatMap(new Function<List<String>, ObservableSource<?>>() {
@Override
public ObservableSource<?> apply(List<String> strings) throws Exception {
return Observable.fromIterable(strings);
}
}).subscribe(new Observer<Object>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Object o) {
DemonstrateUtil.showLogResult("flatMap转换后,接收到的"+o);
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
}
private void map0() {
Observable.just("hellorxjava")
.map(new Function<String, Integer>() {
@Override
public Integer apply(String s) throws Exception {
return s.length();
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Integer integer) {
DemonstrateUtil.showLogResult("接收到被转换的数据结果:"+integer);
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
}
private void test5() {
DialogUtil.showListDialog(this, "rxjava的其他操作", new String[]{
"0just()方式创建Observable",
"1fromIterable()方式创建Observable",
"2defer()方式创建Observable",
"3interval( )方式创建Observable",
"4timer( )方式创建Observable",
"5range( )方式创建Observable",
"6repeat( )方式创建Observable",
}, new DialogInterface.OnClickListener() {
@Override
public void onClick(DialogInterface dialog, int which) {
switch (which) {
case 0:
other0();
break;
case 1:
other1();
break;
case 2:
other2();
break;
case 3:
other3();
break;
case 4:
other4();
break;
case 5:
other5();
break;
case 6:
other6();
break;
}
}
});
}
private void other6() {
Observable.just(123).repeat().subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Integer integer) {
DemonstrateUtil.showLogResult("重复integer" + integer);
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
}
private void other5() {
Observable.range(1, 5).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Integer integer) {
DemonstrateUtil.showLogResult("连续收到:" + integer);
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
}
private void other4() {
Observable.timer(5, TimeUnit.SECONDS).subscribe(new Observer<Long>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Long aLong) {
DemonstrateUtil.showLogResult("延迟5s后调用了:onNext");
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
}
private void other3() {
Observable.interval(3, TimeUnit.SECONDS).subscribe(new Observer<Long>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Long aLong) {
DemonstrateUtil.showLogResult("数字是:" + aLong);
//DemonstrateUtil.showToastResult(RxJavaDemo1Activity.this,"数字是:"+aLong);
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
}
private void other2() {
Observable<String> observable = Observable.defer(new Callable<ObservableSource<? extends String>>() {
@Override
public ObservableSource<? extends String> call() throws Exception {
return Observable.just("hello,defer");
}
});
//上游衔接下游!
observable.subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String s) {
DemonstrateUtil.showLogResult(s);
DemonstrateUtil.showToastResult(RxJavaDemo1Activity.this, s);
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
}
private void other1() {
Observable.fromIterable(new ArrayList<String>() {
{
for (int i = 0; i < 5; i++) {
add("Hello," + i);
}
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String s) {
DemonstrateUtil.showToastResult(RxJavaDemo1Activity.this, s);
DemonstrateUtil.showLogResult(s);
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
}
private void other0() {
Observable.just("hello,you hao!").subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String s) {
DemonstrateUtil.showLogResult(s);
DemonstrateUtil.showToastResult(RxJavaDemo1Activity.this, s);
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
}
private void test4() {
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
DemonstrateUtil.showLogResult("emitter 1");
emitter.onNext(1);
DemonstrateUtil.showLogResult("emitter 2");
emitter.onNext(2);
DemonstrateUtil.showLogResult("emitter 3");
emitter.onNext(3);
DemonstrateUtil.showLogResult("complete");
emitter.onComplete();
DemonstrateUtil.showLogResult("emitter 4");
emitter.onNext(4);
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
DemonstrateUtil.showLogResult("accept:" + integer);
}
});
}
private void test3() {
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
DemonstrateUtil.showLogResult("emitter 1");
emitter.onNext(1);
DemonstrateUtil.showLogResult("emitter 2");
emitter.onNext(2);
DemonstrateUtil.showLogResult("emitter 3");
emitter.onNext(3);
DemonstrateUtil.showLogResult("complete");
emitter.onComplete();
DemonstrateUtil.showLogResult("emitter 4");
emitter.onNext(4);
}
}).subscribe(new Observer<Integer>() {
private Disposable mDisposable;
private int i;
@Override
public void onSubscribe(Disposable d) {
DemonstrateUtil.showLogResult("subscribe");
mDisposable = d;
}
@Override
public void onNext(Integer value) {
DemonstrateUtil.showLogResult("onNext:" + value);
i++;
if (i == 2) {
DemonstrateUtil.showLogResult("dispose:" + value);
mDisposable.dispose();
DemonstrateUtil.showLogResult("isDisposed : " + mDisposable.isDisposed());
}
}
@Override
public void onError(Throwable e) {
DemonstrateUtil.showLogResult("error:");
}
@Override
public void onComplete() {
DemonstrateUtil.showLogResult("complete");
}
});
}
private void test2() {
//链式调用
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
e.onNext(1);
e.onNext(2);
e.onNext(3);
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
DemonstrateUtil.showLogResult("onSubscribe");
}
@Override
public void onNext(Integer integer) {
DemonstrateUtil.showLogResult("onNext-->integer" + integer);
}
@Override
public void onError(Throwable e) {
DemonstrateUtil.showLogResult("onError");
}
@Override
public void onComplete() {
DemonstrateUtil.showLogResult("onComplete");
}
});
}
private void test1() {
//创建上游,数据发射源!
//ObservableOnSubscribe对象作为参数，它的作用相当于一个计划表，当 Observable被订阅的时候，
// ObservableOnSubscribe的subscribe()方法会自动被调用，事件序列就会依照设定依次触发
//ObservableEmitter,发射器,触发事件.
Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
e.onNext(1);
e.onNext(2);
e.onNext(3);
}
});
//创建下游,数据接收处!
Observer<Integer> observer = new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
DemonstrateUtil.showLogResult("onSubscribe");
}
@Override
public void onNext(Integer integer) {
DemonstrateUtil.showLogResult("onNext--integer" + integer);
}
@Override
public void onError(Throwable e) {
DemonstrateUtil.showLogResult("onError");
}
@Override
public void onComplete() {
DemonstrateUtil.showLogResult("onComplete");
}
};
//数据源连接接收处,上游衔接下游!
//只有当上游和下游建立连接之后, 上游才会开始发送事件
observable.subscribe(observer);
}
private void initView() {
btnSend1 = (Button) findViewById(R.id.btn_send1);
btnSend1.setOnClickListener(RxJavaDemo1Activity.this);
btnSend2 = (Button) findViewById(R.id.btn_send2);
btnSend2.setOnClickListener(RxJavaDemo1Activity.this);
btnSend3 = (Button) findViewById(R.id.btn_send3);
btnSend3.setOnClickListener(RxJavaDemo1Activity.this);
btnSend4 = (Button) findViewById(R.id.btn_send4);
btnSend4.setOnClickListener(RxJavaDemo1Activity.this);
btnSend5 = (Button) findViewById(R.id.btn_send5);
btnSend5.setOnClickListener(RxJavaDemo1Activity.this);
btnSend6 = (Button) findViewById(R.id.btn_send6);
btnSend6.setOnClickListener(RxJavaDemo1Activity.this);
}
}
public class RxJavaDemo2Activity extends AppCompatActivity implements View.OnClickListener {
protected Button btn;
protected ImageView iv;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
super.setContentView(R.layout.activity_rx_java_demo2);
initView();
}
@Override
public void onClick(View view) {
if (view.getId() == R.id.btn) {
DialogUtil.showListDialog(this, "rxJava操作！", new String[]{
"0发送事件io线程并变换主线程接收",
"1子线程发送事件主线程接收",
"2默认线程发送事件默认线程接收",
}, new DialogInterface.OnClickListener() {
@Override
public void onClick(DialogInterface dialog, int which) {
switch (which) {
case 0:
show0();
break;
case 1:
show1();
break;
case 2:
show2();
break;
}
}
});
}
}
private void show2() {
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
DemonstrateUtil.showLogResult("发送的线程名称：" + Thread.currentThread().getName());
DemonstrateUtil.showLogResult("发送的线程id：" + Thread.currentThread().getId());
DemonstrateUtil.showLogResult("发送的数据:" + 1);
e.onNext(1);
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
DemonstrateUtil.showLogResult("接收的线程：" + Thread.currentThread().getName());
DemonstrateUtil.showLogResult("接收的线程id：" + Thread.currentThread().getId());
DemonstrateUtil.showLogResult("接收到的数据:-integer:" + integer);
}
});
}
private void show1() {
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
DemonstrateUtil.showLogResult("发送的线程名称：" + Thread.currentThread().getName());
DemonstrateUtil.showLogResult("发送的线程id：" + Thread.currentThread().getId());
DemonstrateUtil.showLogResult("发送的数据:" + 1);
e.onNext(1);
}
}).subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
DemonstrateUtil.showLogResult("接收的线程：" + Thread.currentThread().getName());
DemonstrateUtil.showLogResult("接收的线程id：" + Thread.currentThread().getId());
DemonstrateUtil.showLogResult("接收到的数据:-integer:" + integer);
}
});
}
private void show0() {
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
DemonstrateUtil.showLogResult("所在的线程：", Thread.currentThread().getName());
DemonstrateUtil.showLogResult("发送的数据:", 1 + "");
e.onNext(1);
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
DemonstrateUtil.showLogResult("所在的线程：", Thread.currentThread().getName());
DemonstrateUtil.showLogResult("接收到的数据:", "integer:" + integer);
}
});
}
private void initView() {
btn = (Button) findViewById(R.id.btn);
btn.setOnClickListener(RxJavaDemo2Activity.this);
iv = (ImageView) findViewById(R.id.iv);
}
}
public class RxJavaDemo3Activity extends AppCompatActivity implements View.OnClickListener {
protected Button btnBackpressure;
private Flowable mFlowable;
private Subscriber mSubscriber;
private Subscription mSubscription;
private Flowable flowableLATEST;
private Subscriber subscriberLatest;
private Subscription subscriptionLatest;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
super.setContentView(R.layout.activity_rx_java_demo3);
initView();
init4();
init6();
}
private void init6() {
flowableLATEST = Flowable.create(new FlowableOnSubscribe<Integer>() {
@Override
public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
for (int i = 1; i<=200; i++) {
emitter.onNext(i);
DemonstrateUtil.showLogResult("LATEST生产onNext:"+i);
}
}
}, BackpressureStrategy.LATEST);
//mSubscription = s;
subscriberLatest = new Subscriber<Integer>() {
@Override
public void onSubscribe(Subscription s) {
subscriptionLatest = s;
s.request(100);
}
@Override
public void onNext(Integer integer) {
DemonstrateUtil.showLogResult("Latest消费onNext:" + integer);
}
@Override
public void onError(Throwable t) {
DemonstrateUtil.showLogResult("onError");
DemonstrateUtil.showLogResult(t.getMessage());
DemonstrateUtil.showLogResult(t.toString());
}
@Override
public void onComplete() {
DemonstrateUtil.showLogResult("onComplete");
}
};
}
private void init4() {
mFlowable = Flowable.create(new FlowableOnSubscribe<Integer>() {
@Override
public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
for (int i = 1; i <= 200; i++) {
emitter.onNext(i);
DemonstrateUtil.showLogResult("生产onNext:"+i);
}
}
}, BackpressureStrategy.DROP);
//mSubscription = s;
mSubscriber = new Subscriber<Integer>() {
@Override
public void onSubscribe(Subscription s) {
mSubscription = s;
s.request(100);
}
@Override
public void onNext(Integer integer) {
DemonstrateUtil.showLogResult("消费onNext:" + integer);
}
@Override
public void onError(Throwable t) {
DemonstrateUtil.showLogResult("onError");
DemonstrateUtil.showLogResult(t.getMessage());
DemonstrateUtil.showLogResult(t.toString());
}
@Override
public void onComplete() {
DemonstrateUtil.showLogResult("onComplete");
}
};
}
@Override
public void onClick(View view) {
if (view.getId() == R.id.btn_backpressure) {
DialogUtil.showListDialog(this, "Flowable的理解使用", new String[]{
"0事件堆积现象",
"1正常使用策略ERROR!",
"2使用策略ERROR出现的异常!",
"3使用策略BUFFER,更大的缓存池",
"4使用策略DROP,事件关联100",
"5使用策略DROP,再申请100",
"6使用策略LATEST,事件关联100",
"7使用策略LATEST,再申请100",
"8使用策略MISSING",
}, new DialogInterface.OnClickListener() {
@Override
public void onClick(DialogInterface dialog, int which) {
switch (which) {
case 0:
show0();
break;
case 1:
show1();
break;
case 2:
show2();
break;
case 3:
show3();
break;
case 4:
show4();
break;
case 5:
show5();
break;
case 6:
show6();
break;
case 7:
show7();
break;
case 8:
show8();
break;
}
}
});
}
}
private void show8() {
Flowable.create(new FlowableOnSubscribe<Integer>() {
@Override
public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
for (int i = 0; i < 200; i++) {
DemonstrateUtil.showLogResult("MISSING-生成emitter" + i);
emitter.onNext(i);
}
}
}, BackpressureStrategy.MISSING).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Subscriber<Integer>() {
@Override
public void onSubscribe(Subscription s) {
//mSubscription = s;
//s.request(0);
}
@Override
public void onNext(Integer integer) {
DemonstrateUtil.showLogResult("MISSING-消费onNext" + integer);
}
@Override
public void onError(Throwable t) {
DemonstrateUtil.showLogResult("onError" + t.getMessage());
DemonstrateUtil.showLogResult("onError" + t.toString());
t.printStackTrace();
}
@Override
public void onComplete() {
DemonstrateUtil.showLogResult("onComplete");
}
});
}
private void show7() {
subscriptionLatest.request(100);
}
private void show6() {
flowableLATEST.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(subscriberLatest);
}
private void show5() {
//128-100-100= -72.
mSubscription.request(100);
}
private void show4() {
mFlowable.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(mSubscriber);
}
private void show3() {
Flowable.create(new FlowableOnSubscribe<Integer>() {
@Override
public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
for (int i = 0; i < 200; i++) {
DemonstrateUtil.showLogResult("emitter" + i);
emitter.onNext(i);
}
}
}, BackpressureStrategy.BUFFER).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Subscriber<Integer>() {
@Override
public void onSubscribe(Subscription s) {
//mSubscription = s;
//s.request(0);
}
@Override
public void onNext(Integer integer) {
DemonstrateUtil.showLogResult("onNext" + integer);
}
@Override
public void onError(Throwable t) {
DemonstrateUtil.showLogResult("onError" + t.getMessage());
DemonstrateUtil.showLogResult("onError" + t.toString());
t.printStackTrace();
}
@Override
public void onComplete() {
DemonstrateUtil.showLogResult("onComplete");
}
});
}
private void show2() {
Flowable.create(new FlowableOnSubscribe<Integer>() {
@Override
public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
for (int i = 0; i < 200; i++) {
DemonstrateUtil.showLogResult("emitter" + i);
emitter.onNext(i);
}
}
}, BackpressureStrategy.ERROR).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Subscriber<Integer>() {
@Override
public void onSubscribe(Subscription s) {
//mSubscription = s;
//s.request(0);
}
@Override
public void onNext(Integer integer) {
DemonstrateUtil.showLogResult("onNext" + integer);
}
@Override
public void onError(Throwable t) {
DemonstrateUtil.showLogResult("onError" + t.getMessage());
DemonstrateUtil.showLogResult("onError" + t.toString());
t.printStackTrace();
}
@Override
public void onComplete() {
DemonstrateUtil.showLogResult("onComplete");
}
});
}
private void show1() {
Flowable.create(new FlowableOnSubscribe<Integer>() {
@Override
public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
for (int i = 0; i < 127; i++) {//128--- 0--->126
DemonstrateUtil.showLogResult("emitter " + i);
emitter.onNext(i);
}
DemonstrateUtil.showLogResult("emitter complete");
emitter.onComplete();
}
}, BackpressureStrategy.ERROR) //增加了一个参数,设置处理策略.
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Subscriber<Integer>() {
@Override
public void onSubscribe(Subscription s) {
DemonstrateUtil.showLogResult("onSubscribe");
//用来向生产者申请可以消费的事件数量,这样我们便可以根据本身的消费能力进行消费事件.
s.request(Long.MAX_VALUE);
}
@Override
public void onNext(Integer integer) {
DemonstrateUtil.showLogResult("onNext: " + integer);
}
@Override
public void onError(Throwable t) {
DemonstrateUtil.showLogResult("onError: " + t.getMessage());
DemonstrateUtil.showLogResult("onError: " + t.toString());
t.printStackTrace();
}
@Override
public void onComplete() {
DemonstrateUtil.showLogResult("onComplete: ");
}
});
}
private void show0() {
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
while (true) {
for (int i = 0; i < 129; i++) {
e.onNext(1);
}
}
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Thread.sleep(5000);
DemonstrateUtil.showLogResult("接受到" + integer);
}
});
}
private void initView() {
btnBackpressure = (Button) findViewById(R.id.btn_backpressure);
btnBackpressure.setOnClickListener(RxJavaDemo3Activity.this);
}
}