過濾操作符
3.1 filter
符合某種規則的Observable才會向下傳遞，例子
Observable.range(100,10).filter(new Func1() {
@Override
public Boolean call(Integer integer) {
return integer>105;
}
}).subscribe(new Action1() {
@Override
public void call(Integer integer) {
}
});
上面的Subscriber只會得到大於105的數值
3.2 first
只發射第一項，或者是滿足條件的第一項，兩個例子
Observable.range(100,5).first();// 發射100
Observable.range(100,5).first(new Func1() {
@Override
public Boolean call(Integer integer) {
return integer>102;//發射103
}
});
3.3 Debounce
當操作太頻繁的時候，在第N個節點發射之前會忽略錢N-1個節點，比如我們的Edittext去seach的時候，可能需要過濾前多少秒的操作只是需要後續的操作。簡單點來說，就是，他需要的是在某個時間點之後的結果，在這個時間點之前的結果他會忽略不再發送給下一級。例子：
Observable.create(new Observable.OnSubscribe() {
@Override
public void call(Subscriber subscriber) {
try {
for (int i = 1; i <= 10; i++) {
subscriber.onNext(i);
Thread.sleep(100*i);
}
subscriber.onCompleted();
} catch (InterruptedException e) {
e.printStackTrace();
subscriber.onError(e);
}
}
}).subscribeOn(Schedulers.newThread())
.debounce(500, TimeUnit.MILLISECONDS).subscribe(new Subscriber() {
@Override
public void onCompleted() {
Log.i(TAG, "OnComplete");
}
@Override
public void onError(Throwable e) {
Log.i(TAG, "Throwable");
}
@Override
public void onNext(Integer integer) {
Log.i(TAG, "integer == "+integer.toString());
}
});
輸出就是6.7.8.9.10
總結就是debounce操作符就是類似一個控制入口，他是有一個門卡的，門卡一般是時間，在這之前的忽略，之後的可以進入
例子是來自 Android RxJava使用介紹（三） RxJava的操作符
其中，還有一個throttleWithTimeout操作符，他的底層也是通過調用debounce操作符完成的，作用類似。但是debounce比她在、多了一個功能就是可以通過函數來進行限流操作。例子如下：
Observable.just(1, 2, 3, 4, 5).debounce(new Func1>() {
@Override
public Observable call(Integer integer) {
return Observable.create(new Observable.OnSubscribe() {
@Override
public void call(Subscriber subscriber) {
if (integer % 2 == 0 && !subscriber.isUnsubscribed()) {
subscriber.onNext(integer);
subscriber.onCompleted();
}
}
});
}
}).subscribeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1() {
@Override
public void call(Integer integer) {
Log.i(TAG, integer.toString() + "==integer");
}
});
結果輸出是2.4.5。5為什麼在其中？因為，在debounce操作符中，最後一個Observable，在被調用onComplete之前的時間間隔內被調用，那麼他也會被傳遞到下一級。
3.4 Distinct
只允許沒有發射過的數據發射，可以定義過濾規則，例子
Observable.range(1,5).distinct(new Func1() {
@Override
public String call(Integer integer) {
if(integer<=2) return "1";// 定義過濾規則
return "2";
}
}).subscribe(new Action1() {
@Override
public void call(Integer integer) {
Log.i(TAG, integer.toString());
}
});
Observable.just(1,2,3,1,2).distinct().subscribe(new Action1() { // 使用默認規則
@Override
public void call(Integer integer) {
Log.i(TAG, integer.toString());
}
});
輸出
09-02 00:32:45.009 4762-4762/? I/RxJavaTest: 1==自定義
09-02 00:32:45.009 4762-4762/? I/RxJavaTest: 3==自定義
09-02 00:32:45.009 4762-4762/? I/RxJavaTest: 1===系統
09-02 00:32:45.009 4762-4762/? I/RxJavaTest: 2===系統
09-02 00:32:45.009 4762-4762/? I/RxJavaTest: 3===系統
3.5 * elementAt*
某一位置的元素可以發射，需要需要越界的問題，他是從0開始計算的例子
Observable.range(10,10).elementAt(9).subscribe(new Action1() {
@Override
public void call(Integer integer) {
}
});
3.6 last
最後一個或者滿足條件的最後一個，跟first相反，例子
Observable.range(10,5).last();
Observable.range(10,5).last(new Func1() {
@Override
public Boolean call(Integer integer) {
return integer>13;
}
}).subscribe(new Action1() {
@Override
public void call(Integer integer) {
Log.i(TAG, integer.toString()+"===last");
}
});
3.7 skip
跳過前N項，從第N+1開始發射，例子
Observable.range(10,20).skip(10);
這個例子是丟棄了10~29中的10~19，保留20~29，還有兩個重載的方法
skip(long,TimeUnit));// 跳過前long時間的Observable，發射之後的
skip(long,TimeUnit,Scheduler));//可以定義調度器
結果是14，兩個都是
3.8 skipLast
不發射後n項數據，跟skip相反，例如
Observable.range(1,10).skipLast(3).subscribe(new Action1() {
@Override
public void call(Integer integer) {
Log.i(TAG, integer.toString() + "==skip");
}
});
輸出：1,2,3,4,5,6,7
3.9 take
只取前n項數據，例如有10項數據，那麼take(4)，就是只是圈前4項數據，例如
Observable.range(10,10).take(5).subscribe(new Action1() {
@Override
public void call(Integer integer) {
Log.i(TAG, integer.toString() + "==take");
}
});
結果是：10,11，12,13,14
3.10 taskLast
只發射最後n項數據，比如有10項數據，taskLast(3)，則是只發射最後3項數據，例子：
Observable.range(9,5).takeLast(2).subscribe(new Action1() {
@Override
public void call(Integer integer) {
Log.i(TAG, integer.toString() + "==integer");
}
});
輸出：12和13
需要注意的是他是一個線程同步的，他可能造成Observable發射延遲，因為他只有過濾掉前面的數據才會輪到後面，然後後面的數據才會發射。
重構的方法有：
takeLast(final int count) //發射後面n個數據，n大於數據全部的時候則發射全部n小於0報異常
takeLast(int count, long time, TimeUnit unit)// 發射最後給定時間的n個數據
takeLast(int count, long time, TimeUnit unit, Scheduler scheduler)
takeLast(long time, TimeUnit unit, Scheduler scheduler)//發射最後給定時間的全部數據，可以指定調度器
takeLast(long time, TimeUnit unit)
同時，他還有一個變形，takeLastBuffer*，takeLast是一個一個數據源返回，但是takeLastBuffer是以List的形式返回，例如：
Observable.range(20,10).takeLastBuffer(5).subscribe(new Action1>() {
@Override
public void call(List integers) {
Log.i(TAG, integers.toString() + "==integer");
}
});
3.11 ignoreElements
忽略所有的元素，只是發射結束事件或者是錯誤時間，例如
Observable.range(10,10).ignoreElements().subscribe(new Subscriber() {
@Override
public void onCompleted() {
Log.i(TAG, "onCompleted");
}
@Override
public void onError(Throwable e) {
Log.i(TAG, "onError");
}
@Override
public void onNext(Integer integer) {
Log.i(TAG, integer.toString() + "==ignoreElements");
}
});
3.12 last
只發射最後的一項數據，忽略前面n項數據，例如
Observable.range(10,10).last().subscribe(new Action1() {
@Override
public void call(Integer integer) {
Log.i(TAG, integer.toString() + "==last");
}
});
結果是：19
還有一個last(Func1)的重載函數，他可以規定那個才是last，比如
Observable.range(10,10).last(new Func1() {
@Override
public Boolean call(Integer integer) {
Log.i(TAG, integer.toString() + "==last");
return integer == 14;
}
}).subscribe(new Action1() {
@Override
public void call(Integer integer) {
Log.i(TAG, integer.toString() + "==last func");
}
});
在subscribe裡面輸出的就是14而不是19，還有一個相似函數是:lastOrDefault(T)，他會在Observable沒有發射任何數據的時候發射默認值T。
3.13 sample操作符，
他是在某一段時間內采集過去發射的舊數據再次發射，比如：
Observable.interval(1,TimeUnit.SECONDS).sample(2,TimeUnit.SECONDS).subscribe(new Action1() {
@Override
public void call(Long aLong) {
Log.i(TAG, aLong.toString() + "==sample ");
}
});
interval創造一個Observable之後每個1s發射一個新數據，然後sample會在2s重復的去采集已經發射的舊數據再次發射，輸出的結果是:0,2,4,6…
它默認開啟新線程，我們也可指線程給sample運行
sample例子來自RxJava 過濾操作符 sample
