你一定会用到的RxJava常用操作符

作者博客

http://blog.csdn.net/u014165119

前言

阅读本文前请先了解 RxJava 的基本使用。

目录

1. Observable 的创建
   1. from()
   2. just()
   3. create()
   4. interval()
   5. timer()
   6. range()
   7. error()
   8. defer()
2. 重做
   1. repeat()
   2. repeatWhen()
3. 重试
   1. retry()
   2. retryWhen()
4. 变换
   1. map()
   2. flatMap()
5. 过滤
   1. filter()
   2. first()
   3. last()
   4. skip()
   5. skipLast()
   6. take()
   7. takeLast()
   8. sample()
   9. elementAt()
   10. elementAtOrDefault()
   11. ignoreElements()
6. 功能实现
   1. 延时遍历
   2. 倒计时

1

Observable 的创建

from()

public static <T> Observable<T> from(Iterable<? extends T> iterable);

转换集合为一个每次发射集合中一个元素的 Observable 对象。

使用场景：对集合（数组、List 等）进行遍历。

其他 form() API：

• public static <T> Observable<T> from(Future<? extends T> future);
• public static <T> Observable<T> from(Future<? extends T> future, long timeout, TimeUnit unit);
• public static <T> Observable<T> from(Future<? extends T> future, Scheduler scheduler);
• public static <T> Observable<T> from(T[] array);

举例：

just()

public static <T> Observable<T> just(final T value);

转换一个或多个 Object 为依次发射这些 Object 的 Observable 对象。

使用场景：转换一个或多个普通 Object 为 Observable 对象，如转换数据库查询结果、网络查询结果等。

其他 just() API：

• public static <T> Observable<T> just(T t1, T t2);
• public static <T> Observable<T> just(T t1, T t2, T t3);
• public static <T> Observable<T> just(T t1, T t2, T t3, T t4);
• public static <T> Observable<T> just(T t1, T t2, T t3, T t4, T t5);
• public static <T> Observable<T> just(T t1, T t2, T t3, T t4, T t5, T t6);
• public static <T> Observable<T> just(T t1, T t2, T t3, T t4, T t5, T t6, T t7);
• public static <T> Observable<T> just(T t1, T t2, T t3, T t4, T t5, T t6, T t7, T t8);
• public static <T> Observable<T> just(T t1, T t2, T t3, T t4, T t5, T t6, T t7, T t8, T t9);
• public static <T> Observable<T> just(T t1, T t2, T t3, T t4, T t5, T t6, T t7, T t8, T t9, T t10);

举例：

just() 方法可传入 1~10 个参数，也就说当元素个数小于等于 10 的时候既可以使用just() 也可以使用 from()，否则只能用 from() 方法。

create()

public static <T> Observable<T> create(OnSubscribe<T> f);

返回一个在被 OnSubscribe 订阅时执行特定方法的 Observable 对象， 使用场景：不推荐使用，可使用其他操作符替代，如使用 from()操作符完成遍历。

其他 create() API：

• @Beta public static <S, T> Observable<T>create(SyncOnSubscribe<S, T> syncOnSubscribe);
• @Experimental public static <S, T> Observable<T> create(AsyncOnSubscribe<S, T> asyncOnSubscribe);

举例：

此方法不常用，大多数时候都是用 just()、form() 等方法，如上面那串代码就可以写成：

interval()

public static Observable<Long> interval(long interval, TimeUnit unit);

返回一个每隔指定的时间间隔就发射一个序列号的 Observable 对象。 使用场景：可使用该操作符完成定时、倒计时等功能。

其他 interval() API：

• public static Observable<Long> interval(long interval, TimeUnit unit, Scheduler scheduler);
• public static Observable<Long> interval(long initialDelay, long period, TimeUnit unit);
• public static Observable<Long> interval(long initialDelay, long period, TimeUnit unit, Scheduler scheduler);

举例：

timer()

public static Observable<Long> timer(long delay, TimeUnit unit);

创建一个在指定延迟时间后发射一条数据（固定值：0）的 Observable 对象。 使用场景：可用来完成定时功能。

其他 timer() API：

• public static Observable<Long> timer(long delay, TimeUnit unit, Scheduler scheduler);

举例：

range()

public static Observable<Integer> range(int start, int count);

创建一个发射指定范围内的连续整数的 Observable 对象。 使用场景：可使用该操作符完成一个 fori 的循环，如 for(int i=5;i<=7;i++) -> Observable.range(5, 3)。

其他 range() API：

• public static Observable<Integer> range(int start, int count, Scheduler scheduler);

举例：

error()

public static <T> Observable<T> error(Throwable exception);

创建不发射任何数据就发出 onError 通知的 Observable 对象。 使用场景：程序中捕获异常后，可使用该操作符把捕获的异常传递到后面的逻辑中处理。

举例：

defer()

public static <T> Observable<T> defer(Func0<Observable<T>> observableFactory);

在订阅的时候才会创建 Observable 对象；每一次订阅都创建一个新的 Observable 对象。 使用场景：可以使用该操作符封装需要被多次执行的函数。

举例：

2

重做

repeat()

public final Observable<T> repeat();

使Observable 对象在发出 onNext() 通知之后重复发射数据。重做结束才会发出 onComplete() 通知，若重做过程中出现异常则会中断并发出 onError() 通知。 使用场景：可使用该操作符指定一次任务执行完成后立即重复执行上一次的任务，如发送多次网络请求等。

其他 repeat() API：

• public final Observable<T> repeat(final long count);
• public final Observable<T> repeat(Scheduler scheduler);
• public final Observable<T> repeat(final long count, Scheduler scheduler);

举例：

repeatWhen()

public final Observable<T> repeatWhen(final Func1<? super Observable<? extends Void>, ? extends Observable<?>> notificationHandler)

使Observable 对象在发出 onNext() 通知之后有条件的重复发射数据。重做结束才会发出 onCompleted() 通知，若重做过程中出现异常则会中断并发出 onError() 通知。 使用场景：可使用该操作符指定满足一定条件时重复执行一个任务，如发送多次网络请求等。

其他 repeatWhen() API：

• public final Observable<T> repeatWhen(final Func1<? super Observable<? extends Void>, ? extends Observable>?> notificationHandler, Scheduler scheduler);

举例：

3

重试

retry()

public final Observable<T> retry();

在执行 Observable对象的序列出现异常时，不直接发出 onError() 通知，而是重新订阅该 Observable对象，直到重做过程中未出现异常，则会发出 onNext() 和 onCompleted()通知；若重做过程中也出现异常，则会继续重试，直到达到重试次数上限，超出次数后发出最新的 onError() 通知。

使用场景：网络等请求异常出错后，可重新发起请求。

其他 retry() API：

• public final Observable<T> retry(final long count);
• public final Observable<T> retry(Func2<Integer, Throwable, Boolean> predicate);

举例：

retryWhen()

public final Observable<T> retryWhen(final Func1<? super Observable<? extends Throwable>, ? extends Observable<?>> notificationHandler);

有条件的执行重试。 使用场景：网络等请求异常出错后，若满足一定条件，则重新发起请求。

其他 retryWhen() API：

• public final Observable<T> retryWhen(final Func1<? super Observable<? extends Throwable>, ? extends Observable<?>>notificationHandler, Scheduler scheduler);

举例：

4

变换

map()

public final <R> Observable<R> map(Func1<? super T, ? extends R> func);

把源 Observable 发射的元素应用于指定的函数，并发送该函数的结果。

使用场景：将从网络获取的数据（NetData 对象）转换为数据库相关对象（DBData对象）并使用 Observable 发送。

举例：

flatMap()

public final <R> Observable<R> flatMap(Func1<? super T, ? extends Observable<? extends R>> func);

转换源 Observable 对象为另一个 Observable 对象。

使用场景：从网络获取数据并使用 obsA 对象发射，flatMap() 操作符中可将数据存进数据库并返回一个新的对象 obsB。

其他 flatMap() API：

• @Beta public final <R> Observable<R> flatMap(Func1<? super T, ? extends Observable<? extends R>> func, int maxConcurrent);
• public final <R> Observable<R> flatMap(Func1<? super T, ? extends Observable<? extends R>> onNext, Func1<? super Throwable, ? extends Observable<? extends R>> onError, Func0<? extends Observable<? extends R>> onCompleted);
• @Beta public final <R> Observable<R> flatMap(Func1<? super T, ? extends Observable<? extends R>> onNext, Func1<? super Throwable, ? extends Observable<? extends R>> onError, Func0<? extends Observable<? extends R>> onCompleted, int maxConcurrent);
• public final <U, R> Observable<R> flatMap(final Func1<? super T, ? extends Observable<? extends U>> collectionSelector, final Func2<? super T, ? super U, ? extends R> resultSelector);
• @Beta public final <U, R> Observable<R> flatMap(final Func1<? super T, ? extends Observable<? extends U>> collectionSelector, final Func2<? super T, ? super U, ? extends R> resultSelector, int maxConcurrent);

举例：

5

过滤

filter()

public final Observable<T> filter(Func1<? super T, Boolean> predicate);

只发射满足指定谓词的元素。 使用场景：可使用 filter 代替 if 语句。

举例：

first()

public final Observable<T> first();

返回一个仅仅发射源 Observable 发射的第一个［满足指定谓词的］元素的 Observable，如果源 Observable 为空，则会抛出一个 NoSuchElementException。 使用场景： 顺序发出多条数据，只接收第一条。

其他 first() API：

• public final Observable<T> first(Func1<? super T, Boolean> predicate)

举例：

last()

public final Observable<T> last();

返回一个仅仅发射源 Observable 发射的倒数第一个［满足指定谓词的］元素的 Observable，如果源 Observable 为空，则会抛出一个 NoSuchElementException。 使用场景： 顺序发出多条数据，只接收最后一条。

其他 last() API：

• public final Observable<T> last(Func1<? super T, Boolean> predicate)

举例：

skip()

public final Observable<T> skip(int count);

跳过前面指定数量或指定时间内的元素，只发射后面的元素。

其他 skip() API：

• public final Observable<T> skip(long time, TimeUnit unit);
• public final Observable<T> skip(long time, TimeUnit unit, Scheduler scheduler);

举例：

skipLast()

public final Observable<T> skipLast(int count);

跳过前面指定数量或指定时间内的元素，只发射后面的元素。指定时间时会延迟源 Observable 发射的任何数据。

其他 skipLast() API：

• public final Observable<T> skipLast(long time, TimeUnit unit);
• public final Observable<T> skipLast(long time, TimeUnit unit, Scheduler scheduler);

举例：

take()

public final Observable<T> take(final int count);

只发射前面指定数量或指定时间内的元素。

其他 take() API：

• public final Observable<T> take(long time, TimeUnit unit);
• public final Observable<T> take(long time, TimeUnit unit, Scheduler scheduler);

举例：

takeLast()

public final Observable<T> takeLast(final int count);

只发射后面指定数量或指定时间内的元素。指定时间时会延迟源 Observable 发射的任何数据。

其他 takeLast() API：

• public final Observable<T> takeLast(int count, long time, TimeUnit unit);
• public final Observable<T> takeLast(int count, long time, TimeUnit unit, Scheduler scheduler);
• public final Observable<T> takeLast(long time, TimeUnit unit);
• public final Observable<T> takeLast(long time, TimeUnit unit, Scheduler scheduler);

举例：

sample()

public final Observable<T> sample(long period, TimeUnit unit);

定期发射 Observable 发射的最后一条数据。

其他 sample() API：

• public final Observable<T> sample(long period, TimeUnit unit, Scheduler scheduler);
• public final <U> Observable<T> sample(Observable<U> sampler);

举例：

elementAt()

public final Observable<T> elementAt(int index);

只发射指定索引的元素。 使用场景： 按索引去集合中的元素等。

举例：

elementAtOrDefault()

public final Observable<T> elementAtOrDefault(int index, T defaultValue);

只发射指定索引的元素，若该索引对应的元素不存在，则发射默认值。

举例：

// 发射索引为 9的数据，若不存在，则发射 -5

ignoreElements()

public final Observable<T> ignoreElements();

不发射任何数据，直接发出 onCompleted() 通知。

举例：

6

功能实现

延时遍历

倒计时