RxJava 和 RxAndroid 二（操作符的使用）

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

 

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

        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" ) ;

        Observable observable1 = Observable.from( list1 ) ;
        Observable observable2 = Observable.from( list2 ) ;

        //合並數據  先發送observable2的全部數據，然后發送 observable1的全部數據
        Observable observable = Observable.merge( observable2 , observable1 ) ;

        observable.subscribe(new Action1() {
            @Override
            public void call(Object o) {
             System.out.println( "rx-- " + o );
            }
        }) ;

運行結果

2、zip  操作符，合並多個觀察對象的數據。並且允許 Func2（）函數重新發送合並后的數據

 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 >() {
            @Override
            public String call(String s1 , String s2 ) {
                return s1 + s2  ;
            }
        }) ;

        observable3.subscribe(new Action1() {
            @Override
            public void call(Object o) {
                System.out.println( "zip-- " + o );
            }
        }) ;

　　運行效果：從效果圖上可以看出，合並兩個的觀察對象數據項應該是相等的；如果出現了數據項不等的情況，合並的數據項以最小數據隊列為准。

 

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

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

　　運行效果：

        第一次發射得到1，作為結果與2相加；發射得到3，作為結果與3相加，以此類推，打印結果：

       

 

4、filter 過濾操作符的使用

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

　　運行效果

 

 

5、 消息數量過濾操作符的使用    

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

 //take 發送前3個數據
        Observable observable = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;
        observable.take( 3 )
                .subscribe(new Action1() {
                    @Override
                    public void call(Object o) {
                        System.out.println( "take-- " +  o );
                    }
                })   ;

        //takeLast 發送最后三個數據
        Observable observable2 = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;
        observable2.takeLast( 3 )
                .subscribe(new Action1() {
                    @Override
                    public void call(Object o) {
                        System.out.println( "takeLast-- " +  o );
                    }
                })   ;

        //first 只發送第一個數據
        Observable observable3 = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;
        observable3.first()
                .subscribe(new Action1() {
                    @Override
                    public void call(Object o) {
                        System.out.println( "first-- " +  o );
                    }
                })   ;

        //last 只發送最后一個數據
        Observable observable4 = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;
        observable4.last()
                .subscribe(new Action1() {
                    @Override
                    public void call(Object o) {
                        System.out.println( "last-- " +  o );
                    }
                })   ;

        //skip() 跳過前2個數據發送后面的數據
        Observable observable5 = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;
        observable5.skip( 2 )
                .subscribe(new Action1() {
                    @Override
                    public void call(Object o) {
                        System.out.println( "skip-- " +  o );
                    }
                })   ;

        //skipLast() 跳過最后兩個數據，發送前面的數據
        Observable observable6 = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;
        observable5.skipLast( 2 )
                .subscribe(new Action1() {
                    @Override
                    public void call(Object o) {
                        System.out.println( "skipLast-- " +  o );
                    }
                })   ;

效果圖

 

 6、elementAt 、elementAtOrDefault

//elementAt() 發送數據序列中第n個數據 ，序列號從0開始
        //如果該序號大於數據序列中的最大序列號，則會拋出異常，程序崩潰
        //所以在用elementAt操作符的時候，要注意判斷發送的數據序列號是否越界

        Observable observable7 = Observable.just( 1 , 2 , 3 , 4 , 5 , 6 , 7 ) ;
        observable7.elementAt( 3 )
                .subscribe(new Action1() {
                    @Override
                    public void call(Object o) {
                        System.out.println( "elementAt-- " +  o );
                    }
                })   ;

        //elementAtOrDefault( int n , Object default ) 發送數據序列中第n個數據 ，序列號從0開始。
        //如果序列中沒有該序列號，則發送默認值
        Observable observable9 = Observable.just( 1 , 2 , 3 , 4 , 5 ) ;
        observable9.elementAtOrDefault(  8 , 666  )
                .subscribe(new Action1() {
                    @Override
                    public void call(Object o) {
                        System.out.println( "elementAtOrDefault-- " +  o );
                    }
                })   ;

　　運行結果

 

 7、startWith() 插入數據

        //插入普通數據
        //startWith 數據序列的開頭插入一條指定的項 , 最多插入9條數據
        Observable observable = Observable.just( "aa" , "bb" , "cc" ) ;
        observable
                .startWith( "11" , "22" )
                .subscribe(new Action1() {
                    @Override
                    public 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() {
                    @Override
                    public void call(Object o) {
                        System.out.println( "startWith2 -- " + o );
                    }
                }) ;

　　運行結果

 

 8、delay操作符，延遲數據發送

 Observable<String> observable = Observable.just( "1" , "2" , "3" , "4" , "5" , "6" , "7" , "8" ) ;

        //延遲數據發射的時間，僅僅延時一次，也就是發射第一個數據前延時。發射后面的數據不延時
        observable.delay( 3 , TimeUnit.SECONDS )  //延遲3秒鍾
                .subscribe(new Action1() {
                    @Override
                    public void call(Object o) {
                        System.out.println("delay-- " + o);
                    }
                }) ;

　　

9、Timer  延時操作符的使用

      使用場景：xx秒后，執行xx      

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

     timer 返回一個 Observable , 它在延遲一段給定的時間后發射一個簡單的數字0

    timer 操作符默認在computation調度器上執行，當然也可以用 Scheduler 在定義執行的線程。

 

 delay 、timer 總結：　

•  相同點：delay 、 timer 都是延時操作符。
•  不同點：delay  延時一次，延時完成后，可以連續發射多個數據。timer延時一次，延時完成后，只發射一次數據。

                                             

 10、interval 輪詢操作符，循環發送數據，數據從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 ;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_interval);

        //參數一：延遲時間  參數二：間隔時間  參數三：時間顆粒度
        Observable observable =  Observable.interval(3000, 3000, TimeUnit.MILLISECONDS) ;
        subscription = observable.subscribe(new Action1() {
            @Override
            public void call(Object o) {
                System.out.println( "interval-  " + o );
            }
        })  ;
    }

    @Override
    protected void onDestroy() {
        super.onDestroy();
        if ( subscription != null ){
            subscription.unsubscribe();
        }
    }
}

　　

 

 11、doOnNext() 操作符，在每次 OnNext() 方法被調用前執行

       使用場景：從網絡請求數據，在數據被展示前，緩存到本地

 Observable observable = Observable.just( "1" , "2" , "3" , "4" ) ;

        observable.doOnNext(new Action1() {
            @Override
            public void call(Object o) {
                System.out.println( "doOnNext--緩存數據" + o  );
            }
        })
                .subscribe(new Observer() {
                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onNext(Object o) {
                        System.out.println( "onNext--" + o  );
                    }
                }) ;

　　

 

 12、Buffer 操作符

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

         

    使用場景：一個按鈕每點擊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)   //把每兩個數據為一組打成一個包，然后發送
                .subscribe(new Action1<List<String>>() {
                    @Override
                    public void call(List<String> strings) {
                        System.out.println( "buffer---------------" );
                        Observable.from( strings ).subscribe(new Action1<String>() {
                            @Override
                            public void call(String s) {
                                System.out.println( "buffer data --" + s  );
                            }
                        }) ;
                    }
                });

　　

     例子2： 

 //第1、2 個數據打成一個數據包，跳過第三個數據 ； 第4、5個數據打成一個包，跳過第6個數據
        observable.buffer( 2 , 3 )  //把每兩個數據為一組打成一個包，然后發送。第三個數據跳過去
                .subscribe(new Action1<List<String>>() {
                    @Override
                    public void call(List<String> strings) {

                        System.out.println( "buffer22---------------" );
                        Observable.from( strings ).subscribe(new Action1<String>() {
                            @Override
                            public void call(String s) {
                                System.out.println( "buffer22 data --" + s  );
                            }
                        }) ;
                    }
                }) ;

　　

 

 13、throttleFirst 操作符

         在一段時間內，只取第一個事件，然后其他事件都丟棄。

         使用場景：1、button按鈕防抖操作，防連續點擊   2、百度關鍵詞聯想，在一段時間內只聯想一次，防止頻繁請求服務器   

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

這段代碼，是循環發送數據，每秒發送一個。throttleFirst( 3 , TimeUnit.SECONDS )   在3秒內只取第一個事件，其他的事件丟棄。

運行結果

14、distinct    過濾重復的數據

 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>() {
                    @Override
                    public void call(String s) {
                        System.out.println( "distinct--" + s );
                    }
                }) ;

從結果可以看出，重復的數據已經被過濾掉了

 

  distinctUntilChanged()  過濾連續重復的數據

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>() {
                    @Override
                    public void call(String s) {
                        System.out.println( "distinctUntilChanged--" + s );
                    }
                }) ;

 運行結果

從結果可以看出，連續重復的數據已經被過濾掉了

 

15、debounce() 操作符

       一段時間內沒有變化，就會發送一個數據。

       使用場景：百度關鍵詞聯想提示。在輸入的過程中是不會從服務器拉數據的。當輸入結束后，在400毫秒沒有輸入就會去獲取數據。

                   避免了，多次請求給服務器帶來的壓力.

 

 16、doOnSubscribe() 

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

         注意：

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

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

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

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

    Observable.create(onSubscribe)
    .subscribeOn(Schedulers.io())
    .doOnSubscribe(new Action0() {
        @Override
        public void call() {
            progressBar.setVisibility(View.VISIBLE); // 需要在主線程執行
        }
    })
    .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)));
    }

    
    //可以通過第三個參數控制range執行的線程
    public static Observable<Integer> range(int start, int count, Scheduler scheduler) {
        return range(start, count).subscribeOn(scheduler);
    }

Range操作符發射一個范圍內的有序整數序列，你可以指定范圍的起始和長度。

RxJava將這個操作符實現為range函數，它接受兩個參數，一個是范圍的起始值，一個是范圍的數據的數目。如果你將第二個參數設為0，將導致Observable不發射任何數據（如果設置為負數，會拋異常）。

range默認不在任何特定的調度器上執行。有一個變體可以通過可選參數指定Scheduler。

例子

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

　　結果

/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" ;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_defer);

        i = "11 " ;

        Observable<String> defer = Observable.defer(new Func0<Observable<String>>() {
            @Override
            public Observable<String> call() {
                return Observable.just( i ) ;
            }
        }) ;

        Observable test = Observable.just(  i ) ;

        i = "12" ;

        defer.subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                Log.v( "rx_defer  " , "" + s ) ;
            }
        }) ;

        test.subscribe(new Action1() {
            @Override
            public void call(Object o) {
                Log.v( "rx_just " , "" + o ) ;
            }
        }) ;
    }
}

　　結果

/rx_defer: 12
/rx_just: 11

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

 

 

 

RxJava 和 RxAndroid 三（生命周期控制和內存優化）