目錄

建立操作

create

建立一個具有發射能力的Observable 

Observable.create(e -> {
    e.onNext("Love");
    e.onNext("For");
    e.onNext("You!");
    e.onComplete();
}).subscribe(s -> System.out.println(s));

just

只是簡單的原樣發射,可將陣列或Iterable當做單個數據。它接受一至九個引數

Observable.just("Love", "For", "You!")
                .subscribe
 
(s -> System.out.println(s));

empty

建立一個不發射任何資料但是正常終止的Observable

Observable.empty();

never

建立一個不發射資料也不終止的Observable

Observable.never();

error

建立一個不發射資料以一個錯誤終止的Observable 

Observable.error(new Throwable("O__O"));

timer

在延遲一段給定的時間後發射一個簡單的數字0

Observable.timer(1000, TimeUnit.MILLISECONDS
 
)
              .subscribe(s -> System.out.println(s));

range

• start:起始值
• count:一個是範 圍的資料的數目。0不傳送 ，負數 異常

Observable.range(5, 3)
        //輸出 5,6,7
        .subscribe(s -> System.out.println(s));

intervalRange

• start,count:同range
• initialDelay 傳送第一個值的延遲時間
• period 每兩個發射物的間隔時間
• unit,scheduler 額你懂的

Observable.intervalRange(5, 100, 3000, 100,
                TimeUnit.MILLISECONDS, Schedulers.io())
                .subscribe(s -> System.out.println(s));

interval

相當於intervalRange的start=0
period 這個值一旦設定後是不可變化的

//period 以後的美每次間隔 這個值一旦設定後是不可變化的  所以 count方法無效的！
int[] s = new int[]{0};
 Observable.interval(3000, 100 + count(s), TimeUnit.MILLISECONDS, Schedulers.io())
         .subscribe(s2 -> System.out.println(s2));
 private int count(int[] s) {
         int result = s[0] * 1000;
         s[0] = s[0] + 1;
         return result;
 }

defer

直到有觀察者訂閱時才建立Observable，並且為每個觀察者建立一個新的Observable 

Observable.defer(() -> Observable.just("Love", "For", "You!"))
              .subscribe(s -> System.out.println(s));

from系列

fromArray

Integer[] items = {0, 1, 2, 3, 4, 5};
       Observable.fromArray(items).subscribe(
               integer -> System.out.println(integer));

fromCallable

Observable.fromCallable(() -> Arrays.asList("hello", "gaga"))
                .subscribe(strings -> System.out.println(strings))

fromIterable

Observable.fromIterable(Arrays.<String>asList("one", "two", "three"))
             .subscribe(integer -> System.out.println(integer));

fromFuture

Observable.fromFuture(Observable.just(1).toFuture())
               .doOnComplete(() -> System.out.println("complete"))
               .subscribe();

過濾操作

elementAt

只發射第N項資料

<!-- 無預設值版本 -->
 Observable.just(1,2)
            .elementAt(0)
            .subscribe(o -> System.out.print(o ));//結果:1
<!-- 帶預設值的變體版本 -->
Observable.range(0, 10)
//        如果索引值大於資料 項數，它會發射一個預設值(通過額外的引數指定)，而不是丟擲異常。
//    但是如果你傳遞一 個負數索引值，它仍然會丟擲一個 IndexOutOfBoundsException 異常。
                .elementAt(100, -100)
                .subscribe(o -> System.out.print(o + "\t"));

IgnoreElements

如果你不關心一個Observable發射的資料，但是希望在它完成時或遇到錯誤終止時收到通知

Observable.range(0, 10)
               .ignoreElements()
               .subscribe(() -> System.out.println("complete")
                       , throwable -> System.out.println("throwable"));

take系列

take

• 變體 count系列:只發射前面的N項資料

Observable.range(0,10)
               .take(3)
               .subscribe(o -> System.out.print(o + "\t"))

• 變體 time系列: 發射Observable開始的那段時間發射 的資料

Observable.range(0,10)
          .take(100, TimeUnit.MILLISECONDS)
          .subscribe(o -> System.out.print(o + "\t"));

takeLast

• 變體 count系列:只發射後面的N項資料

Observable.range(0,10)
                    .takeLast(3)
                    .subscribe(o -> System.out.print(o + "\t"));

• 變體 time系列: 發射在原始Observable的生命周 期內最後一段時間內發射的資料

Observable.range(0,10)
               .takeLast(100, TimeUnit.MILLISECONDS)
               .subscribe(o -> System.out.print(o + "\t"));

takeUntil

傳送complete的結束條件 當然傳送結束之前也會包括這個值

Observable.just(2,3,4,5)
        //傳送complete的結束條件 當然傳送結束之前也會包括這個值
        .takeUntil(integer ->  integer>3)
        .subscribe(o -> System.out.print(o + "\t"));//2,3,4

takeWhile

當不滿足這個條件 會發送結束 不會包括這個值

Observable.just(2,3,4,5)
        //當不滿足這個條件 會發送結束 不會包括這個值
        .takeWhile(integer ->integer<=4 )
        .subscribe(o -> System.out.print(o + "\t"));//2,3,4

skip系列

skip

• 變體 count系列:丟棄Observable發射的前N項資料

Observable.range(0,5)
               .skip(3)
               .subscribe(o -> System.out.print(o + "\t"));

• 變體 time系列:丟棄原始Observable開始的那段時間發 射的資料

Observable.range(0,5)
               .skip(3)
               .subscribe(o -> System.out.print(o + "\t"));

skipLast

• 變體 count系列:丟棄Observable發射的前N項資料

Observable.range(0,5)
               .skipLast(3)
               .subscribe(o -> System.out.print(o + "\t"));

• 變體 time系列:丟棄在原始Observable的生命周 期內最後一段時間內發射的資料

Observable.range(0,10)
             .skipLast(100, TimeUnit.MILLISECONDS)
             .subscribe(o -> System.out.print(o + "\t"));

distinct

去重

• keySelector:這個函式根據原始Observable發射的資料項產生一個 Key，然後，比較這些Key而不是資料本身，來判定兩個資料是否是不同的

Observable.just(1, 2, 1, 2, 3)
            //這個函式根據原始Observable發射的資料項產生一個 Key，
            // 然後，比較這些Key而不是資料本身，來判定兩個資料是否是不同的
            .distinct(integer -> Math.random())
            .subscribe(o -> System.out.print(o + "\t"));
日誌:
原因 key不同 所以當做資料不同處理
1   2   1   2   3

• 無參版本 就是內部實現了的keySelector通過生成的key就是value本身

Observable.just(1, 2, 1, 2, 3)
             .distinct()
             .subscribe(o -> System.out.print(o + "\t"));
 日誌:
 1   2   3

distinctUntilChanged

(相鄰去重):它只判定一個數據和它的直接前驅是 否是不同的
其他概念與distinct一樣

throttleWithTimeout/debounce

操作符會過濾掉髮射速率過快的資料項
throttleWithTimeout/debounce： 含義相同
如果傳送資料後 指定時間段內沒有新資料的話 。則傳送這條
如果有新資料 則以這個新資料作為將要傳送的資料項，並且重置這個時間段，重新計時。

  Observable.create(e -> {
            e.onNext("onNext 0");
            Thread.sleep(100);
            e.onNext("onNext 1");
            Thread.sleep(230);
            e.onNext("onNext 2");
            Thread.sleep(300);
            e.onNext("onNext 3");
            Thread.sleep(400);
            e.onNext("onNext 4");
            Thread.sleep(500);
            e.onNext("onNext 5");
            e.onNext("onNext 6");
        })
                .debounce(330, TimeUnit.MILLISECONDS)
//                .throttleWithTimeout(330, TimeUnit.MILLISECONDS)
                .subscribeOn(Schedulers.newThread())
                .observeOn(Schedulers.newThread())
                .subscribe(o -> System.out.println(o));//結果 3 4 6

filter

只發射通過了謂詞測試的資料項

Observable.range(0, 10)
        //過濾掉false的元素
        .filter(integer -> integer % 2 == 0)
        .subscribe(o -> System.out.print(o + "\t"));

ofType

ofType 是 filter 操作符的一個特殊形式。它過濾一個Observable只返回指定型別的資料

Observable.just(0, "what?", 1, "String", 3)
             //ofType 是 filter 操作符的一個特殊形式。它過濾一個Observable只返回指定型別的資料。
             .ofType(String.class)
             .subscribe(o -> System.out.print(o + "\t"));

first

只發射第一項(或者滿足某個條件的第一項)數 感覺和take(1) elementAt(0)差不多

Observable.range(0, 10)
          //如果元資料沒有傳送  則有傳送預設值
          .first(-1)
          .subscribe(o -> System.out.print(o + "\t"));

last

只發射最後一項(或者滿足某個條件的最後一項)資料 感覺和takeLast(1)差不多

Observable.empty()
              //如果元資料沒有傳送  則有傳送預設值
              .last(-1)
              .subscribe(o -> System.out.print(o + "\t"));

sample/throttleLast

週期取樣後 傳送最後的資料

throttleFirst

週期取樣 的第一條資料 傳送
注意: 如果是已經被髮送過的 則不會繼續傳送

 Observable.create(e -> {
            e.onNext("onNext 0");
            Thread.sleep(100);
            e.onNext("onNext 1");
            Thread.sleep(50);
            e.onNext("onNext 2");
            Thread.sleep(70);
            e.onNext("onNext 3");
            Thread.sleep(200);
            e.onNext("onNext 4");
            e.onNext("onNext 5");
        })
                .subscribeOn(Schedulers.newThread())
                .observeOn(Schedulers.newThread())
                <!--  結果 : onNext 2 onNext 3    onNext 5    -->
                .sample(200, TimeUnit.MILLISECONDS,Schedulers.newThread())
                <!--  結果 : onNext 2 onNext 3    onNext 5    -->
//                .throttleLast(200, TimeUnit.MILLISECONDS,Schedulers.newThread())
                <!--  結果 : onNext 0 onNext 3    onNext 4    -->
//                .throttleFirst(200, TimeUnit.MILLISECONDS,Schedulers.newThread())
                .subscribe(o -> System.out.print(o + "\t"));