RxJava从今以后你就会了

作者: JC_Mobile | 来源:发表于2016-09-22 09:31 被阅读2064次

RxJava技术分享

京金所—时光

2016.9.22

这里我拿出来给 Android 开发者的 RxJava 详解中的例子

假设有这样一个需求：界面上有一个自定义的视图 imageCollectorView ，它的作用是显示多张图片，并能使用 addImage(Bitmap) 方法来任意增加显示的图片。现在需要程序将一个给出的目录数组 File[] folders 中每个目录下的 png 图片都加载出来并显示在 imageCollectorView 中。需要注意的是，由于读取图片的这一过程较为耗时，需要放在后台执行，而图片的显示则必须在 UI 线程执行。常用的实现方式有多种，我这里贴出其中一种：

//开启一条子线程
new Thread() {
@Override
public void run() {
    super.run();
    //遍历给出的目录数组，获取每一个目录里面的文件的数组
    for (File folder : folders) {
        File[] files = folder.listFiles();
        //遍历这个文件数组，筛选出来png图片
        for (File file : files) {
            if (file.getName().endsWith(".png")) {
                //将路径转换成Bitmap图片
                final Bitmap bitmap = getBitmapFromFile(file);
                //切换到主线程更新UI。
                getActivity().runOnUiThread(new Runnable() {
                    @Override
                    public void run() {
                        imageCollectorView.addImage(bitmap);
                    }
                });
            }
        }
    }
}

}.start();

而如果使用 RxJava ，实现方式是这样的：

`
//将文件夹目录创建成一个可观测的序列

Observable.from(folders)

//遍历目录数组中的每一个文件对象，并将这每一个文件对象转换成可观测序列
.flatMap(new Func1<File, Observable<File>>() {
    @Override
    public Observable<File> call(File file) {
        return Observable.from(file.listFiles());
    }
})
//对这每一个文件对象进行过滤，选出png图片
.filter(new Func1<File, Boolean>() {
    @Override
    public Boolean call(File file) {
        return file.getName().endsWith(".png");
    }
})
//将这每一个png文件都转成Bitmap传递
.map(new Func1<File, Bitmap>() {
    @Override
    public Bitmap call(File file) {
        return getBitmapFromFile(file);
    }
})
//耗时操作运行在子线程
.subscribeOn(Schedulers.io())
//修改UI的操作放在主线程
.observeOn(AndroidSchedulers.mainThread())
//订阅，展示
.subscribe(new Action1<Bitmap>() {
    @Override
    public void call(Bitmap bitmap) {
        imageCollectorView.addImage(bitmap);
    }
});

今天主要从以下几个方面来着重介绍RxJava

RxJava的重要概念
RxJava的操作符
RxJava的应用场景

一：RxJava的重要概念

1.1 什么是RxJava？

RxJava 在 GitHub 主页上的自我介绍是 "a library for composing asynchronous and event-based programs using observable sequences for the Java VM"（一个在 Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库）。

到底什么是RxJava，一个词，就是异步。两个词就是异步+链式操作。

1.2 RxJava难吗？

RxJava其实一点也不难。RxJava能做到的东西，别人也一样可以做到。但是，如果使用RxJava，那么可以比他们逻辑更简洁。有人说Rx太好用了，也有人说RxJava太难用了。好用是因为RxJava使用操作符（函数式编程）极大的简化了我们代码编写的逻辑，难用可能就是对操作符用的还不习惯，对命令式编程的思想有些固话。

1.3 但是很多人说学习RxJava起点高啊？

高在哪里。我认为是几个重要的思想。在我刚学习Java语言的时候，我认识了面向对象的编程思想。简单来说就是指挥对象帮我们做事。通过一个点.即可。同样，我认为学习RxJava也是学习一个编程思想。就是函数式编程，和响应式编程。我不知道这算不算是编程思想，但我知道一旦明白这两个东东，对我们学习Rx有着意想不到的好处。

今天，今天通过介绍 Rx里面的概念(这部分我觉得才是最重要的一点)，Rx里面的操作符，Rx里面的常用场景，还有Rx的扩展，将读者带入Rx的大门，了解他，触摸它，使用它，掌握它。

1.4 响应式编程：

什么是响应式编程(Reactive programming)？ RxJava ==>ReactiveX Java

百度百科的定义：

响应式编程是一种==面向数据流和变化传播==的==编程范式==。这意味着可以在编程语言中很方便地表达静态或动态的数据流，而相关的计算模型会自动将变化的值通过数据流进行传播。

例如，在命令式编程环境中(面向过程和面向对象的语言中)，a=b+c表示将表达式的结果赋给a，而之后改变b或c的值不会影响a。但在响应式编程中，a的值会随着b或c的更新而更新。响应式编程最初是为了简化交互式用户界面的创建和实时系统动画的绘制而提出来的一种方法，但它本质上是一种通用的编程范式。

这里什么是编程范式呢？说白了就是一种编码的风格和模式。

此时，我们不免会疑惑，数据我们知道，那么数据流呢？

在RxJava Essentials一书中，是这么说的。

响应式编程是一种基于异步数据流概念的编程模式。数据流就像一条河：它可以被观测，被过滤，被操作，或者为新的消费者与另外一条流合并为一条新的流。

响应式编程的一个关键概念是事件。事件可以被等待，可以触发过程，也可以触发其它事件。

我相信看到这里大部分人已经晕了。
不过我是越来越清晰，一字一句的多读几遍。

在那些年我们错过的响应式编程中是这么介绍的：

响应式编程就是与异步数据流交互的编程方式。（和上面基本一样）

一方面，这已经不是什么新事物了。事件总线(Event Buses)或一些典型的点击事件本质上就是一个异步事件流(asynchronous event stream)，这样你就可以观察它的变化并使其做出一些反应(do some side effects)。响应式是这样的一个思路：除了点击和悬停(hover)的事件外，你可以给任何事物创建数据流。数据流无处不在，任何东西都可以成为一个数据流，例如变量、用户输入、属性、缓存、数据结构等等。举个栗子，你可以把你的微博订阅功能想象成跟点击事件一样的数据流，你可以监听这样的数据流，并做出相应的反应。

最重要的是，你会拥有一些令人惊艳的函数去结合、创建和过滤任何一组数据流。这就是”函数式编程”的魔力所在。一个数据流可以作为另一个数据流的输入，甚至多个数据流也可以作为另一个数据流的输入。你可以合并两个数据流，也可以过滤一个数据流得到另一个只包含你感兴趣的事件的数据流，还可以映射一个数据流的值到一个新的数据流里。

最后我说一下我的理解:

首先响应式编程仅仅是一种编程的模式，风格，并不具有特殊的含义和操作，响应式编程他是用来对数据流进行操作的，所有对数据流进行异步操作，观测等行为，我们都可以管这种操作行为叫做响应式编程。那么什么又是数据流呢？

1.4 数据流的概念：

数据流是整个响应式编程体系中的核心，要想学习响应式编程，首先我们就需要搞明白什么是数据流。其实，一个数据流是一个按时间排序的即将发生的事件(Ongoing events ordered in time)的序列。就比如我们警察抓小偷，小偷偷东西的时候，警察抓住他了吗，小偷偷东西之前做了什么，之后做了什么，警察抓他之前做了什么，抓住他做了什么或者没抓住又该怎么去做。这一系列事件组成的序列就是数据流。数据流就像一条河：它可以被观测，被过滤，被操作，或者为新的消费者与另外一条流合并为一条新的流。

也就是说我们的响应式编程，就是对这一系列异步事件进行处理的一种编程模式。其实，你会发现，看了一大圈，又回到了第一句话，但是我相信，你现在应该不会在迷糊了。

1.5 函数式编程:

其实我一直搞不明白函数式编程和响应式编程。但是现在我明白了，而且他俩根本就不一样。

1.5.1 什么是函数式编程:

应该有至少三种编程思想：

命令式编程 Imperative programming
逻辑式编程 Object-oriented Programming
函数式编程 Functional Programming

命令式编程关心解决问题的步骤，面向对象编程是也是一种命令式编程，面向过程的C语言必然也是了。
而函数式编程关心数据的映射，即一种东西和另一种东西之间的对应关系。它的主要思想是把运算过程尽量写成一系列嵌套的函数调用。

比如这种运算：(1 + 2) * 3 - 4 ;
之前我们写是这样：

```
int a = 1 + 2; int b = a * 3; int result = b - 4;
System.out.println(b);
```

然而用函数式编程思想写就是这样：

```
System.out.print(subtract( multiply(add(1,2)，3),4) );
```

全部变为了函数调用，这样看起来也简洁、见名之意。(示例来自阮一峰博客)

看到这，想一下，使用RxJava时是不是全部调用各种操作符进行处理，这就是对事件流进行运算啊，全部调用函数进行处理。

在这里我们来对比一下函数式编程和响应式编程。函数式编程是对数据关系进行==映射==，比如说y=ax;这个一元函数。函数式编程我们只需要定义处来这个映射关系即参数a的算法，我们就可以得到y。而响应式编程的重点则不在于这个参数a，而是当我的x发生改变的时候（就是我们注册的x的状态改变）的时候，我要做出一个什么样反应。这和我大学学到的激励和响应是差不多的。我给你一个什么样的激励，你能反馈我一个什么样的响应。

1.6 观察者模式

在今天，观察者模式是出现的最常用的软件设计模式之一。它基于subject这个概念。subject是一种特殊对象，当它改变时，那些由它保存的一系列对象将会得到通知。而这一系列对象被称作Observers(观察者),它们会对外暴漏了一个通知方法,当subject状态发生变化时会调用的这个方法。

就比如说，当A的某状态发生改变或者接收到某种激励的时候，B需要对这种激励做出一个反馈，那么我们只需要让A持有B的引用，同时去调用B的方法即可（我们最终的目的是将发生改变的状态传递出去，通过方法传递，我们的B需要这个状态）。但是通常，我们的A不知道自己什么时候能获取到这个激励（什么时候状态会发生改变），这个时候我们的B呢就可以通过注册，来处理这个响应。同时，我们也可以不用拿到B的引用，我们可以通过一个中间接口，将发生改变的状态（或者是B需要的状态）通过接口传递过去，同时B只需要拿到这个接口的实例对象即可，这就是B的注册过程。也是最简单的回调。而回调也是观察者模式的一种简化形式，1-1的关系，当然我们平时不仅setOnClickListener()也addOnXxxListener()，这种一对多的关系就是观察者模式。

1.7 扩展的观察者模式

与传统观察者模式不同， RxJava的事件回调方法除了普通事件 onNext() （相当于 onClick() / onEvent()）之外，还定义了两个特殊的事件：onCompleted() 和 onError()。

1.8 几个常见的类

在RxJava的世界里，我们有四种角色：

Observable

Observer

Subscriber

Subjects

Observables和Subjects是两个“生产”实体，Observers和Subscribers是两个“消费”实体

```
public class Observable<T> {
    final OnSubscribe<T> onSubscribe;
    }
public interface Observer<T>{
    onComplete();onError;onNext();
    }
public abstract class Subscriber<T> implements Observer<T>, Subscription 
public abstract class Subject<T, R> extends Observable<R> implements Observer<T> 
*/
```

总览：

Creating Observables
Operators that originate new Observables.

Create — create an Observable ++from scratch（从头）++ by calling observer methods ++programmatically（以编程的方式）++
++Defer（延迟创建）++ — do not create the Observable until the observer subscribes, and create a fresh Observable for each observer
Empty/Never/Throw — create Observables that have very ++precise（精确的）++ and limited behavior
From — ++convert（转换）++ some other object or data structure into an Observable
++Interval（间隔）++ — create an Observable that ++emits（发射）++ a sequence of integers spaced by a particular time interval
Just — convert an object or a set of objects into an Observable that emits that or those objects
Range — create an Observable that emits a range of sequential integers
Repeat — create an Observable that emits a particular item or sequence of items repeatedly
Start — create an Observable that emits the return value of a function
Timer — create an Observable that emits a single item after a given delay

一：Create的使用

可以通过create操作去创建一个被观察者Observable。Create是最基本的创建Observable的操作符。

image

创建一个Observable（被观察者）最重要的就是要和合适的时机调用Subscriber（观察者）的onNext/onComplete/onError方法。onNext就是发射处理好的数据给Subscriber; onComplete用来告诉Subscriber所有的数据都已发射完毕；onError是在发生错误的时候发射一个Throwable对象给Subscriber。需要注意的一点就是Observable必须调用所有的Subscriber的onComplete方法并且只能调用一次，出错的时候调用onError方法也是一样的，并且一旦调用后就不能调用Subscriber的任何其他方法了。

介绍完毕。

我再详细介绍一下。

我们可以看到Observable.create(这里经常会传入一个OnSubscribe接口);

OnSubscribe接口又是什么呢？它继承自Action1<Subscriber<? super T>这个接口，他的call（<Subscriber<? super T> subscriber）方法里只有一个参数，它传入了一个观察者！！！也就是说我们在使用create的时候，我们将观察者包裹在了被观察者内部。并且在被观察这订阅观察者的时候，我们通过一个hook对象将观察者Subscriber的对象传入到OnSubscribe.call(传到这里面)。这样就实现了由被观察者在订阅的时候调用观察者的方法。并且在事件完成的时候自动取消订阅。

    Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() {
        @Override
        public void call(Subscriber<? super String> subscriber) {
            //将依次发射3次onNext ，之后发射onComplete(发射完这个将不会再有输出，同时自动取消注册)
            subscriber.onNext("Hello第一次发射");
            subscriber.onNext("World第二次发射");
            subscriber.onNext("Over 第三次发射");
            subscriber.onCompleted();
            subscriber.onError(new Throwable());
        }
    })

2.Just、From的使用

为了简化创建操作，我们平时最常使用的一般是这两个操作符。just(T...):from(T[])。

Just操作符将某个对象转化为Observable对象，并且将其发射出去，可以使一个数字、一个字符串、数组、Iterate对象等。其实当just传入一个参数的时候，就是调用create()方法，传入多参数的时候，就是调用from()方法。从而做到将参数依次发射出去

From操作符用来将某个对象转化为Observable对象，并且依次将其内容发射出去。这个类似于just，但是just会将这个对象整个发射出去。比如说一个含有10个数字的数组，使用from就会发射10次，每次发射一个数字，而使用just会发射一次来将整个的数组发射出去。

image

3.Defer

Defer操作符只有当有Subscriber来订阅的时候才会创建一个新的Observable对象,也就是说每次订阅都会得到一个刚创建的最新的Observable对象，这可以确保Observable对象里的数据是最新的

image

4.Interval

Interval所创建的Observable对象会从0开始，每隔固定的时间发射一个数字。需要注意的是这个对象是运行在computation Scheduler,基本上和时间有关系的都是运行在computation Scheduler 中,所以如果需要在view中显示结果，要在主线程中订阅。

image

5.Range

这个应该好理解一点，看图依次发射区间（n,n+m-1）里面的数字，左闭右开

image

6.Repeat、Timer

Repeat会将一个Observable对象重复发射，我们可以指定其发射的次数

image

Timer会在指定时间后发射一个数字0，注意其也是运行在computation Scheduler

image

总览：

Transforming Observables
Operators that transform items that are emitted by an Observable.

Buffer — ++periodically（定期的）++ gather items from an Observable into bundles and emit these bundles rather than emitting the items ++one at a time（一次一个）++
FlatMap — transform the items emitted by an Observable into Observables, then flatten the emissions from those into a single Observable
GroupBy — divide an Observable into a set of Observables that each emit a different group of items from the original Observable, organized by key
Map— transform the items emitted by an Observable by applying a function to each item
Scan — apply a function to each item emitted by an Observable, sequentially, and emit each successive value
Window — periodically subdivide items from an Observable into Observable windows and emit these windows rather than emitting the items one at a time

RxJava 提供了对事件序列进行变换的支持，这是它的核心功能之一，也是大多数人说『RxJava 真是太好用了』的最大原因。所谓变换，就是将事件序列中的对象或整个序列进行加工处理，转换成不同的事件或事件序列。

1.Buffer

Buffer操作符所要做的事情就是将数据按照规定的大小、或者时间做一下缓存，然后将缓存的数据作为一个集合发射出去。

image
在上图中，按照规定的大小（count = 3）来收集数据，之后将他们整合成一个集合发射出去。

image
在上图中，加入了一个skip参数用来指定每次发射一个集合需要跳过几个数据，图中指定（count = 2,skip = 3)，就会每3个数据发射一个包含两个数据的集合，如果count==skip的话，我们就会发现其等效于第一种情况了。这里还是很好理解的。

2.flatMap & concatMap

Flatmap是一个用处特别多的操作符。它可以将数据根据你想要的规则进行转化后再发射出去。其原理就是将这个Observable转化为多个以原Observable发射的数据作为源数据的Observable，然后再将这多个Observable发射的数据整合发射出来，需要注意的是最后的顺序可能会交错地发射出来，如果对顺序有严格的要求的话可以使用concatmap操作符。

image

如上图，我们按照一个圆对应两个菱形的规则，将一个Observable<圆>转换成两个Observable<菱形>。并将两个Observable<菱形>发射出去。

3.Map

Map操作符的功能类似于FlatMap，不同之处在于它对数据的转化是直接进行的，而FlatMap需要通过一些中间的Observables来进行。

image

4.GroupBy

GroupBy操作符将原始Observable发射的数据按照key来拆分成一些小的Observable，然后这些小的Observable分别发射其所包含的的数据，类似于sql里面的groupBy。在使用中，我们需要提供一个生成key的规则，所有key相同的数据会包含在同一个小的Observable种。

image

5.Scan

Scan操作符对一个序列的数据应用一个函数，并将这个函数的结果发射出去作为下个数据应用这个函数时候的第一个参数使用，有点类似于递归操作

image

总览：过滤

Filtering Observables
Operators that selectively emit items from a source Observable.

Debounce — only emit an item from an Observable if a particular timespan has passed without it emitting another item
Distinct — suppress duplicate items emitted by an Observable
ElementAt — emit only item n emitted by an Observable
Filter — emit only those items from an Observable that pass a predicate test
First — emit only the first item, or the first item that meets a condition, from an Observable
IgnoreElements — do not emit any items from an Observable but mirror its termination notification
Last — emit only the last item emitted by an Observable
Sample — emit the most recent item emitted by an Observable within periodic time intervals
Skip — suppress the first n items emitted by an Observable
SkipLast — suppress the last n items emitted by an Observable
Take — emit only the first n items emitted by an Observable
TakeLast — emit only the last n items emitted by an Observable

1.Filter

Filter只会返回满足过滤条件的数据.比如说，我们想让数据{a1,a2,b1,v1,c1}中以v开头的数据排除掉，或者我们经常用到的AppInfo中数据为null的过滤掉，我们可以使用filter轻松实现。

image
如上图，我们应该filter(东西 == 圆)

2.take/takelast

当我们不需要整个序列时，而是只想取开头或结尾的几个元素，我们可以用take()或takeLast()。

image

```
//太简单了，我就不写demo了，take（3）从前面取3个，takeLast(3)从后面取3个
 Observable.from(apps).take(3)
```

3.Distinct 有且仅有一次

我们可以对我们的序列使用distinct()函数去掉重复的。就像takeLast()一样，distinct()作用于一个完整的序列，然后得到重复的过滤项，它需要记录每一个发射的值。如果你在处理一大堆序列或者大的数据记得关注内存使用情况。
Distinct操作符的用处就是用来去重，非常好理解。如下图所示，所有重复的数据都会被过滤掉。还有一个操作符distinctUntilChanged,是用来过滤掉连续的重复数据。

image

4.First and last

first()方法和last()方法很容易弄明白。它们从Observable中只发射第一个元素或者最后一个元素。连图我都不上了。
firstOrDefault()和lastOrDefault().这两个函数当可观测序列完成时不再发射任何值时用得上。在这种场景下，如果Observable不再发射任何值时我们可以指定发射一个默认的值

5.Skip and SkipLast

skip()和skipLast()函数与take()和takeLast()相对应。它们用整数N作参数，从本质上来说，它们不让Observable发射前N个或者后N个值。如果我们知道一个序列以没有太多用的“可控”元素开头或结尾时我们可以使用它。

image

Combining Observables

1.Zip

Zip操作符将多个Observable发射的数据按顺序组合起来，每个数据只能组合一次，而且都是有序的。最终组合的数据的数量由发射数据最少的Observable来决定。

image

2.Merege

Merge操作符将多个Observable发射的数据整合起来发射，就如同是一个Observable发射的数据一样。但是其发射的数据有可能是交错的，如果想要没有交错，可以使用concat操作符。当某一个Observable发出onError的时候，merge的过程会被停止并将错误分发给Subscriber，如果不想让错误终止merge的过程，可以使用MeregeDelayError操作符，会将错误在merge结束后再分发。

image

1.使用Scheduler进行线程的切换

在不指定线程的情况下， RxJava 遵循的是线程不变的原则，即：在哪个线程调用订阅方法 subscribe()，就在哪个线程生产事件；在哪个线程生产事件，就在哪个线程消费事件。如果需要切换线程，就需要用到 Scheduler （调度器）。

几个常用的Api:

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。
Android 还有一个专用的 AndroidSchedulers.mainThread()，它指定的操作将在 Android 主线程运行。

有了这几个 Scheduler ，就可以使用 subscribeOn() 和 observeOn() 两个方法来对线程进行控制了。

subscribeOn(): 指定 subscribe() 所发生的线程，即 Observable.OnSubscribe 被激活时所处的线程。或者叫做事件产生的线程。
observeOn(): 指定 Subscriber 所运行在的线程。或者叫做事件消费的线程。

精简代码：（教简单就不放全部代码了）

```
 mSubscription = getObservable()//创建一个被观察者
   .subscribeOn(Schedulers.io())//指定订阅事件发生的线程
   .doOnSubscribe(new Action0() {//事件发生之前做什么事
//onStart() 也可以用作流程开始前的初始化。但是 onStart() 由于在 subscribe() 发生时就被调用了，因此不能指定线程，而是只能执行在 subscribe() 被调用时的线程。
//doOnSubscribe同样是在 subscribe() 调用后而且在事件发送前执行，但区别在于它可以指定线程。
//默认情况下， doOnSubscribe() 执行在 subscribe() 发生的线程；而如果在 doOnSubscribe() 之后有 subscribeOn() 的话，它将执行在离它最近的 subscribeOn() 所指定的线程。
                            @Override
                            public void call() {
                                mProgressBar.setVisibility(View.VISIBLE);
                                _log("按钮被点击了");
                            }
                        })
                        .observeOn(AndroidSchedulers.mainThread())//指定事件消费在那个线程里面
                        .subscribe(getSubscriber());//订阅观察者
```

### 2.使用Buffer收集数据流后在发射

Buffer顾名思义，就相当一个缓冲区，我们可以指定一个时间或者数量来收集这些数据，满足了这个事件或者数量之后，在统一发射出去。在这里我们举个例子，来学习Buffer,同时更深入的理解一个更重要的操作符map

//观察在两秒的时间内，按钮被点击的次数
 public Subscription getClickNumInTime(){
 //这里需要引入compile 'com.jakewharton.rxbinding:rxbinding:0.4.0'依赖，这样我们可以观测操作这些控件的数据流
    return RxView
    //将点击事件包装成数据（创建一个可观测的序列）流发射出去
           .clicks(tapButton)
           //这里我们可以很清楚，很明了的看到，我们将点击事件map成了Integer类型的数据
           .map(new Func1<Void, Integer>() {
               @Override
               public Integer call(Void aVoid) {
                   _log("得到了一个点击");
                   return 1;
               }
           })
           //通过Buffer指定2秒钟发射一次收集的数据
            .buffer(2, TimeUnit.SECONDS)
            //Buffer指定的线程是computation线程，我们修改UI要改变线程
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Subscriber<List<Integer>>() {
                @Override
                public void onCompleted() {
                    _log("你永远不会走到这里，我不信这句话能打印处来");
                }

                @Override
                public void onError(Throwable e) {
                    _log("出错了"+e.getMessage());
                }
//一旦订阅之后，除非取消订阅，或者彻底杀死进程，否则会一直走下去，造成内存泄露
                @Override
                public void onNext(List<Integer> integers) {
                    if(integers.size() > 0)
                    _log("你在两秒内一共点击了："+integers.size()+"次");
                }
            });
}


### 3.使用Debounce做TextSearch
比如说当我们输入的内容发生改变的时候，我们需要实时的去异步请求网络/数据库来提示用户某样东西也需要改变。就比如，当我输入金额的时候，没当我EditText内容发生改变的时候我去请求服务器，将得到的京金币展示在桌面上。有些时候，用户连续几个数字输入的过快，没必要去请求网络，这个时候，我们可以使用Debounce来做限流操作，一定时间内，我们只要最后一次得到的数据。

    ```
    mSubscription = RxTextView
    //将EditText的内容改变事件转换成一组可观测的序列（数据流）
                .textChangeEvents(inputTxtDebounce)
                //开启限流操作，在400毫秒内，我们只要最后一次改变的结果
                .debounce(400, TimeUnit.MILLISECONDS)
                //过滤掉空字符串，只有满足条件的能留下
                .filter(new Func1<TextViewTextChangeEvent, Boolean>() {
                    @Override
                    public Boolean call(TextViewTextChangeEvent textViewTextChangeEvent) {
                        return !inputTxtDebounce.getText().toString().trim().isEmpty();
                    }
                })
                //和Buffer一样，生产事件运行在computation线程
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(getSubscriber());
    ```
很多时候我们都需要给订阅事件来一个返回值，来方便我们在离开页面的时候取消订阅。

### 4.Retrofit+RxJava
目前最火的网络请求就是Retrofit了，而目前最火的框架，二者也有一合之地，而二者的无缝结合，完全使我们的网络请求随心所欲。
Retrofit我就不多做介绍了。

    ```
    //先来看一个最简单的。
    //首先我们用一个装订阅set集合，将这一个网络请求的订阅装进来
    mSubscriptions.add(
    //通过Retrofit创建网络请求
    mGithubApi
        .contributors(_username.getText().toString(),_repo.getText().toString())
        //指定网络请求（生产事件）运行的线程
        .subscribeOn(Schedulers.io())
        //指定响应的线程（消费事件）
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Subscriber<List<Contributor>>() {
             @Override
             public void onCompleted() {
                 _adapter.add("完毕");
              }

              @Override
              public void onError(Throwable e) {
                _adapter.add("加载失败"+e.getMessage());
              }

            @Override
             public void onNext(List<Contributor> contributors) {
             //网络请求得到了一组集合，我们对这一组集合遍历拿到我们想要的数据
                for (int i = 0; i < contributors.size(); i++) {
                    _adapter.add(format("%s 为 %s 库做出了 %d 个贡献",contributors.get(i).login,_repo.getText().toString(),contributors.get(i).contributions));
            }
        }
      })
    );
    ```
如果！我在拿到List<Contributor>响应数据的时候，我要根据里面的作者，去搜索每一个作者的详细信息，怎么办？我想一想就觉得头疼，先把这些作者收集起来，遍历，再去挨个请求，再将得到的数据和刚才的数据整合起来，显示到桌面？
不，RxJava的优点就是，你的需求越复杂，我的逻辑越简单！！！！！！

    ```
    mSubscriptions.add(
    //先创建网络请求
    mGithubApi.contributors(_username.getText().toString(),_repo.getText().toString())
    //我们先将得到的数据<List<Contributor>，遍历发射并观测其中的每一个元素
    .flatMap(new Func1<List<Contributor>, Observable<Contributor>>() {
        @Override
        public Observable<Contributor> call(List<Contributor> contributors) {
            return Observable.from(contributors);
        }
    })
    //这里，重点来了，先说一下Pair，他就是一个包含了两个对象的容器。在这里我们用它来装User和Contributor
    .flatMap(new Func1<Contributor, Observable<Pair<User,Contributor>>>() {
        @Override
        public Observable<Pair<User,Contributor>> call(Contributor contributor) {
        //在这里，我们在根据longin参数再去请求user的详细信息
        Observable<User> userObservable = mGithubApi.user(contributor.login)
            //过滤掉非空数据
            .filter(new Func1<User, Boolean>() {
                @Override
                public Boolean call(User user) {
                    return !isEmpty(user.name) && !isEmpty(user.email);
                }
            });
            //到了整合的时候了。我们将user的详细信息的数据流，同时将contributor这个值再次包装成数据流发整合之后
            //怎么整合呢，这时候就要看Func2这个函数了
            return Observable.zip(userObservable, Observable.just(contributor), new Func2<User, Contributor, Pair<User, Contributor>>() {
                @Override
                public Pair<User, Contributor> call(User user, Contributor contributor) {
                //它将我们的user用户信息和contributor整合在Pair容器中返回回去
                     return new Pair<User, Contributor>(user,contributor);
                }
            });
        }
    })
    //下面的不用多说了，我只想说，这个代码写的真的很精彩
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Subscriber<Pair<User, Contributor>>() {
                            @Override
                            public void onCompleted() {
                                _adapter.add("完毕");
                            }

                            @Override
                            public void onError(Throwable e) {
                                _adapter.add("加载失败");
                            }

                            @Override
                            public void onNext(Pair<User, Contributor> pair) {
                                User user = pair.first;
                                Contributor contributor = pair.second;

                                _adapter.add(format("%s(%s) 为 %s 库做出了 %d 个贡献",
                                        user.name,
                                        user.email,
                                        _repo.getText().toString()
                                        ,contributor.contributions));

                                _adapter.notifyDataSetChanged();
                            }
                        })
                );

5.使用combineLatest做登录注册模块

CombineLatest操作符可以将2~9个Observable发射的数据组装起来然后再发射出来。不过还有两个前提：
1.所有的Observable都发射过数据。（用户操作APP可以感觉到，只有当我三个EditText全部输入过内容，才会触发combineLatest的发射事件）
2.满足条件1的时候任何一个Observable发射一个数据，就将所有Observable最新发射的数据按照提供的函数组装起来发射出去。(每当我任何一个EditText发生改变的时候都会重新走一遍combineLatest的发射事件)
这里我们看着例子，来解释这些话语

image

这个也是挺常用的，他可以将我们需要的数据流整合在一起。比如：
注册的时候所有输入信息（邮箱、密码、电话号码等）合法才点亮注册按钮。

```
//这里我们分别将3个EditText的内容变化的事件转换成数据流
_emailChangeObservable = RxTextView.textChanges(_email).skip(1);
_passwordChangeObservable = RxTextView.textChanges(_password).skip(1);
_numberChangeObservable = RxTextView.textChanges(_number).skip(1);

private void _combineLatestEvents() {
//我们将三个数据流整合起来，并加入整合的规则（这里可以将规则看成返回值Boolean，满足规则我返回True，否则返回false）
    _subscription = Observable.combineLatest(_emailChangeObservable,
            _passwordChangeObservable, _numberChangeObservable, new Func3<CharSequence, CharSequence, CharSequence, Boolean>() {
                @Override
                public Boolean call(CharSequence email, CharSequence psw, CharSequence num) {
                    boolean emailB = "123456".equals(email.toString());
                    if(!emailB){
                        _email.setError("必须是123456");
                    }
                    boolean pswB = "654321".equals(psw.toString());
                    if(!pswB){
                        _password.setError("必须是654321");
                    }
                    boolean numB = "521".equals(num.toString());
                    if(!numB){
                        _number.setError("必须是521");
                    }
                    return emailB && pswB && numB;
                }
            })
            .subscribe(new Subscriber<Boolean>() {
                @Override
                public void onCompleted() {

                }

                @Override
                public void onError(Throwable e) {

                }

                @Override
                public void onNext(Boolean aBoolean) {
                    if (aBoolean) {
                        _btnValidIndicator.setBackgroundColor(Color.BLUE);
                    } else {
                        _btnValidIndicator.setBackgroundColor(Color.parseColor("#888888"));
                    }
                }
            });
}


### 6.使用timer做定时操作。当有“x秒后执行y操作”类似的需求的时候想到使用timer
我们一般做APP会用到定时器，当用到定时器的。比如说再按一次退出程序，我们可以这么做。我们通过Timer计时，如果退出了的话会造成不到两秒的内存泄露，当然也可以先取消订阅在finish()页面。这里使用Timer来记录用户点击的次数，如果时间不足两秒，复原count

    `    int count = 1 ;
    onBackPressed(){
    if(count != 2){
        Toast.makeText(this, "再按一次退出应用", Toast.LENGTH_SHORT).show();
    }else{ finish();
    }
    count = 2;
    Observable.timer(2, TimeUnit.SECONDS)
        .subscribe(new Action1<Long>() {
            @Override
            public void call(Long aLong) {
                count = 1;
            }
        });
    }`

### 7.使用interval做周期性操作。当有“每隔xx秒后执行yy操作”类似的需求的时候，想到使用interval.
这也是我们经常用到的，就比如首页的轮播图就可以通过Interval来实现，我们每3秒让ViewPager当前显示的页面+1即可。具体代码可以见我的另一篇简书。

### 8.防抖点击
这里我使用的Fragmention也做了防抖点击，而且思想特别好，简单。他的做法是当我开启一个Fragmention的时候，我禁用掉Activity全屏的触摸事件,在开启之后在恢复。
这里呢，我们使用RxJava的限流操作符，在1秒内只允许流出来一个点击事件。

    ```
    RxView.clicks(button)  
              .throttleFirst(1, TimeUnit.SECONDS)  
              .subscribe(new Observer<Object>() {  
                  @Override  
                  public void onCompleted() {  
                        log.d ("completed");  
                  }  
  
                  @Override  
                  public void onError(Throwable e) {  
                        log.e("error");  
                  }  
  
                  @Override  
                  public void onNext(Object o) {  
                       log.d("button clicked");  
                  }  
              });  
    ```

RxJava的使用场景很多，遇到自己没见到过得操作符，可以看看图片和文档介绍，很详细，加油。

最后感谢awesome-Rxjava;你想要就的都在这里=>https://github.com/lzyzsd/Awesome-RxJava

附上本文的Demo：https://github.com/SshiGguang/Kits

RxJava从今以后你就会了

RxJava技术分享

京金所—时光

2016.9.22

这里我拿出来给 Android 开发者的 RxJava 详解中的例子

今天主要从以下几个方面来着重介绍RxJava

一：RxJava的重要概念

1.1 什么是RxJava？

1.2 RxJava难吗？

1.3 但是很多人说学习RxJava起点高啊？

1.4 响应式编程：

百度百科的定义：

在RxJava Essentials一书中，是这么说的。

在那些年我们错过的响应式编程中是这么介绍的：

最后我说一下我的理解:

1.4 数据流的概念：

1.5 函数式编程:

1.5.1 什么是函数式编程:

1.6 观察者模式

1.7 扩展的观察者模式

1.8 几个常见的类

总览：

一：Create的使用

2.Just、From的使用

3.Defer

4.Interval

5.Range

6.Repeat、Timer

总览：

1.Buffer

2.flatMap & concatMap

3.Map

4.GroupBy

5.Scan

总览：过滤

1.Filter

2.take/takelast

3.Distinct 有且仅有一次

4.First and last

5.Skip and SkipLast

Combining Observables

1.Zip

2.Merege

1.使用Scheduler进行线程的切换

几个常用的Api:

5.使用combineLatest做登录注册模块

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

android框架学习

Android知识

Android开发

Android技术知识

首页投稿（暂停使用，暂停投稿）

android成神之路

android技术