前言：對Rx不了解的朋友可以先看我的第一篇博文? RxJava 和 RxAndroid 一 (基礎(chǔ))，是對Rxjava的基本介紹

?

1、merge操作符，合并觀察對象

19 List<String> list1 = new ArrayList<>() ; 20 List<String> list2 = new ArrayList<>() ; 21 22 list1.add( "1" ) ; 23 list1.add( "2" ) ; 24 list1.add( "3" ) ; 25 26 list2.add( "a" ) ; 27 list2.add( "b" ) ; 28 list2.add( "c" ) ; 29 30 Observable observable1 = Observable.from( list1 ) ; 31 Observable observable2 = Observable.from( list2 ) ; 32 33 //合并數(shù)據(jù) 先發(fā)送observable2的全部數(shù)據(jù)，然后發(fā)送 observable1的全部數(shù)據(jù) 34 Observable observable = Observable.merge( observable2 , observable1 ) ; 35 36 observable.subscribe(new Action1() { 37 @Override 38 public void call(Object o) { 39 System.out.println( "rx-- " + o ); 40 } 41 }) ; 42

運行結(jié)果

2、zip??操作符，合并多個觀察對象的數(shù)據(jù)。并且允許?Func2（）函數(shù)重新發(fā)送合并后的數(shù)據(jù)

List<String> list1 = new ArrayList<>() ;List<String> list2 = new ArrayList<>() ;list1.add( "1" ) ;list1.add( "2" ) ;list1.add( "3" ) ;list2.add( "a" ) ;list2.add( "b" ) ;list2.add( "c" ) ;list2.add( "d" ) ;Observable observable1 = Observable.from( list1 ) ;Observable observable2 = Observable.from( list2 ) ;Observable observable3 = Observable.zip(observable1, observable2, new Func2<String , String , String >() {@Overridepublic String call(String s1 , String s2 ) {return s1 + s2 ;}}) ;observable3.subscribe(new Action1() {@Overridepublic void call(Object o) {System.out.println( "zip-- " + o );}}) ;

　　運行效果：從效果圖上可以看出，合并兩個的觀察對象數(shù)據(jù)項應(yīng)該是相等的；如果出現(xiàn)了數(shù)據(jù)項不等的情況，合并的數(shù)據(jù)項以最小數(shù)據(jù)隊列為準。

?

3、scan累加器操作符的使用

Observable observable = Observable.just( 1 , 2 , 3 , 4 , 5 ) ;observable.scan(new Func2<Integer,Integer,Integer>() {@Overridepublic Integer call(Integer o, Integer o2) {return o + o2 ;}}).subscribe(new Action1() {@Overridepublic void call(Object o) {System.out.println( "scan-- " + o );}}) ;

　　運行效果：

? ? ? ??第一次發(fā)射得到1，作為結(jié)果與2相加；發(fā)射得到3，作為結(jié)果與3相加，以此類推，打印結(jié)果：

? ? ? ?

?

4、filter 過濾操作符的使用

Observable observable = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;observable.filter(new Func1<Integer , Boolean>() {@Overridepublic Boolean call(Integer o) {//數(shù)據(jù)大于4的時候才會被發(fā)送return o > 4 ;}}).subscribe(new Action1() {@Overridepublic void call(Object o) {System.out.println( "filter-- " + o );}}) ;

　　運行效果

?

?

5、 消息數(shù)量過濾操作符的使用? ??

• take ：取前n個數(shù)據(jù)
• takeLast：取后n個數(shù)據(jù)
• first 只發(fā)送第一個數(shù)據(jù)
• last 只發(fā)送最后一個數(shù)據(jù)
• skip() 跳過前n個數(shù)據(jù)發(fā)送后面的數(shù)據(jù)
• skipLast() 跳過最后n個數(shù)據(jù)，發(fā)送前面的數(shù)據(jù)

//take 發(fā)送前3個數(shù)據(jù)Observable observable = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;observable.take( 3 ).subscribe(new Action1() {@Overridepublic void call(Object o) {System.out.println( "take-- " + o );}}) ;//takeLast 發(fā)送最后三個數(shù)據(jù)Observable observable2 = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;observable2.takeLast( 3 ).subscribe(new Action1() {@Overridepublic void call(Object o) {System.out.println( "takeLast-- " + o );}}) ;//first 只發(fā)送第一個數(shù)據(jù)Observable observable3 = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;observable3.first().subscribe(new Action1() {@Overridepublic void call(Object o) {System.out.println( "first-- " + o );}}) ;//last 只發(fā)送最后一個數(shù)據(jù)Observable observable4 = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;observable4.last().subscribe(new Action1() {@Overridepublic void call(Object o) {System.out.println( "last-- " + o );}}) ;//skip() 跳過前2個數(shù)據(jù)發(fā)送后面的數(shù)據(jù)Observable observable5 = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;observable5.skip( 2 ).subscribe(new Action1() {@Overridepublic void call(Object o) {System.out.println( "skip-- " + o );}}) ;//skipLast() 跳過最后兩個數(shù)據(jù)，發(fā)送前面的數(shù)據(jù)Observable observable6 = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;observable5.skipLast( 2 ).subscribe(new Action1() {@Overridepublic void call(Object o) {System.out.println( "skipLast-- " + o );}}) ; 效果圖

?

?6、elementAt 、elementAtOrDefault

//elementAt() 發(fā)送數(shù)據(jù)序列中第n個數(shù)據(jù) ，序列號從0開始//如果該序號大于數(shù)據(jù)序列中的最大序列號，則會拋出異常，程序崩潰//所以在用elementAt操作符的時候，要注意判斷發(fā)送的數(shù)據(jù)序列號是否越界Observable observable7 = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;observable7.elementAt( 3 ).subscribe(new Action1() {@Overridepublic void call(Object o) {System.out.println( "elementAt-- " + o );}}) ;//elementAtOrDefault( int n , Object default ) 發(fā)送數(shù)據(jù)序列中第n個數(shù)據(jù) ，序列號從0開始。//如果序列中沒有該序列號，則發(fā)送默認值Observable observable9 = Observable.just( 1 , 2 , 3 , 4 , 5 ) ;observable9.elementAtOrDefault( 8 , 666 ).subscribe(new Action1() {@Overridepublic void call(Object o) {System.out.println( "elementAtOrDefault-- " + o );}}) ;

　　運行結(jié)果

?

?7、startWith() 插入數(shù)據(jù)

//插入普通數(shù)據(jù)//startWith 數(shù)據(jù)序列的開頭插入一條指定的項 , 最多插入9條數(shù)據(jù)Observable observable = Observable.just( "aa" , "bb" , "cc" ) ;observable.startWith( "11" , "22" ).subscribe(new Action1() {@Overridepublic void call(Object o) {System.out.println( "startWith-- " + o );}}) ;//插入Observable對象List<String> list = new ArrayList<>() ;list.add( "ww" ) ;list.add( "tt" ) ;observable.startWith( Observable.from( list )).subscribe(new Action1() {@Overridepublic void call(Object o) {System.out.println( "startWith2 -- " + o );}}) ;

　　運行結(jié)果

?

?8、delay操作符，延遲數(shù)據(jù)發(fā)送

Observable<String> observable = Observable.just( "1" , "2" , "3" , "4" , "5" , "6" , "7" , "8" ) ;//延遲數(shù)據(jù)發(fā)射的時間，僅僅延時一次，也就是發(fā)射第一個數(shù)據(jù)前延時。發(fā)射后面的數(shù)據(jù)不延時observable.delay( 3 , TimeUnit.SECONDS ) //延遲3秒鐘.subscribe(new Action1() {@Overridepublic void call(Object o) {System.out.println("delay-- " + o);}}) ;

　　

9、Timer ?延時操作符的使用

? ? ? 使用場景：xx秒后，執(zhí)行xx? ? ??

//5秒后輸出 hello world , 然后顯示一張圖片Observable.timer( 5 , TimeUnit.SECONDS ).observeOn(AndroidSchedulers.mainThread() ).subscribe(new Action1<Long>() {@Overridepublic void call(Long aLong) {System.out.println( "timer--hello world " + aLong );findViewById( R.id.image).setVisibility(View.VISIBLE );}}) ;

? ? ?timer 返回一個?Observable , 它在延遲一段給定的時間后發(fā)射一個簡單的數(shù)字0

? ? timer 操作符默認在computation調(diào)度器上執(zhí)行，當然也可以用 Scheduler 在定義執(zhí)行的線程。

?

?delay 、timer 總結(jié)：　

• ?相同點：delay 、 timer 都是延時操作符。
• ?不同點：delay ?延時一次，延時完成后，可以連續(xù)發(fā)射多個數(shù)據(jù)。timer延時一次，延時完成后，只發(fā)射一次數(shù)據(jù)。

? ? ? ? ? ? ? ? ? ? ? ?? ? ? ? ? ? ? ? ? ? ??

?10、interval 輪詢操作符，循環(huán)發(fā)送數(shù)據(jù)，數(shù)據(jù)從0開始遞增

package app.com.myapplication; import android.support.v7.app.AppCompatActivity; import android.os.Bundle; import java.util.concurrent.TimeUnit; import rx.Observable; import rx.Subscription; import rx.functions.Action1;public class IntervalActivity extends AppCompatActivity {Subscription subscription ;@Overrideprotected void onCreate(Bundle savedInstanceState) {super.onCreate(savedInstanceState);setContentView(R.layout.activity_interval);//參數(shù)一：延遲時間 參數(shù)二：間隔時間 參數(shù)三：時間顆粒度Observable observable = Observable.interval(3000, 3000, TimeUnit.MILLISECONDS) ;subscription = observable.subscribe(new Action1() {@Overridepublic void call(Object o) {System.out.println( "interval- " + o );}}) ;}@Overrideprotected void onDestroy() {super.onDestroy();if ( subscription != null ){subscription.unsubscribe();}} }

　　

?

?11、doOnNext() 操作符，在每次 OnNext() 方法被調(diào)用前執(zhí)行

? ? ? ?使用場景：從網(wǎng)絡(luò)請求數(shù)據(jù)，在數(shù)據(jù)被展示前，緩存到本地

Observable observable = Observable.just( "1" , "2" , "3" , "4" ) ;
observable.doOnNext(new Action1() {@Overridepublic void call(Object o) {System.out.println( "doOnNext--緩存數(shù)據(jù)" + o );}}).subscribe(new Observer() {@Overridepublic void onCompleted() {}@Overridepublic void onError(Throwable e) {}@Overridepublic void onNext(Object o) {System.out.println( "onNext--" + o );}}) ;

　　

?

?12、Buffer 操作符

• Buffer( int n ) ? ? ?把n個數(shù)據(jù)打成一個list包，然后再次發(fā)送。
• Buffer( int n , int skip) ??把n個數(shù)據(jù)打成一個list包，然后跳過第skip個數(shù)據(jù)。

? ? ? ? ?

? ? 使用場景：一個按鈕每點擊3次，彈出一個toast? ? ??

List<String> list = new ArrayList<>();for (int i = 1; i < 10; i++) {list.add("" + i);}Observable<String> observable = Observable.from(list);observable.buffer(2) //把每兩個數(shù)據(jù)為一組打成一個包，然后發(fā)送.subscribe(new Action1<List<String>>() {@Overridepublic void call(List<String> strings) {System.out.println( "buffer---------------" );Observable.from( strings ).subscribe(new Action1<String>() {@Overridepublic void call(String s) {System.out.println( "buffer data --" + s );}}) ;}});

　　

? ? ?例子2：?

//第1、2 個數(shù)據(jù)打成一個數(shù)據(jù)包，跳過第三個數(shù)據(jù) ； 第4、5個數(shù)據(jù)打成一個包，跳過第6個數(shù)據(jù)observable.buffer( 2 , 3 ) //把每兩個數(shù)據(jù)為一組打成一個包，然后發(fā)送。第三個數(shù)據(jù)跳過去.subscribe(new Action1<List<String>>() {@Overridepublic void call(List<String> strings) {System.out.println( "buffer22---------------" );Observable.from( strings ).subscribe(new Action1<String>() {@Overridepublic void call(String s) {System.out.println( "buffer22 data --" + s );}}) ;}}) ;

　　

?

?13、throttleFirst 操作符

? ? ? ? ?在一段時間內(nèi)，只取第一個事件，然后其他事件都丟棄。

? ? ? ? ?使用場景：1、button按鈕防抖操作，防連續(xù)點擊 ? 2、百度關(guān)鍵詞聯(lián)想，在一段時間內(nèi)只聯(lián)想一次，防止頻繁請求服務(wù)器? ?

Observable.interval( 1 , TimeUnit.SECONDS).throttleFirst( 3 , TimeUnit.SECONDS ).subscribe(new Action1<Long>() {@Overridepublic void call(Long aLong) {System.out.println( "throttleFirst--" + aLong );}}) ;

這段代碼，是循環(huán)發(fā)送數(shù)據(jù)，每秒發(fā)送一個。throttleFirst( 3 , TimeUnit.SECONDS ) ??在3秒內(nèi)只取第一個事件，其他的事件丟棄。

運行結(jié)果

14、distinct ? ?過濾重復(fù)的數(shù)據(jù)

List<String> list = new ArrayList<>() ;list.add( "1" ) ;list.add( "2" ) ;list.add( "1" ) ;list.add( "3" ) ;list.add( "4" ) ;list.add( "2" ) ;list.add( "1" ) ;list.add( "1" ) ;Observable.from( list ).distinct().subscribe(new Action1<String>() {@Overridepublic void call(String s) {System.out.println( "distinct--" + s );}}) ;

從結(jié)果可以看出，重復(fù)的數(shù)據(jù)已經(jīng)被過濾掉了

?

??distinctUntilChanged() ?過濾連續(xù)重復(fù)的數(shù)據(jù)

List<String> list = new ArrayList<>() ;list.add( "1" ) ;list.add( "2" ) ;list.add( "1" ) ;list.add( "3" ) ;list.add( "4" ) ;list.add( "4" ) ;list.add( "2" ) ;list.add( "1" ) ;list.add( "1" ) ;Observable.from( list ).distinctUntilChanged().subscribe(new Action1<String>() {@Overridepublic void call(String s) {System.out.println( "distinctUntilChanged--" + s );}}) ;

?運行結(jié)果

從結(jié)果可以看出，連續(xù)重復(fù)的數(shù)據(jù)已經(jīng)被過濾掉了

?

15、debounce() 操作符

? ? ? ?一段時間內(nèi)沒有變化，就會發(fā)送一個數(shù)據(jù)。

? ? ? ?使用場景：百度關(guān)鍵詞聯(lián)想提示。在輸入的過程中是不會從服務(wù)器拉數(shù)據(jù)的。當輸入結(jié)束后，在400毫秒沒有輸入就會去獲取數(shù)據(jù)。

? ? ? ? ? ? ? ? ? ?避免了，多次請求給服務(wù)器帶來的壓力.

?

?16、doOnSubscribe()?

? ? ? ? 使用場景： 可以在事件發(fā)出之前做一些初始化的工作，比如彈出進度條等等

? ? ? ? ?注意：

? ? ? ? ? ? ?1、doOnSubscribe() 默認運行在事件產(chǎn)生的線程里面，然而事件產(chǎn)生的線程一般都會運行在 io 線程里。那么這個時候做一些，更新UI的操作，是線程不安全的。

? ? ? ? ? ? ? ? ?所以如果事件產(chǎn)生的線程是io線程，但是我們又要在doOnSubscribe() 更新UI ， 這時候就需要線程切換。

? ? ? ? ? ? ?2、如果在?doOnSubscribe()?之后有?subscribeOn()?的話，它將執(zhí)行在離它最近的?subscribeOn()?所指定的線程。? ?

? ? ? ? ? ? ?3、 subscribeOn() 事件產(chǎn)生的線程 ；?observeOn() : 事件消費的線程

Observable.create(onSubscribe).subscribeOn(Schedulers.io()).doOnSubscribe(new Action0() {@Overridepublic void call() {progressBar.setVisibility(View.VISIBLE); // 需要在主線程執(zhí)行 }}).subscribeOn(AndroidSchedulers.mainThread()) // 指定主線程 .observeOn(AndroidSchedulers.mainThread()).subscribe(subscriber);

?

17、range 操作符的使用?

? ? 首先看range 方法的源碼

　

public static Observable<Integer> range(int start, int count) {if (count < 0) {throw new IllegalArgumentException("Count can not be negative");}if (count == 0) {return Observable.empty();}if (start > Integer.MAX_VALUE - count + 1) {throw new IllegalArgumentException("start + count can not exceed Integer.MAX_VALUE");}if(count == 1) {return Observable.just(start);}return Observable.create(new OnSubscribeRange(start, start + (count - 1)));}//可以通過第三個參數(shù)控制range執(zhí)行的線程public static Observable<Integer> range(int start, int count, Scheduler scheduler) {return range(start, count).subscribeOn(scheduler);}

Range操作符發(fā)射一個范圍內(nèi)的有序整數(shù)序列，你可以指定范圍的起始和長度。

RxJava將這個操作符實現(xiàn)為range函數(shù)，它接受兩個參數(shù)，一個是范圍的起始值，一個是范圍的數(shù)據(jù)的數(shù)目。如果你將第二個參數(shù)設(shè)為0，將導(dǎo)致Observable不發(fā)射任何數(shù)據(jù)（如果設(shè)置為負數(shù)，會拋異常）。

range默認不在任何特定的調(diào)度器上執(zhí)行。有一個變體可以通過可選參數(shù)指定Scheduler。

例子

Observable.range( 10 , 3 ).subscribe(new Action1<Integer>() {@Overridepublic void call(Integer integer) {Log.v( "rx_range " , "" + integer ) ;}}) ;

　　結(jié)果

/rx_range: 10
/rx_range: 11
/rx_range: 12

?

?18、defer 操作符

例子

package app.com.myapplication;import android.support.v7.app.AppCompatActivity; import android.os.Bundle; import android.util.Log; import rx.Observable; import rx.functions.Action1; import rx.functions.Func0;public class DeferActivity extends AppCompatActivity {String i = "10" ;@Overrideprotected void onCreate(Bundle savedInstanceState) {super.onCreate(savedInstanceState);setContentView(R.layout.activity_defer);i = "11 " ;Observable<String> defer = Observable.defer(new Func0<Observable<String>>() {@Overridepublic Observable<String> call() {return Observable.just( i ) ;}}) ;Observable test = Observable.just( i ) ;i = "12" ;defer.subscribe(new Action1<String>() {@Overridepublic void call(String s) {Log.v( "rx_defer " , "" + s ) ;}}) ;test.subscribe(new Action1() {@Overridepublic void call(Object o) {Log.v( "rx_just " , "" + o ) ;}}) ;} }

　　結(jié)果

/rx_defer: 12
/rx_just: 11

• 可以看到，just操作符是在創(chuàng)建Observable就進行了賦值操作，而defer是在訂閱者訂閱時才創(chuàng)建Observable，此時才進行真正的賦值操作。
• Defer操作符會一直等待直到有觀察者訂閱它，然后它使用Observable工廠方法生成一個Observable。它對每個觀察者都這樣做，因此盡管每個訂閱者都以為自己訂閱的是同一個Observable，事實上每個訂閱者獲取的是它們自己的單獨的數(shù)據(jù)序列。
• 在某些情況下，等待直到最后一分鐘（就是知道訂閱發(fā)生時）才生成Observable可以確保Observable包含最新的數(shù)據(jù)。

?

?

?

RxJava 和 RxAndroid 三（生命周期控制和內(nèi)存優(yōu)化）

?

總結(jié)

以上是生活随笔為你收集整理的RxJava 和 RxAndroid 二（操作符的使用）的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。