Rxjava2.X的一些讲解

rxjava是什么？

总结一句话就是异步切换操作的工具

Rxjava常用的一些操作符

1.concat（英文翻译为连接的意思）

就是将多个Observable（上游发送器又名被观察者）统一起来按顺序返回一个Observable

下面是源码的一重要部分，返回的是一个T类型，所以两个被观察者的返回的类型应该是一样的，或者可以传入一个Object

如下图所示，创建两个Observable

然后用concat连接起来

如果Text（12）这样打印的结构会是什么呢？

没错只走了第一个Observable,因为他只有等第一个执行完onnext()并且执行onComplete()方法后第二个才会执行

我们验证一些

看下打印结果

并且observer(下游又名观察者)onComplete（）方法是又最后一个Observable所调用的

2.map()操作符，用来转换返回结果值的

传入一个function(传入的值，需要转换为的值)

Observable.create(new ObservableOnSubscribe<String>() {@Override public void subscribe(ObservableEmitter<String> e) throws Exception {e.onNext("234");}
}).map(new Function<String, Integer>() {@Override public Integer apply(String s) throws Exception {if (s.equals("234")){return 3;}else{return 5;}}
}).subscribe(new Observer<Integer>() {@Override public void onSubscribe(Disposable d) {System.out.println("Disposable == " + d.isDisposed());}@Override public void onNext(Integer integer) {System.out.println("onNext == " + integer);}@Override public void onError(Throwable e) {}@Override public void onComplete() {System.out.println("onComplete == ");}
});

我们看下打印的结果是什么

没错转换了，这就是map转换器的作用

大家可能看到

这是是干嘛的呢？

取消订阅的功能

3.ZIP操作符

顾明思议是打包的意思，就是相当于把两个observable ,随意取出数据来组合

组合的过程是分别从两根水管里各取出一个事件来进行组合, 并且一个事件只能被使用一次, 组合的顺序是严格按照事件发送的顺利来进行的, 也就是说不会出现圆形1 事件和三角形B 事件进行合并, 也不可能出现圆形2 和三角形A 进行合并的情况.
最终下游收到的事件数量是和上游中发送事件最少的那一根水管的事件数量相同. 这个也很好理解, 因为是从每一根水管里取一个事件来进行合并, 最少的那个肯定就最先取完 , 这个时候其他的水管尽管还有事件 , 但是已经没有足够的事件来组合了, 因此下游就不会收到剩余的事件了.

实用的地方

zip在Android中的使用，可以适用于如下场景，一个界面需要展示用户的一些信息，这些信息分别要从两个服务器接口中获取，只有当两个数据都获取后才能进行展示。这类同时的信息请求比较适用zip

这里注意BIGFunction是三个参数，前面两个是用来组合的observable的参数，最后一个是用来返回的参数

4schedulePeriodically

轮询请求，在一些需要隔段时间进行联网请求反复需要获取数据的场景

 Observable.create(new ObservableOnSubscribe<String>() {@Overridepublic void subscribe(final ObservableEmitter<String> e) throws Exception {Schedulers.io().createWorker().schedulePeriodically(new Runnable() {  //指定请求网络的在io线程上@Overridepublic void run() {e.onNext("我来请求数据了");}},0,500,TimeUnit.MILLISECONDS);}
}).subscribe(new Observer<String>() {@Overridepublic void onSubscribe(Disposable d) {}@Overridepublic void onNext(String s) {Log.e("tag",s);}@Overridepublic void onError(Throwable e) {}@Overridepublic void onComplete() {}
});

结果是每隔500毫米请请求一次一直打印log