RxJava 常用操作符.doc

RxJava 常用操作符1 Observable 的创建1.1 from( )转换集合为一个每次发射集合中一个元素的 Observable 对象。可用来遍历集合。方法列表：public static Observable from(Future future)public static Observable from(Future future, long timeout, TimeUnit unit)public static Observable from(Future future, Scheduler scheduler)public static Observable from(Iterable iterable)public static Observable from(T array)栗子：/ 1. 遍历集合Observable observable = Observable.from(new Stringhello, hi);12/ 2. 使用 Future 创建 Observable，Future 表示一个异步计算的结果。FutureTask futureTask = new FutureTask(new Callable() Override public String call() throws Exception / TODO 执行异步操作并返回数据 return hihi; );Scheduler.Worker worker = Schedulers.io().createWorker();worker.schedule(new Action0() Override public void call() futureTask.run(); );Observable observable = Observable.from(futureTask);1.2 just( )转换一个或多个 Object 为依次发射这些 Object 的 Observable 对象。方法列表：public static Observable just(final T value)public static Observable just(T t1, T t2)public static Observable just(T t1, T t2, T t3)public static Observable just(T t1, T t2, T t3, T t4)public static Observable just(T t1, T t2, T t3, T t4, T t5)public static Observable just(T t1, T t2, T t3, T t4, T t5, T t6)public static Observable just(T t1, T t2, T t3, T t4, T t5, T t6, T t7)public static Observable just(T t1, T t2, T t3, T t4, T t5, T t6, T t7, T t8)public static Observable just(T t1, T t2, T t3, T t4, T t5, T t6, T t7, T t8, T t9)public static Observable just(T t1, T t2, T t3, T t4, T t5, T t6, T t7, T t8, T t9, T t10)栗子：Observable observable = Observable.just(hello);/ 使用 just() 遍历几个元素Observable observable = Observable.just(hello, hi, .);/ 使用 from() 方法遍历，效果和 just() 一样。String stringArrs = new Stringhello, hi, .;Observable observable = Observable.from(stringArrs);just() 方法可传入 110 个参数，也就说当元素个数小于等于 10 的时候既可以使用 just() 也可以使用 from()，否则只能用 from() 方法。1.3 create( )返回一个在被 OnSubscribe 订阅时执行特定方法的 Observable 对象。方法列表：public static Observable create(OnSubscribe f)Beta public static Observablecreate(SyncOnSubscribe syncOnSubscribe)Experimental public static Observable create(AsyncOnSubscribe asyncOnSubscribe)栗子：Observable.OnSubscribe onSubscribe = new Observable.OnSubscribe() Override public void call(Subscriber subscriber) / onNext() 方法可执行多次 subscribe.onNext(hello); subscribe.onCompleted(); ;Observable observable = Observable.create(onSubscribe);此方法不常用，大多数时候都是使用 just( )、form( ) 等方法，如上面那串代码就可以写成：Observable observable = Observable.just(hello);11.4 interval( )返回一个每隔指定的时间间隔就发射一个序列号的 Observable 对象，可用来做倒计时等操作。方法列表：public static Observable interval(long interval, TimeUnit unit)public static Observable interval(long interval, TimeUnit unit, Scheduler scheduler)public static Observable interval(long initialDelay, long period, TimeUnit unit)public static Observable interval(long initialDelay, long period, TimeUnit unit, Scheduler scheduler)栗子：/ 每隔 1 s 发送一个序列号，序列号从 0 开始，每次累加 1。Observable observable = Oerval(1, TimeUnit.SECONDS);121.5 timer( )创建一个在指定延迟时间后发射一条数据（ 固定值：0 ）的 Observable 对象，可用来做定时操作。方法列表：public static Observable timer(long delay, TimeUnit unit)public static Observable timer(long delay, TimeUnit unit, Scheduler scheduler)栗子：/ 定时 3 sObservable observable = Observable.timer(3, TimeUnit.SECONDS);121.6 range( )创建一个发射指定范围内的连续整数的 Observable 对象。方法列表：public static Observable range(int start, int count)public static Observable range(int start, int count, Scheduler scheduler)栗子：/ 依次发射 5、6、7Observable observable = Observable.range(5, 3);121.7 empty()创建一个不发射任何数据就发出 onCompleted() 通知的 Observable 对象。方法列表：public static Observable empty()栗子：/ 发出一个 onCompleted() 通知Observable observable = Observable.empty();121.8 error( )创建不发射任何数据就发出 onError 通知的 Observable 对象。方法列表：public static Observable error(Throwable exception)栗子：/ 发出一个 onError() 通知Observable observable = Observable.error(new Throwable(message);121.9 never()创建一个不发射任何数据和通知的 Observable 对象。方法列表：public static Observable never()栗子：Observable observable = Observable.never();11.10 defer( )在订阅的时候才会创建 Observable 对象；每一次订阅都创建一个新的 Observable 对象。方法列表：public static Observable defer(Func0Observable observableFactory)栗子：Observable observable = Observable.defer(new Func0Observable() Override public Observable call() return Observable.just(string); );2 重做2.1 repeat( )使Observable 对象在发出 onNext() 通知之后重复发射数据。重做结束才会发出 onComplete() 通知，若重做过程中出现异常则会中断并发出 onError() 通知。方法列表：public final Observable repeat()public final Observable repeat(final long count)public final Observable repeat(Scheduler scheduler)public final Observable repeat(final long count, Scheduler scheduler)栗子：Observable observable = Observable.just(string);/ 无限重复执行observable.repeat();/ 重复执行 5 次observable.repeat(5);2.2 repeatWhen( )使Observable 对象在发出 onNext() 通知之后有条件的重复发射数据。重做结束才会发出 onCompleted() 通知，若重做过程中出现异常则会中断并发出 onError() 通知。方法列表：public final Observable repeatWhen(final Func1? super Observable, ? extends Observable notificationHandler)public final Observable repeatWhen(final Func1? super Observable, ? extends Observable? notificationHandler, Scheduler scheduler)栗子：observable.repeatWhen(new Func1Observable, Observable() Override public Observable call(Observable observable) / 重复 3 次, 每次间隔 1 s return observable.zipWith(Observable.range(1, 3), new Func2() Override public Integer call(Void aVoid, Integer integer) return integer; ).flatMap(integer - Observable.timer(1, TimeUnit.SECONDS); );3 重试3.1 retry( )在执行 Observable对象的序列出现异常时，不直接发出 onError() 通知，而是重新订阅该 Observable对象，直到重做过程中未出现异常，则会发出 onNext() 和 onCompleted() 通知；若重做过程中也出现异常，则会继续重试，直到达到重试次数上限，超出次数后发出最新的 onError() 通知。方法列表：public final Observable retry()public final Observable retry(final long count)public final Observable retry(Func2 predicate)栗子：Observable observable = Observable.create(new Observable.OnSubscribe() Override public void call(Subscriber subscriber) System.out.println(.); int a = 1 / 0; subscriber.onNext(a); subscriber.onCompleted(); );/ 无限次的重试observable.retry();/ 重试 3 次observable.retry(3);/ 使用谓语函数决定是否重试observable.retry(new Func2() Override public Boolean call(Integer integer, Throwable throwable) / 参数 integer 是订阅的次数; 参数 throwable 是抛出的异常 / 返回值为 true 表示重试, 返回值为 false 表示不重试 return false; );3.2 retryWhen( )作用： 有条件的执行重试。方法列表：public final Observable retryWhen(final Func1? super Observable, ? extends Observable notificationHandler)public final Observable retryWhen(final Func1? super Observable, ? extends ObservablenotificationHandler, Scheduler scheduler)栗子：/ 重试 3 次，每次间隔 1 sobservable.retryWhen(new Func1Observable, Observable() Override public Observable call(Observable observable) return observable.zipWith(Observable.range(1, 3), new Func2() Override public Object call(Throwable throwable, Integer integer) return integer; ).flatMap(new Func1Object, Observable() Override public Observable call(Object o) return Observable.timer(1, TimeUnit.SECONDS); ); );4 变换4.1 map( )把源 Observable 发射的元素应用于指定的函数，并发送该函数的结果。方法列表：public final Observable map(Func1 func)栗子：Observable.just(2) .map(new Func1() Override public String call(Integer integer) return String.valueOf(String.format(原始数据的两倍为: %s, integer * 2); );4.2 flatMap( )转换源 Observable 对象为另一个 Observable 对象。方法列表：public final Observable flatMap(Func1? super T, ? extends Observable func)Beta public final Observable flatMap(Func1? super T, ? extends Observable func, int maxConcurrent)public final Observable flatMap(Func1? super T, ? extends Observable onNext, Func1? super Throwable, ? extends Observable onError, Func0? extends Observable onCompleted)Beta public final Observable flatMap(Func1? super T, ? extends Observable onNext, Func1? super Throwable, ? extends Observable onError, Func0? extends Observable onCompleted, int maxConcurrent)public final Observable flatMap(final Func1? super T, ? extends Observable collectionSlector, final Func2 resultSelector)Beta public final Observable flatMap(final Func1? super T, ? extends Observable collectionSelector, final Func2 resultSelector, int maxConcurrent)栗子：Observable.just(2) .flatMap(new Func1Integer, Observable() Override public Observable call(Integer integer) / 转换为一个定时 integer 秒的 Observable 对象 return Observable.timer(integer, TimeUnit.SECONDS); );5 过滤5.1 filter( )只发射满足指定谓词的元素。方法列表：public final Observable filter(Func1 predicate)栗子：Observable.just(-1, -2, 0, 1, 2) .filter(new Func1() Override public Boolean call(Integer integer) return integer 0; );5.2 first( )返回一个仅仅发射源 Observable 发射的第一个满足指定谓词的元素的 Observable，如果如果源 Observable 为空，则会抛出一个 NoSuchElementException。方法列表：public final Observable first()public final Observable first(Func1 predicate)栗子：/ 发射第一个元素Observable.just(-1, -2, 0, 1, 2).first();/ 发射满足条件的第一个元素Observable.just(-1, -2, 0, 1, 2) .first(new Func1() Override public Boolean call(Integer integer) return integer 0; );/ 会抛出 NoSuchElementException 异常Observable.empty().first();5.3 last( )返回一个仅仅发射源 Observable 发射的倒数第一个满足指定谓词的元素的 Observable，如果如果源 Observable 为空，则会抛出一个 NoSuchElementException。方法列表：public final Observable last()public final Observable last(Func1 predicate)栗子：/ 发射倒数第一个元素Observable.just(-1, -2, 0, 1, 2).first();/ 发射满足条件的倒数第一个元素Observable.just(-1, -2, 0, 1, 2) .first(new Func1() Override public Boolean call(Integer integer) return integer 0; );/ 会抛出 NoSuchElementException 异常Observable.empty().last();5.4 skip( )跳过前面指定数量或指定时间内的元素，只发射后面的元素。方法列表：public final Observable skip(int count)public final Observable skip(long time, TimeUnit unit)public final Observable skip(long time, TimeUnit unit, Scheduler scheduler)栗子：Observable.just(-1, -2, 0, 1, 2) .skip(2) / 跳过前两条数据125.5 skipLast( )跳过前面指定数量或指定时间内的元素，只发射后面的元素。指定时间时会延迟源 Observable 发射的任何数据。方法列表：public final Observable skipLast(int count)public final Observable skipLast(long time, TimeUnit unit)public final Observable skipLast(long time, TimeUnit unit, Scheduler scheduler)栗子：Observable.just(-1, -2, 0, 1, 2) .skip(2) / 跳过后两条数据125.6 take( )只发射前面指定数量或指定时间内的元素。方法列表：public final Observable take(final int count)public final Observable take(long time, TimeUnit unit)public final Observable take(long time, TimeUnit unit, Scheduler scheduler)栗子：Observable.just(-1, -2, 0, 1, 2).take(3); / 只发射前三条数据15.7 takeLast( )只发射后面指定数量或指定时间内的元素。指定时间时会延迟源 Observable 发射的任何数据。方法列表：public final Observable takeLast(final int count)public final Observable takeLast(int count, long time, TimeUnit unit)public final Observable takeLast(int count, long time, TimeUnit unit, Scheduler scheduler)public final Observable takeLast(long time, TimeUnit unit)public final Observable takeLast(long time, TimeUnit unit, Scheduler scheduler)栗子：Observable.just(-1, -2, 0, 1, 2).takeLast(3); / 只发射后三条数据15.8 sample( )定期发射 Observable 发射的最后一条数据。方法列表：public final Observable sample(long period, TimeUnit unit)public final Observable sample(long period, TimeUnit unit, Scheduler scheduler)public final Observable sample(Observable sampler)栗子：Oerval(300, TimeUnit.MILLISECONDS) .sample(2, TimeUnit.SECONDS)5.9 elementAt( )只发射指定索引的元素。方法列表：public final Observable elementAt(int index)栗子：Observable.just(-1, -2, 0, 1, 2).elementAt(2); / 发射索引为 2 的数据15.10 elementAtOrDefault( )只发射指定索引的元素，若该索引对应的元素不存在，则发射默认值。方法列表：public final Observable elementAtOrDefault(int index, T defaultValue)栗子：Observable.just(-1, -2, 0, 1, 2).elementAtOrDefault(9, -5); / 发射索引为 9的数据，若不存在，则发射 -515.11 ignoreElements( )不发射任何数据，直接发出 onCompleted() 通知。方法列表：public final Observable ignoreElements()栗子：Observable.just(-1, -2, 0, 1, 2).ignoreElements()15.12 distinct( )过滤重复的元素，过滤规则是：只允许还没有发射过的元素通过。方法列表：public final Observable distinct()public final Observable distinct(Func1 keySelector)栗子：/ 直接过滤Observable.just(-1, -2, 0, 1, 2, 1).distinct();/ 通过生成的 key 值过滤Observable.just(-1, -2, 0, 1, 2, 1).distinct(new Func1() Override public Integer call(Integer integer) / 随机生成 key return integer * (int)(Math.random() * 10); );5.13 debounce( )源 Observable 每产生结果后，如果在规定的间隔时间内没有产生新的结果，则发射这个结果，否则会忽略这个结果。该操作符会过滤掉发射速率过快的数据项。方法列表：public final Observable debounce(long timeout, TimeUnit unit)public final Observable debounce(long timeout, TimeUnit unit, Scheduler scheduler)public final Observable debounce(Func1? super T, ? extends Observable debounceSelector)栗子：Observable observable = Observable.create(new Observable.OnSubscribe() Override public vo