前言:對Rx不了解的朋友可以先看我的第一篇博文 RxJava 和 RxAndroid 一 (基礎),是對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 //合并資料 先發送observable2的全部資料,然後發送 observable1的全部資料
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
運作結果
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,此時才進行真正的指派操作。
-
操作符會一直等待直到有觀察者訂閱它,然後它使用Observable工廠方法生成一個Observable。它對每個觀察者都這樣做,是以盡管每個訂閱者都以為自己訂閱的是同一個Observable,事實上每個訂閱者擷取的是它們自己的單獨的資料序列。Defer
- 在某些情況下,等待直到最後一分鐘(就是知道訂閱發生時)才生成Observable可以確定Observable包含最新的資料。