RxJava常用操作符

1、RxJava 常用操作符1 Observable 的創(chuàng)建1.1 from( )轉(zhuǎn)換集合為一個每次發(fā)射集合中一個元素的 Observable 對象?？捎脕肀闅v集合。方法列表：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&g

2、t; Observable<T> from(Future<? extends T> future, Scheduler scheduler)public static <T> Observable<T> from(Iterable<? extends T> iterable)public static <T> Observable<T> from(T array)栗子：/ 1. 遍歷集合Observable<String> observable = Observable.from(new Strin

3、g"hello", "hi");12/ 2. 使用 Future 創(chuàng)建 Observable，F(xiàn)uture 表示一個異步計算的結(jié)果。FutureTask<String> futureTask = new FutureTask<String>(new Callable<String>() Override public String call() throws Exception / TODO 執(zhí)行異步操作并返回數(shù)據(jù) return "hihi" );Scheduler.Worker worker =

4、Schedulers.io().createWorker();worker.schedule(new Action0() Override public void call() futureTask.run(); );Observable<String> observable = Observable.from(futureTask);1.2 just( )轉(zhuǎn)換一個或多個 Object 為依次發(fā)射這些 Object 的 Observable 對象。方法列表：public static <T> Observable<T> just(final T value)

5、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>

6、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,

7、 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)栗子：Observable<String> observable = Observable.just("hello");/ 使用 just() 遍歷幾個元素Observable<String> observable = Observable.

8、just("hello", "hi", ".");/ 使用 from() 方法遍歷，效果和 just() 一樣。String stringArrs = new String"hello", "hi", "."Observable<String> observable = Observable.from(stringArrs);just() 方法可傳入 110 個參數(shù)，也就說當(dāng)元素個數(shù)小于等于 10 的時候既可以使用 just() 也可以使用 from()，否則只能

9、用 from() 方法。1.3 create( )返回一個在被 OnSubscribe 訂閱時執(zhí)行特定方法的 Observable 對象。方法列表：public static <T> Observable<T> create(OnSubscribe<T> f)Beta public static <S, T> Observable<T>create(SyncOnSubscribe<S, T> syncOnSubscribe)Experimental public static <S, T> Observable

10、<T> create(AsyncOnSubscribe<S, T> asyncOnSubscribe)栗子：Observable.OnSubscribe<String> onSubscribe = new Observable.OnSubscribe< String >() Override public void call(Subscriber<? super String > subscriber) / onNext() 方法可執(zhí)行多次 subscribe.onNext("hello"); subscribe.

11、onCompleted(); ;Observable<Object> observable = Observable.create(onSubscribe);此方法不常用，大多數(shù)時候都是使用 just( )、form( ) 等方法，如上面那串代碼就可以寫成：Observable<Object> observable = Observable.just("hello");11.4 interval( )返回一個每隔指定的時間間隔就發(fā)射一個序列號的 Observable 對象，可用來做倒計時等操作。方法列表：public static Observabl

12、e<Long> interval(long interval, TimeUnit unit)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 initialDe

13、lay, long period, TimeUnit unit, Scheduler scheduler)栗子：/ 每隔 1 s 發(fā)送一個序列號，序列號從 0 開始，每次累加 1。Observable<Long> observable = Oerval(1, TimeUnit.SECONDS);121.5 timer( )創(chuàng)建一個在指定延遲時間后發(fā)射一條數(shù)據(jù)（ 固定值：0 ）的 Observable 對象，可用來做定時操作。方法列表：public static Observable<Long> timer(long delay, TimeUni

14、t unit)public static Observable<Long> timer(long delay, TimeUnit unit, Scheduler scheduler)栗子：/ 定時 3 sObservable<Long> observable = Observable.timer(3, TimeUnit.SECONDS);121.6 range( )創(chuàng)建一個發(fā)射指定范圍內(nèi)的連續(xù)整數(shù)的 Observable 對象。方法列表：public static Observable<Integer> range(int start, int count)

15、public static Observable<Integer> range(int start, int count, Scheduler scheduler)栗子：/ 依次發(fā)射 5、6、7Observable<Integer> observable = Observable.range(5, 3);121.7 empty()創(chuàng)建一個不發(fā)射任何數(shù)據(jù)就發(fā)出 onCompleted() 通知的 Observable 對象。方法列表：public static <T> Observable<T> empty()栗子：/ 發(fā)出一個 onComplet

16、ed() 通知Observable<Object> observable = Observable.empty();121.8 error( )創(chuàng)建不發(fā)射任何數(shù)據(jù)就發(fā)出 onError 通知的 Observable 對象。方法列表：public static <T> Observable<T> error(Throwable exception)栗子：/ 發(fā)出一個 onError() 通知Observable<Object> observable = Observable.error(new Throwable("message&quo

17、t;);121.9 never()創(chuàng)建一個不發(fā)射任何數(shù)據(jù)和通知的 Observable 對象。方法列表：public static <T> Observable<T> never()栗子：Observable<Object> observable = Observable.never();11.10 defer( )在訂閱的時候才會創(chuàng)建 Observable 對象；每一次訂閱都創(chuàng)建一個新的 Observable 對象。方法列表：public static <T> Observable<T> defer(Func0<Observa

18、ble<T>> observableFactory)栗子：Observable<String> observable = Observable.defer(new Func0<Observable<String>>() Override public Observable<String> call() return Observable.just("string"); );2 重做2.1 repeat( )使Observable 對象在發(fā)出 onNext() 通知之后重復(fù)發(fā)射數(shù)據(jù)。重做結(jié)束才會發(fā)出 onCom

19、plete() 通知，若重做過程中出現(xiàn)異常則會中斷并發(fā)出 onError() 通知。方法列表：public final Observable<T> repeat()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)栗子：Observable<

20、String> observable = Observable.just("string");/ 無限重復(fù)執(zhí)行observable.repeat();/ 重復(fù)執(zhí)行 5 次observable.repeat(5);2.2 repeatWhen( )使Observable 對象在發(fā)出 onNext() 通知之后有條件的重復(fù)發(fā)射數(shù)據(jù)。重做結(jié)束才會發(fā)出 onCompleted() 通知，若重做過程中出現(xiàn)異常則會中斷并發(fā)出 onError() 通知。方法列表：public final Observable<T> repeatWhen(final Func1<?

21、 super Observable<? extends Void>, ? extends Observable<?>> notificationHandler)public final Observable<T> repeatWhen(final Func1<? super Observable<? extends Void>, ? extends Observable>?> notificationHandler, Scheduler scheduler)栗子：observable.repeatWhen(new Func

22、1<Observable<? extends Void>, Observable<?>>() Override public Observable<?> call(Observable<? extends Void> observable) / 重復(fù) 3 次, 每次間隔 1 s return observable.zipWith(Observable.range(1, 3), new Func2<Void, Integer, Integer>() Override public Integer call(Void aVoi

23、d, Integer integer) return integer; ).flatMap(integer -> Observable.timer(1, TimeUnit.SECONDS); );3 重試3.1 retry( )在執(zhí)行 Observable對象的序列出現(xiàn)異常時，不直接發(fā)出 onError() 通知，而是重新訂閱該 Observable對象，直到重做過程中未出現(xiàn)異常，則會發(fā)出 onNext() 和 onCompleted() 通知；若重做過程中也出現(xiàn)異常，則會繼續(xù)重試，直到達(dá)到重試次數(shù)上限，超出次數(shù)后發(fā)出最新的 onError() 通知。方法列表：public final

24、Observable<T> retry()public final Observable<T> retry(final long count)public final Observable<T> retry(Func2<Integer, Throwable, Boolean> predicate)栗子：Observable<Integer> observable = Observable.create(new Observable.OnSubscribe<Integer>() Override public void ca

25、ll(Subscriber<? super Integer> subscriber) System.out.println("."); int a = 1 / 0; subscriber.onNext(a); subscriber.onCompleted(); );/ 無限次的重試observable.retry();/ 重試 3 次observable.retry(3);/ 使用謂語函數(shù)決定是否重試observable.retry(new Func2<Integer, Throwable, Boolean>() Override public Bo

26、olean call(Integer integer, Throwable throwable) / 參數(shù) integer 是訂閱的次數(shù); 參數(shù) throwable 是拋出的異常 / 返回值為 true 表示重試, 返回值為 false 表示不重試 return false; );3.2 retryWhen( )作用： 有條件的執(zhí)行重試。方法列表：public final Observable<T> retryWhen(final Func1<? super Observable<? extends Throwable>, ? extends Observable

27、<?>> notificationHandler)public final Observable<T> retryWhen(final Func1<? super Observable<? extends Throwable>, ? extends Observable<?>>notificationHandler, Scheduler scheduler)栗子：/ 重試 3 次，每次間隔 1 sobservable.retryWhen(new Func1<Observable<? extends Throwable

28、>, Observable<?>>() Override public Observable<?> call(Observable<? extends Throwable> observable) return observable.zipWith(Observable.range(1, 3), new Func2<Throwable, Integer, Object>() Override public Object call(Throwable throwable, Integer integer) return integer;

29、 ).flatMap(new Func1<Object, Observable<?>>() Override public Observable<?> call(Object o) return Observable.timer(1, TimeUnit.SECONDS); ); );4 變換4.1 map( )把源 Observable 發(fā)射的元素應(yīng)用于指定的函數(shù)，并發(fā)送該函數(shù)的結(jié)果。方法列表：public final <R> Observable<R> map(Func1<? super T, ? extends R>

30、func)栗子：Observable.just(2) .map(new Func1<Integer, String>() Override public String call(Integer integer) return String.valueOf(String.format("原始數(shù)據(jù)的兩倍為: %s", integer * 2); );4.2 flatMap( )轉(zhuǎn)換源 Observable 對象為另一個 Observable 對象。方法列表：public final <R> Observable<R> flatMap(Func

31、1<? super T, ? extends Observable<? extends R>> func)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

32、<? 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>> on

33、Next, 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>> colle

34、ctionSlector, 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,

35、int maxConcurrent)栗子：Observable.just(2) .flatMap(new Func1<Integer, Observable<Long>>() Override public Observable<Long> call(Integer integer) / 轉(zhuǎn)換為一個定時 integer 秒的 Observable 對象 return Observable.timer(integer, TimeUnit.SECONDS); );5 過濾5.1 filter( )只發(fā)射滿足指定謂詞的元素。方法列表：public final Ob

36、servable<T> filter(Func1<? super T, Boolean> predicate)栗子：Observable.just(-1, -2, 0, 1, 2) .filter(new Func1<Integer, Boolean>() Override public Boolean call(Integer integer) return integer > 0; );5.2 first( )返回一個僅僅發(fā)射源 Observable 發(fā)射的第一個滿足指定謂詞的元素的 Observable，如果如果源 Observable 為空，則

37、會拋出一個 NoSuchElementException。方法列表：public final Observable<T> first()public final Observable<T> first(Func1<? super T, Boolean> predicate)栗子：/ 發(fā)射第一個元素Observable.just(-1, -2, 0, 1, 2).first();/ 發(fā)射滿足條件的第一個元素Observable.just(-1, -2, 0, 1, 2) .first(new Func1<Integer, Boolean>() Ov

38、erride public Boolean call(Integer integer) return integer > 0; );/ 會拋出 NoSuchElementException 異常Observable.empty().first();5.3 last( )返回一個僅僅發(fā)射源 Observable 發(fā)射的倒數(shù)第一個滿足指定謂詞的元素的 Observable，如果如果源 Observable 為空，則會拋出一個 NoSuchElementException。方法列表：public final Observable<T> last()public final Obse

39、rvable<T> last(Func1<? super T, Boolean> predicate)栗子：/ 發(fā)射倒數(shù)第一個元素Observable.just(-1, -2, 0, 1, 2).first();/ 發(fā)射滿足條件的倒數(shù)第一個元素Observable.just(-1, -2, 0, 1, 2) .first(new Func1<Integer, Boolean>() Override public Boolean call(Integer integer) return integer < 0; );/ 會拋出 NoSuchElement

40、Exception 異常Observable.empty().last();5.4 skip( )跳過前面指定數(shù)量或指定時間內(nèi)的元素，只發(fā)射后面的元素。方法列表：public final Observable<T> skip(int count)public final Observable<T> skip(long time, TimeUnit unit)public final Observable<T> skip(long time, TimeUnit unit, Scheduler scheduler)栗子：Observable.just(-1, -

41、2, 0, 1, 2) .skip(2) / 跳過前兩條數(shù)據(jù)125.5 skipLast( )跳過前面指定數(shù)量或指定時間內(nèi)的元素，只發(fā)射后面的元素。指定時間時會延遲源 Observable 發(fā)射的任何數(shù)據(jù)。方法列表：public final Observable<T> skipLast(int count)public final Observable<T> skipLast(long time, TimeUnit unit)public final Observable<T> skipLast(long time, TimeUnit unit, Sched

42、uler scheduler)栗子：Observable.just(-1, -2, 0, 1, 2) .skip(2) / 跳過后兩條數(shù)據(jù)125.6 take( )只發(fā)射前面指定數(shù)量或指定時間內(nèi)的元素。方法列表：public final Observable<T> take(final int count)public final Observable<T> take(long time, TimeUnit unit)public final Observable<T> take(long time, TimeUnit unit, Scheduler sch

43、eduler)栗子：Observable.just(-1, -2, 0, 1, 2).take(3); / 只發(fā)射前三條數(shù)據(jù)15.7 takeLast( )只發(fā)射后面指定數(shù)量或指定時間內(nèi)的元素。指定時間時會延遲源 Observable 發(fā)射的任何數(shù)據(jù)。方法列表：public final Observable<T> takeLast(final int count)public final Observable<T> takeLast(int count, long time, TimeUnit unit)public final Observable<T>

44、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)栗子：Observable.just(-1, -2, 0, 1, 2).takeLast(3); / 只發(fā)射后三條數(shù)據(jù)15.8 sample( )定期發(fā)射 O

45、bservable 發(fā)射的最后一條數(shù)據(jù)。方法列表：public final Observable<T> sample(long period, TimeUnit unit)public final Observable<T> sample(long period, TimeUnit unit, Scheduler scheduler)public final <U> Observable<T> sample(Observable<U> sampler)栗子：Oerval(300, TimeUnit.MILLI

46、SECONDS) .sample(2, TimeUnit.SECONDS)5.9 elementAt( )只發(fā)射指定索引的元素。方法列表：public final Observable<T> elementAt(int index)栗子：Observable.just(-1, -2, 0, 1, 2).elementAt(2); / 發(fā)射索引為 2 的數(shù)據(jù)15.10 elementAtOrDefault( )只發(fā)射指定索引的元素，若該索引對應(yīng)的元素不存在，則發(fā)射默認(rèn)值。方法列表：public final Observable<T> elementAtOrDefault(

47、int index, T defaultValue)栗子：Observable.just(-1, -2, 0, 1, 2).elementAtOrDefault(9, -5); / 發(fā)射索引為 9的數(shù)據(jù)，若不存在，則發(fā)射 -515.11 ignoreElements( )不發(fā)射任何數(shù)據(jù)，直接發(fā)出 onCompleted() 通知。方法列表：public final Observable<T> ignoreElements()栗子：Observable.just(-1, -2, 0, 1, 2).ignoreElements()15.12 distinct( )過濾重復(fù)的元素，過濾規(guī)

48、則是：只允許還沒有發(fā)射過的元素通過。方法列表：public final Observable<T> distinct()public final <U> Observable<T> distinct(Func1<? super T, ? extends U> keySelector)栗子：/ 直接過濾Observable.just(-1, -2, 0, 1, 2, 1).distinct();/ 通過生成的 key 值過濾Observable.just(-1, -2, 0, 1, 2, 1).distinct(new Func1<Integ

49、er, Integer>() Override public Integer call(Integer integer) / 隨機(jī)生成 key return integer * (int)(Math.random() * 10); );5.13 debounce( )源 Observable 每產(chǎn)生結(jié)果后，如果在規(guī)定的間隔時間內(nèi)沒有產(chǎn)生新的結(jié)果，則發(fā)射這個結(jié)果，否則會忽略這個結(jié)果。該操作符會過濾掉發(fā)射速率過快的數(shù)據(jù)項。方法列表：public final Observable<T> debounce(long timeout, TimeUnit unit)public final Observable<T> debounce(long timeout, TimeUnit unit, Scheduler scheduler)public final <U> Observable<T> debounce(Func1<? super T, ? extends Observable<U>> debounceSelector)栗子：Observable&