android学习之路(三)----RxJava/RxAndroid

RxJava学习一、基础篇

一、概述

https://github.com/ReactiveX/RxJava

二、基础

RxJava最核心的两个东西是Observables（被观察者，事件源）和Subscribers（观察者）。Observables发出一系列事件，Subscribers处理这些事件。这里的事件可以是任何你感兴趣的东西（触摸事件，web接口调用返回的数据）

一个Observable可以发出零个或者多个事件，直到结束或者出错。每发出一个事件，就会调用它的Subscriber的onNext方法，最后调用Subscriber.onNext()或者Subscriber.onError()结束。

Rxjava的看起来很像设计模式中的观察者模式，但是有一点明显不同，那就是如果一个Observerble没有任何的的Subscriber，那么这个Observable是不会发出任何事件的。

1. Hello World

创建一个Observable对象很简单，直接调用Observable.create即可

Observable<String> myObservable = Observable.create(
new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> sub) {
sub.onNext("Hello, world!");
sub.onCompleted();
}
}
);

这里定义的Observable对象仅仅发出一个Hello World字符串，然后就结束了。接着我们创建一个

Subscriber

来处理Observable对象发出的字符串。

Subscriber<String> mySubscriber = new Subscriber<String>() {
@Override
public void onNext(String s) { System.out.println(s); }
@Override
public void onCompleted() { }
@Override
public void onError(Throwable e) { }
};

这里subscriber仅仅就是打印observable发出的字符串。通过subscribe函数就可以将我们定义的myObservable对象和mySubscriber对象关联起来，这样就完成了subscriber对observable的订阅。

myObservable.subscribe(mySubscriber);

一旦mySubscriber订阅了myObservable，myObservable就是调用mySubscriber对象的onNext和onComplete方法，mySubscriber就会打印出Hello World！

2. 更简洁的代码

是不是觉得仅仅为了打印一个hello world要写这么多代码太啰嗦？我这里主要是为了展示RxJava背后的原理而采用了这种比较啰嗦的写法，RxJava其实提供了很多便捷的函数来帮助我们减少代码。

首先来看看如何简化Observable对象的创建过程。RxJava内置了很多简化创建Observable对象的函数，比如Observable.just就是用来创建只发出一个事件就结束的Observable对象，上面创建Observable对象的代码可以简化为一行

Observable<String> myObservable = Observable.just("Hello, world!");

接下来看看如何简化Subscriber，上面的例子中，我们其实并不关心OnComplete和OnError，我们只需要在onNext的时候做一些处理，这时候就可以使用Action1类。

Action1<String> onNextAction = new Action1<String>() {
@Override
public void call(String s) {
System.out.println(s);
}
};

subscribe方法有一个重载版本，接受三个Action1类型的参数，分别对应OnNext，OnComplete， OnError函数。

myObservable.subscribe(onNextAction, onErrorAction, onCompleteAction);

这里我们并不关心onError和onComplete，所以只需要第一个参数就可以

myObservable.subscribe(onNextAction);

上面的代码最终可以写成这样

Observable.just("Hello, world!")
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
System.out.println(s);
}
});

使用java8的lambda可以使代码更简洁

Observable.just("Hello, world!")
.subscribe(s -> System.out.println(s));

3. 变换

让我们做一些更有趣的事情吧！

比如我想在hello world中加上我的签名，你可能会想到去修改Observable对象：

Observable.just("Hello, world! -Dan")
.subscribe(s -> System.out.println(s));

如果你能够改变Observable对象，这当然是可以的，但是如果你不能修改Observable对象呢？比如Observable对象是第三方库提供的？比如我的Observable对象被多个Subscriber订阅，但是我只想在对某个订阅者做修改呢？

那么在Subscriber中对事件进行修改怎么样呢？比如下面的代码：

Observable.just("Hello, world!")
.subscribe(s -> System.out.println(s + " -Dan"));

这种方式仍然不能让人满意，因为我希望我的Subscribers越轻量越好，因为我有可能会在mainThread中运行subscriber。另外，根据响应式函数编程的概念， Subscribers更应该做的事情是“响应”，响应Observable发出的事件，而不是去修改。如果我能在某些中间步骤中对“Hello World！”进行变换是不是很酷？

4.操作符（Operators）

操作符就是为了解决对Observable对象的变换的问题，操作符用于在Observable和最终的Subscriber之间修改Observable发出的事件。RxJava提供了很多很有用的操作符。

比如map操作符，就是用来把一个事件转换为另一个事件的。

Observable.just("Hello, world!")
.map(new Func1<String, String>() {
@Override
public String call(String s) {
return s + " -Dan";
}
})
.subscribe(s -> System.out.println(s));

使用lambda可以简化为

Observable.just("Hello, world!")
.map(s -> s + " -Dan")
.subscribe(s -> System.out.println(s));

是不是很酷？map()操作符就是用于变换Observable对象的，map操作符返回一个Observable对象，这样就可以实现链式调用，在一个Observable对象上多次使用map操作符，最终将最简洁的数据传递给Subscriber对象。

5.map操作符进阶

map操作符更有趣的一点是它不必返回Observable对象返回的类型，你可以使用map操作符返回一个发出新的数据类型的observable对象。

比如上面的例子中，subscriber并不关心返回的字符串，而是想要字符串的hash值

Observable.just("Hello, world!")
.map(new Func1<String, Integer>() {
@Override
public Integer call(String s) {
return s.hashCode();
}
})
.subscribe(i -> System.out.println(Integer.toString(i)));

很有趣吧？我们初始的Observable返回的是字符串，最终的Subscriber收到的却是Integer，当然使用lambda可以进一步简化代码：

Observable.just("Hello, world!")
.map(s -> s.hashCode())
.subscribe(i -> System.out.println(Integer.toString(i)));

前面说过，Subscriber做的事情越少越好，我们再增加一个map操作符

Observable.just("Hello, world!")
.map(s -> s.hashCode())
.map(i -> Integer.toString(i))
.subscribe(s -> System.out.println(s));

总结：

1.Observable和Subscriber可以做任何事情

Observable可以是一个数据库查询，Subscriber用来显示查询结果；Observable可以是屏幕上的点击事件，Subscriber用来响应点击事件；Observable可以是一个网络请求，Subscriber用来显示请求结果。

2.Observable和Subscriber是独立于中间的变换过程的。

在Observable和Subscriber中间可以增减任何数量的map。整个系统是高度可组合的，操作数据是一个很简单的过程。

RxJava学习二、操作符

三、准备工作

假设我有这样一个方法：

这个方法根据输入的字符串返回一个网站的url列表（啊哈，搜索引擎）

Observable<List<String>> query(String text);

现在我希望构建一个健壮系统，它可以查询字符串并且显示结果。根据上一篇blog的内容，我们可能会写出下面的代码：

query("Hello, world!")
.subscribe(urls -> {
for (String url : urls) {
System.out.println(url);
}
});

这种代码当然是不能容忍的，因为上面的代码使我们丧失了变化数据流的能力。一旦我们想要更改每一个URL，只能在Subscriber中来做。我们竟然没有使用如此酷的map()操作符！！！

当然，我可以使用map操作符，map的输入是urls列表，处理的时候还是要for each遍历，一样很蛋疼,万幸，还有Observable.from()方法，它接收一个集合作为输入，然后每次输出一个元素给subscriber：

Observable.from("url1", "url2", "url3")
.subscribe(url -> System.out.println(url));

我们来把这个方法使用到刚才的场景：

query("Hello, world!")
.subscribe(urls -> {
Observable.from(urls)
.subscribe(url -> System.out.println(url));
});

虽然去掉了for each循环，但是代码依然看起来很乱。多个嵌套的subscription不仅看起来很丑，难以修改，更严重的是它会破坏某些我们现在还没有讲到的RxJava的特性。

四、改进

救星来了,他就是flatMap()。

Observable.flatMap()

接收一个Observable的输出作为输入，同时输出另外一个Observable。直接看代码：

query("Hello, world!")
.flatMap(new Func1<List<String>, Observable<String>>() {
@Override
public Observable<String> call(List<String> urls) {
return Observable.from(urls);
}
})
.subscribe(url -> System.out.println(url));

Lambda:

query("Hello, world!")
.flatMap(urls -> Observable.from(urls))
.subscribe(url -> System.out.println(url));

flatMap()

是不是看起来很奇怪？为什么它要返回另外一个

Observable

呢？理解

flatMap

的关键点在于，

flatMap

输出的新的

Observable

正是我们在

Subscriber

想要接收的。现在Subscriber不再收到

List<String>

，而是收到一些列单个的字符串，就像

Observable.from()

的输出一样。

1.还可以更好

flatMap()

实在不能更赞了，它可以返回任何它想返回的Observable对象。

比如下面的方法：

// 返回网站的标题，如果404了就返回null
Observable<String> getTitle(String URL);

接着前面的例子，现在我不想打印URL了，而是要打印收到的每个网站的标题。问题来了，我的方法每次只能传入一个URL，并且返回值不是一个String，而是一个输出String的Observabl对象。使用flatMap()可以简单的解决这个问题。

query("Hello, world!")
.flatMap(urls -> Observable.from(urls))
.flatMap(new Func1<String, Observable<String>>() {
@Override
public Observable<String> call(String url) {
return getTitle(url);
}
})
.subscribe(title -> System.out.println(title));

简化：

query("Hello, world!")
.flatMap(urls -> Observable.from(urls))
.flatMap(url -> getTitle(url))
.subscribe(title -> System.out.println(title));

是不是感觉很不可思议？我竟然能将多个独立的返回Observable对象的方法组合在一起！帅呆了！

不止这些，我还将两个API的调用组合到一个链式调用中了。我们可以将任意多个API调用链接起来。大家应该都应该知道同步所有的API调用，然后将所有API调用的回调结果组合成需要展示的数据是一件多么蛋疼的事情。这里我们成功的避免了callback hell（多层嵌套的回调，导致代码难以阅读维护）。现在所有的逻辑都包装成了这种简单的响应式调用。

2.丰富的操作符

目前为止，我们已经接触了两个操作符，RxJava中还有更多的操作符，那么我们如何使用其他的操作符来改进我们的代码呢？

getTitle()

返回null如果url不存在。我们不想输出”null”，那么我们可以从返回的title列表中过滤掉null值！

query("Hello, world!")
.flatMap(urls -> Observable.from(urls))
.flatMap(url -> getTitle(url))
.filter(title -> title != null)
.subscribe(title -> System.out.println(title));

filter()

输出和输入相同的元素，并且会过滤掉那些不满足检查条件的。

如果我们只想要最多5个结果：

query("Hello, world!")
.flatMap(urls -> Observable.from(urls))
.flatMap(url -> getTitle(url))
.filter(title -> title != null)
.subscribe(title -> System.out.println(title));

filter()输出和输入相同的元素，并且会过滤掉那些不满足检查条件的。

如果我们只想要最多5个结果：

query("Hello, world!")
.flatMap(urls -> Observable.from(urls))
.flatMap(url -> getTitle(url))
.filter(title -> title != null)
.take(5)
.subscribe(title -> System.out.println(title));

take()输出最多指定数量的结果。

query("Hello, world!")
.flatMap(urls -> Observable.from(urls))
.flatMap(url -> getTitle(url))
.filter(title -> title != null)
.take(5)
.doOnNext(title -> saveTitle(title))
.subscribe(title -> System.out.println(title));

doOnNext()允许我们在每次输出一个元素之前做一些额外的事情，比如这里的保存标题。看到这里操作数据流是多么简单了么。你可以添加任意多的操作，并且不会搞乱你的代码。RxJava包含了大量的操作符。操作符的数量是有点吓人，但是很值得你去挨个看一下，这样你可以知道有哪些操作符可以使用。弄懂这些操作符可能会花一些时间，但是一旦弄懂了，你就完全掌握了RxJava的威力。你甚至可以编写自定义的操作符！这篇blog不打算将自定义操作符，如果你想的话，清自行Google吧。

RxJava学习三、错误处理

五、错误处理

到目前为止我们完全忽略了订阅者的onComplete()和onError()方法，对于原始的订阅者，是有能力去处理onComplete()和onError(),比如：

Observable.just("Hello, world!")
.map(s -> potentialException(s))// potentialException抛出异常
.map(s -> anotherPotentialException(s)) // anotherPotentialException抛出异常
.subscribe(new Subscriber<String>() {
@Override
public void onNext(String s) { System.out.println(s); }
@Override
public void onCompleted() { System.out.println("Completed!"); }
@Override
public void onError(Throwable e) { System.out.println("Ouch!"); }
});

//结果会打印出

Completed!
Ouch!

1.1 onError()在任何时候，只要有异常被抛出，就会被调用

1.2 操作符不必处理异常

1.3 你可以监听到订阅者接收完成的时间

在RxJava的模式下，订阅者甚至不需要知道怎样去处理错误，也不需要知道操作符如何去处理错误，只要不是崩溃式地异常都会跳过操作符的相关操作，所以，对于错误的处理，我们可以放在onErrors方法当中

六、调度器

如果你的android应用需要一个网络请求操作，这个请求会很消耗时间，因此你需要开一子线程，android当中的多线程处理起来不是很容易，因为你必须确保在正确地线程里面运行正确的代码，否则可能就会挂掉，通常的错误就是在非主线程里面操作了界面(subscribeOn, observeOn操作符)

myObservableServices.retrieveImage(url)
.subscribeOn(Schedulers.io())//子线程
.observeOn(AndroidSchedulers.mainThread())//定义订阅者运行的线程主线程
.subscribe(bitmap -> myImageView.setImageBitmap(bitmap));

七、Subscriptions

Observable.subscribe()方法会返回一个对象，这个对象就是Subscriptions ,这个对象表示的就是被观察者和订阅者之间的联系

Subscription subscription = Observable.just("Hello, World!")
.subscribe(s -> System.out.println(s));
subscription.unsubscribe();//取消订阅关系,
subscription.isUnsubscribed()//订阅者是否订阅了被观察者的消息

RxJava学习五、深入学习

八、Observable的创建：

1. Observable<T> create(OnSubscribe<T> f)
通过订阅者创建一个被观察者，当被观察者的消息到达订阅者的时候，订阅者会执行特定的方法

Observable<Data> observable = Observable.<Data>create(subscriber -> {
subscriber.onNext(memory);
subscriber.onCompleted();
});

2.amb:将多个Observable<? extends T>组合为一个Observable，参数可以是一个Iterator或者
2~9个Observable<? extends T>
3. combineLatest:将多个不同元素的Observable<? extends T>组合成一个Observable<R>
4. concat:将1~9个Observable<? extends T>合并成一个
5. defer:通过一个观察者工厂，将原有的被观察者转换成一个新的Observable<? extends T>
6. EMPTY:创建一个空的被观察者，只会调用订阅者的onComplete()
7. error:传递一个Throwable,当到达订阅者的时候，只会调用onError
8.from:接收的参数类型：
Future<? extends T>
Future<? extends T> future, long timeout, TimeUnit unit
//timeout是指future在调用get方法最大的等待时间
Future<? extends T> future, Scheduler scheduler
Iterable<? extends T> iterable
T[] array
9. interval:每隔一段时间发出一段序列
long interval, TimeUnit unit
long interval, TimeUnit unit, Scheduler scheduler
10. just : 发出一个item就会complete的观察者(参数1~10个T)
11. merge: 将一系列的Observable平整为一个Observable,形参类型
Iterable<? extends Observable<? extends T>> sequences
Iterable<? extends Observable<? extends T>> sequences, int maxConcurrent
maxConcurrent:表示最多平整多少个Observable
Observable<? extends Observable<? extends T>> source
Observable<? extends Observable<? extends T>> source, int maxConcurrent
Observable<? extends T>[] sequences
12. mergeDelayError : 将一系列的Observable平整为一个Observable,允许订阅者接收所有的消息，即使某个消息发生异常也不影响后续消息的接收,形参类型
Observable<? extends Observable<? extends T>> source
2~9个Observable<? extends T>
13.never :将本Observable<T>转换成不发出任何元素的Observable<T>
14.range :创建能发出制定int序列的Observable<Integer>,形参：
int start, int count
int start, int count, Scheduler scheduler
15.sequenceEqual :比较两个Observable<? extends T>发出的元素是否相同，并返回boolean类型的Observable,形参类型：
Observable<? extends T> first, Observable<? extends T> second
Observable<? extends T> first, Observable<? extends T> second, Func2<? super T, ? super T, Boolean> equality (equality用来定义比较规则)
16. switchOnNext:将Observable<? extends Observable<? extends T>> origin平整成一个
Observable<? extends T>,比如origin发出了两个Observable A和B，A向外发出T,一段时间后B开始向外发出T，则此时A会停止发出T，订阅者此之后只会接受到B发出的T。
17. timer : 每隔一段时间向外发出0L,形参类型：
long initialDelay, long period, TimeUnit unit
long initialDelay, long period, TimeUnit unit, Scheduler scheduler
long delay, TimeUnit unit
long delay, TimeUnit unit, Scheduler scheduler
18. using :依赖于其他资源对象类型创建一个Observable，参数类型：
final Func0<Resource> resourceFactory, //创建一个资源对象的工厂
final Func1<? super Resource, ? extends Observable<? extends T>> observableFactory,
//根据一个资源对象创建一个Observable<? extends T>>
final Action1<? super Resource> disposeAction,
//当被观察者和订阅者之间的订阅关系解除的时候，或者订阅者调用了onComplete()\onError()的时候调用，这个disposeAction只会执行一次
boolean disposeEagerly//如果设置为false，在解除关系的时候就不会调用disposeAction
19.zip :通过一个接口将原Observable当中发出的元素转换成目标类型的元素，组成新的Observable
形参类型：
②   Iterable<? extends Observable<?>> ws, FuncN<? extends R> zipFunction
② Observable<? extends Observable<?>> ws, final FuncN<? extends R> zipFunction
③ Observable<? extends T1> o1, Observable<? extends T2> o2, final Func2<? super T1, ? super T2, ? extends R> zipFunction
④Observable<? extends T1> o1,…, Observable<? extends T9> o9, Func9<? super T1, …, ? super T9, ? extends R> zipFunction

七．流中间操作符

20.lift:通过一个Func1<Subscriber<? super R>, Subscriber<? super T>>接口，返回一个新的被观察者
21.compose:通过一个Func1<Observable<T>, Observable<R>>接口，返回一个新的被观察者
22.nest : 将本Observable<T>转换成只含有一个item的Observable<Observable<T>>
23 all : 返回是否满足条件的Obserable<Boolean>
24 ambWith:将此Observable和一个Observable组合起来
25 asObservable:将此Observable（可能是Observale的子类）转为顶层的Observalbe
25 buffer:缓存发出的items,参数类型
① Func0<? extends Observable<? extends TClosing>> bufferClosingSelector
② int count
③   int count, int skip
④   long timespan, long timeshift, TimeUnit unit
timespan 缓存区在向外发出item之前，收集items的时间
timeshift 在一个新的缓存区创建之后，的空闲时间
⑤   long timespan, TimeUnit unit, int count, Scheduler scheduler
⑥   long timespan, TimeUnit unit, Scheduler scheduler
⑦  Observable<B> boundary
当boundary发出一个item的时候，返回一个item为一段缓存数据List<T>的Observable
⑧ Observable<B> boundary, int initialCapacity
initialCapacity:初始化每一段缓存数据当中item的数量
26.cache()
缓存从源Observable发出的items，并且将它们依次发出订阅者列表，这个方法与replay()相似,区别是这个方法会自动订阅源Observable,而replay()会返回一个ConnectableObservable，开发者需要手动调用
cache(int capacity):缓存的数量
27. cast(final Class<R> klass):返回一个Observable<R>,会将源Observable当中发出的item转换成目标类型的item,注意，这里的转换只是将原item强制类型转换成R
28. collect:将源Observable里面的items收集到一个单独的、不稳定的数据结构R当中，并返回Observable<R>,参数列表：
Func0<R> stateFactory, 创建数据结构R,这个R会收集到源Observable的所有items
Action2<R, ? super T> collector
30.concatMap:将每个item转换成新类型的item之后，将源Observalbes组合成一个Observable,参数列表： Func1<? super T, ? extends Observable<? extends R>> func
源码：concat(map(func))
31. concatWith:将本Observable和新的Observable合并，参数列表
Observable<? extends T> t1
32. contains():源Observable是否包含制定的元素，如果包含，则返回Observable<true>
33. count()
34. countLong()
35. debounce():在特定的时间间隔里面，只发出一个item，忽略改时间段其他的item
①long timeout, TimeUnit unit
②long timeout, TimeUnit unit, Scheduler scheduler
③Func1<? super T, ? extends Observable<U>> debounceSelector
36. defaultIfEmpty(T t):如果Observable为空，则返回只包含一个t的Observable
37. switchIfEmpty(Observable<? extends T> alternate):如果Observable为空，则返回
alternate
38. delay():延迟item的发出时间,异常不会延迟发出，参数列表
①long delay, TimeUnit unit
②long delay, TimeUnit unit, Scheduler scheduler
39. delaySubscription():延迟订阅时间
① long delay, TimeUnit unit
② long delay, TimeUnit unit, Scheduler scheduler
③ Func0<? extends Observable<U>> subscriptionDelay: subscriptionDelay发出一个事件，结束的时候，创建source Observable与订阅者之间的订阅关系
40. materialize():将源Observable<T>发出的元素T转换成Notification<T>并发出
41. dematerialize:将元素为Notification<T>转换成T
42. distinct(),参数可以为空，也可以是：
Func1<? super T, ? extends U> keySelector:用于决定是否跟another item相同,T为输入元素，U为T元素对应的用于表示T是否相同的key
43. distinctUntilChanged(),参数与distinct()相同，区别是，distinctUntilChanged发出的某个元素与后面紧接着的一个元素相同，则不发出，采用后一个item，依次类推
44. doOnCompleted,doOnNext,doOnError:在订阅者调用到onCompleted,onNext,onError的时候分别调用的方法，注意，doXX是在XX方法体执行之前执行的
45. doOnSubscribe :当一个观察者注册到悲观者的时候调用
46. doOnTerminate : 当流终止的时候调用，正常终止或者异常终止都会调用
47. finallyDo：当观察者调用onCompleted/onError执行
48. first ： 当前的Observable<T>只会发出第一个T,如果为empty,则会抛出NoSuchElement
Exception,参数类型可以是空，或者Func1<? super T, Boolean> predicate,表示满足predicate的第一个元素
49. firstOrDefault:当源Observable在结束之前没有发出任何item，默认发出的元素，参数类型：
① T defaultValue
② T defaultValue, Func1<? super T, Boolean> predicate
50.flatMap:平整成为一个Observable<T>,接收的参数类型
①Func1<? super T, ? extends Observable<? extends R>> func
②Func1<? super T, ? extends Observable<? extends R>> func, int maxConcurrent
maxConcurrent:同时发出事件的Observable的最大个数，
③Func1<? super T, ? extends Observable<? extends R>> onNext,
Func1<? super Throwable, ? extends Observable<? extends R>> onError,
Func0<? extends Observable<? extends R>> onCompleted
④Func1<? super T, ? extends Observable<? extends R>> onNext,
Func1<? super Throwable, ? extends Observable<? extends R>> onError,
Func0<? extends Observable<? extends R>> onCompleted,
int maxConcurrent
⑤final Func1<? super T, ? extends Observable<? extends U>> collectionSelector,
final Func2<? super T, ? super U, ? extends R> resultSelector
U:是Collection发出的数据类型,这种类型的flatMap的作用就是将U类型的Observable平整成为一个新的数据类型R的Observable
⑥final Func1<? super T, ? extends Observable<? extends U>> collectionSelector,
final Func2<? super T, ? super U, ? extends R> resultSelector,
int maxConcurrent
51. flatMapIterable:将一个Iterable平整成为一个Observable<R>
①Func1<? super T, ? extends Iterable<? extends R>> collectionSelector
②Func1<? super T, ? extends Iterable<? extends U>> collectionSelector,
Func2<? super T, ? super U, ? extends R> resultSelector
52. forEach:订阅,参数类型：
①final Action1<? super T> onNext,源码：subscribe(onNext);
②final Action1<? super T> onNext, final Action1<Throwable> onError: subscribe(onNext, onError);
③final Action1<? super T> onNext, final Action1<Throwable> onError, final Action0 onComplete: subscribe(onNext, onError, onComplete);
52. groupBy:将Observable发出的元素根据一定的规则进行分组，分组之后的每组元素以GroupedObservable的形式存在，也就是每组就是一个GroupedObservable,该函数的返回类型是：
Observable<GroupedObservable<K, R>>,注意，一个GroupedObservable会缓存items，应该注意内存泄漏的问题
①final Func1<? super T, ? extends K> keySelector, //生成每组的key(分组条件)
final Func1<? super T, ? extends R> elementSelector
②final Func1<? super T, ? extends K> keySelector

53. groupJoin:当两个相关联的Observable有元素重叠时，将结果分组合并,参数类型：
Observable<T2> right, 与源Observable相关联的Observable
Func1<? super T, ? extends Observable<D1>> leftDuration,表明源Observable向外发出item的持续时间
Func1<? super T2, ? extends Observable<D2>> rightDuration,
表明源right向外发出item的持续时间
Func2<? super T, ? super Observable<T2>, ? extends R> resultSelector
54.join: join操作符结合了两个Observable对象发出的item，并且会根据在每个item上面定义的duration-window规则决定哪些item将被结合，另外的一个Observable发出一个item，就作为本Observable生命周期的开始，据此来定义duration-window,当这样一个被定义了window-defining的Observable无论是发出一个item或者completes,关于这个item的window关联将被自动关闭，只要当一个item的window是打开的，他就可以结合其他Observable发出的任何item，items的结合方法需要开发者自定义,参数类型：
Observable<TRight> right,
Func1<T, Observable<TLeftDuration>> leftDurationSelector,源Observable发出每个item的持续时间，用来表明是否与right发出的元素发出重叠
Func1<TRight, Observable<TRightDuration>> rightDurationSelector,
Func2<T, TRight, R> resultSelector
55. ignoreElements():忽略Observable发出的所有元素，只会调用onComplete和onError
56. isEmpty():如果Observable为空，返回Observable<true>
57. last():参数类型
Func1<? super T, Boolean> predicate
58. lastOrDefault(),参数类型：
①T defaultValue:当源Observable在调用onCompleted之前没有发出任何item,返回defaultValue
②T defaultValue, Func1<? super T, Boolean> predicate
59. limit(int num):发出最前面num个元素
60. map:参数类型Func1<? super T, ? extends R> func,将源Observable里面的每个元素，转换成R类型
61. mapNotification():不说了
62. materialize():对每个元素封装成一个Notification<T>
63. mergeWith():参数类型Observable<? extends T> t1,将源Observable和t1不做任何改变合并成一个Observable
64. observeOn():参数类型Scheduler scheduler,将接收Observable 元素的动作在指定的线程上面执行。
65. ofType():参数类型Class<R> klass,过滤出元素类型为klass的元素
66. onBackpressureBuffer():当Observable发出的元素的速度比订阅者消化得要快，在这些元素被发出之前，会首先将这些元素无限制地缓存起来.
67. onBackpressureDrop():当Observable发出的元素的速度比订阅者消化得要快，没有来得及消化的item会被丢弃，参数类型可以为空，也可以是：Action1<? super T> onDrop:被丢弃的回调
68. onBackpressureBlock():当Observable发出的元素的速度比订阅者消化得要快,当Observer还在处理item的时候，Observable发出元素的线程会被阻塞，接收的参数类型：可以为空，也可以是int maxQueueLength:表示在没有阻塞的情况下，Observable能发出的元素个数
69. onBackpressureLatest():当Observable发出的元素的速度比订阅者消化得要快,订阅者会接收Observable最新发出的item进行处理
70. onErrorResumeNext():默认情况下，当一个Observable发生异常，他的订阅者会调用onError方法，并退出，onErrorResumeNext方法会改变这种流程，当源Observable引发一个错误，不会调用订阅者的onError方法，源Observable会将错误的控制权交给resumeFunction返回的Observable，这个Observable会调用订阅者的onNext(),这样一来，订阅者无法知道Observable是否发生异常了。参数类型：
①final Func1<Throwable, ? extends Observable<? extends T>> resumeFunction
②final Observable<? extends T> resumeSequence:返回一个Observable,当源Observable发生异常，用resumeSequence替换原来的处理流程
71. onErrorReturn:当有异常发生的时候，不会调用订阅者的onError(),而是会发出resumeFunction的返回值，接收的参数类型：
Func1<Throwable, ? extends T> resumeFunction
72. onExceptionResumeNext
73. publish():返回一个ConnectableObservable<T>,这种Observable会在connect（）方法被调用之前一直处于等待的状态
74. reduce():将所有的元素通过运算成为一个包含最终的、唯一的元素的Observable，通常可以执行累加之类的运算，参数类型：
Func2<T, T, T> accumulator : 两个元素经过运算，返回一个T元素
R initialValue, Func2<R, ? super T, R> accumulator:返回Observable<R>
75. repeat(): 将原有的item序列重复发出，接收的参数类型
① 空
②Scheduler scheduler
③final long count : 重复的次数，如果为0会发出一个空的序列
④final long count, Scheduler scheduler
76. repeatWhen():参数类型final Func1<? super Observable<? extends Void>, ? extends Observable<?>> notificationHandler, Scheduler scheduler
77. replay() : 返回一个ConnectableObservable<T>,这个Observable会发出源Observable当中的元素，除非该Observable被订阅，或者被调用了connect()方法，否则不会向外发出item,接收的参数类型可以为空，也可以是
①Func1<? super Observable<T>, ? extends Observable<R>> selector
②Func1<? super Observable<T>, ? extends Observable<R>> selector, final int bufferSize
bufferSize:决定了此ConnectableObservable能发出的元素个数
④Func1<? super Observable<T>, ? extends Observable<R>> selector, int bufferSize, long time, TimeUnit unit
Time:每个item发出的最大持续时间
78. retry():当源Observable发出item的过程当中发生了错误，比如向外发出了[1,2]发生了错误，后续的item不会继续发出，此时会调用订阅者的onError,retry()方法会重新将Observable和订阅者之间的关系连接起来，并重新发出items，所以总的items顺序应该是[1,2,1,2,3,4,5,onComplete],接收的参数：
① 空
② long count : retry的次数
③Func2<Integer n, Throwable e, Boolean> predicate :判断在retry(Integer)n次，异常类型是e的情况下，是否继续retry
79.sample():返回一个Observable,包含源Observable最近发出的items，参数类型
long period, TimeUnit unit :确定采样间隔
long period, TimeUnit unit, Scheduler scheduler
80.scan():对当前发出的item和前一个item进行操作，并转换为新item发出,比如，发出的item序列为[1,2,3,4,5],通过scan累加之后，为[1,3,6,10,15,onComplete],接收的参数类型：
①Func2<T, T, T> accumulator
②R initialValue, Func2<R, ? super T, R> accumulator
81. serialize():通常情况下，Observable当中item的发出和订阅者接收消息默认是异步的，即消息的发出和接收是在不同的线程当中进行，使用serialize()可以使消息的发出和接收在同一个线程
83.refCount():将一个ConnectableObservable转换成为一个普通的Observable, Connectable
Observable需要首先订阅者订阅subscribe()之后，调用connect()之后才会发出item，而普通的Observable不需要connect().
82. share():publish().refCount();
83. single():如果原Observable只有一个item，则发出这个唯一的item，如果为空，则会抛出NoSuchElementException,如果超过一个，则会抛出IllegalArgumentException,接收的参数类型：
①空
②Func1<? super T, Boolean> predicate,限定单一的条件，即如果原Observable有多个item，只有一个item符合predicate，也视为single()
84. singleOrDefault : 如果源Observable为空，则发出默认值的元素，参数类型：
①T defaultValue
②T defaultValue, Func1<? super T, Boolean> predicate
85.skip():跳过前面n个item，发出后续的items
① int num
② long time, TimeUnit unit,定义一段window时间，跳过
③ long time, TimeUnit unit, Scheduler scheduler
86. skipLast():跳过最后的n个items,参数类型同skip()
87. skipUntil():源Observable发出的item会被忽略，直到other发出一个item
88. skipWhile():当predicate发出true的时候，源Observable会跳过发出的items，直到predicate发出false的时候，订阅者才会接收到源Observable后续的items,接收的参数类型：
Func1<? super T, Boolean> predicate
89.startWith():将传入的Observable/集合当中的items合并到源Observable的前面,接收的参数类型：
①Observable<T> values
②Iterable<T> values
③T… t1(1~9个T)
90. subscribe():对被观察者进行订阅，接收的参数类型：
①空：会忽略Observable发出的任何信息
②Action1<? super T> onNext : 接收next()消息，对complete和error不作处理
③final Action1<? super T> onNext, final Action1<Throwable> onError
④final Action1<? super T> onNext, final Action1<Throwable> onError, final Action0 onComplete
⑤final Observer<? super T> observer: observer会处理next/error/complete
⑥ Subscriber<? super T> subscriber
91. unsafeSubscribe:取消订阅关系,接收参数Subscriber<? super T> subscriber
92. subscribeOn:在scheduler指定的线程策略接收Observable发出的消息，接收的参数类型：Scheduler scheduler
93. switchMap():将源Observable发出的每个item转换成一个对应的Observable<R>,然后将这些Observable<R>最新发出的item分发出去,参数类型：
Func1<? super T, ? extends Observable<? extends R>> func
94. take():发出源Observable前n个元素，或者在发出在指定时间内源Observable发出的元素，
接收的参数类型：
①int num;
②long time, TimeUnit unit
③long time, TimeUnit unit, Scheduler scheduler
95.takeFirst():发出满足给定条件下的第一个item,参数类型
Func1<? super T, Boolean> predicate
96. takeLast(final int count) 不解释
97. throttleFirst():在指定连续的时间内向外发出指定时间内的第一个item,接收的参数类型
①long windowDuration, TimeUnit unit
②long skipDuration, TimeUnit unit, Scheduler scheduler
98. timeInterval():返回源Observable发出的相邻两个元素的时间间隔，所以该方法的返回值是：
Observable<TimeInterval<T>>,接收的参数类型可以是空，也可以是Scheduler
99.timeout():在指定的时间内，如果源Observable没有发出任何元素，则会报TimeOutException,这样开发者就可以在onError处理这个异常，并结束整个过程,接收的参数类型：
①Func0<? extends Observable<U>> firstTimeoutSelector, 决定了源Observable发出第一个元素的超时时间，
Func1<? super T, ? extends Observable<V>> timeoutSelector) ,决定了源Observable后续发出的元素时间间隔，以确定是否超时
②Func0<? extends Observable<U>> firstTimeoutSelector,
Func1<? super T, ? extends Observable<V>> timeoutSelector,
Observable<? extends T> other :如果发生超时，就用这个Observable替换源Observable
③Func1<? super T, ? extends Observable<V>> timeoutSelector :在上一个item到达的时间作为起点时间，经过一段时间之后，如果没有另外的item到达，则视为超时，timeoutSelector就是定义超时时间的
④Func1<? super T, ? extends Observable<V>> timeoutSelector, Observable<? extends T> other
⑤long timeout, TimeUnit timeUnit:定义超时时间
⑥long timeout, TimeUnit timeUnit, Observable<? extends T> other
⑦long timeout, TimeUnit timeUnit, Observable<? extends T> other, Scheduler scheduler
⑧long timeout, TimeUnit timeUnit, Scheduler scheduler
100. timestamp():将源Observable当中的元素封装成Timestamped<T>,替代原来的元素发出去
101. toBlocking():将原Observable转换为BlockingObservable<T>
102. toList():将原Observable当中的所有元素装到一个List当中，这样就只能发出一个item了，这个item就是List<T>,所以返回值是一个Observable<List<T>>
103. toMap:参数类型是Func1<? super T, ? extends K> keySelector,对于每个item，以K类型为Key,T类型为value构建一个Map,所以返回值是Observable<Map<K, T>>,该Observable只能返回一个类型为Map<K,T>的item,除此之外，参数类型还包括：
①Func1<? super T, ? extends K> keySelector, Func1<? super T, ? extends V> valueSelector
分别对应了生成key和生成value的function
②Func1<? super T, ? extends K> keySelector, Func1<? super T, ? extends V> valueSelector, Func0<? extends Map<K, V>> mapFactory:生成一个map实例
104. toMultimap:将原Observable当中元素生成多个Map,返回值类型Observable<Map<K, Collection<T>>>接收的参数类型
①Func1<? super T, ? extends K> keySelector:定义生成key的策略，这个操作符处理后的Observable只包含一个元素Map<K,Collection<T>>，map类型为hashMap
②Func1<? super T, ? extends K> keySelector, Func1<? super T, ? extends V> valueSelector
③Func1<? super T, ? extends K> keySelector, Func1<? super T, ? extends V> valueSelector, Func0<? extends Map<K, Collection<V>>> mapFactory
④Func1<? super T, ? extends K> keySelector, Func1<? super T, ? extends V> valueSelector, Func0<? extends Map<K, Collection<V>>> mapFactory, Func1<? super K, ? extends Collection<V>> collectionFactory
105. toSortedList(): 返回一个Observable<List<T>>,其中的唯一元素List<T>是一个经过排序之后的集合，接收的参数类型：
①空，源Observable当中的每一个T都必须实现接口Comparable
②Func2<? super T, ? super T, Integer> sortFunction ： 定义了两个元素之间的比较规则
③int initialCapacity,定义了list当中在进行排序前，盛放的元素个诉， T都必须实现接口Comparable,
④Func2<? super T, ? super T, Integer> sortFunction, int initialCapacity
106. unsubscribeOn:将Observable与它的观察者们接触对应关系,接收的参数类型Scheduler scheduler
107. withLatestFrom:将源Observable<T>和另外一个Observable<U>进行合并，当源Observable发出一个元素T的时候，会与Observable<U>进行对应的元素运算，得到一个R并发出去，接收的参数类型：
Observable<? extends U> other, Func2<? super T, ? super U, ? extends R> resultSelector
108.window:将源Observable当中元素装入多个Observable当中，装入的Observable就叫做window,根据需要，将这些Observable中的元素发送出去,所以返回值的类型，应该是Observable<Observable<T>>
①Func0<? extends Observable<? extends TClosing>> closingSelector
当源Observable发出一个item，closingSelector就会发出一个window,并创建一个新的window
②int count
每个window包含元素的最大个数
③int count, int skip:
skip定义了在源Observable上面进行元素收集的步长，count决定了每个window包含的元素个数，比如skip=3,count=2,那么将原Observable以3为单位划分为n个window，然后取最前面2个元素放入window
④long timespan, 在每个window向外发出item之前，收集item的时间
long timeshift, 每个window创建之前的持续时间
TimeUnit unit
⑤long timespan, long timeshift, TimeUnit unit, Scheduler scheduler
⑥long timespan, long timeshift, TimeUnit unit, int count, Scheduler scheduler
⑦long timespan, TimeUnit unit
⑧long timespan, TimeUnit unit, int count
⑨long timespan, TimeUnit unit, int count, Scheduler scheduler
⑩long timespan, TimeUnit unit, Scheduler scheduler
109.zipWith():根据源Observable中发出的item和指定的Observable发出的item,发出一个其他类型的item,返回值类型Observable<R>,接收的参数类型是：
①Iterable<? extends T2> other, Func2<? super T, ? super T2, ? extends R> zipFunction
②Observable<? extends T2> other, Func2<? super T, ? super T2, ? extends R> zipFunction