一.使用Disposable切斷被觀察與觀察者之間連接
語法：disposable.dispose();
public class SimpleActivity extends Activity {
private static final String TAG = "SimpleActivity";
private Disposable disposable;
@Override
protected void onCreate(@Nullable Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Exception {
//2
Log.e(TAG, "subscribe");
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onComplete();
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
//1
Log.e(TAG, "onSubscribe");
disposable = d;
}
@Override
public void onNext(@NonNull Integer integer) {
//3
Log.e(TAG, "onNext收到：" + integer);
if (integer == 2) {
disposable.dispose();//切斷被觀察者與觀察者之間連接
Log.e(TAG, "切斷了連接");
}
}
@Override
public void onError(@NonNull Throwable e) {
Log.e(TAG, "onError");
}
@Override
public void onComplete() {
//4
Log.e(TAG, "onComplete");
}
});
}
}
打印為：
E/SimpleActivity: onSubscribe
E/SimpleActivity: subscribe
E/SimpleActivity: onNext收到：1
E/SimpleActivity: onNext收到：2
E/SimpleActivity: 切斷了連接
二.被觀察者的多種創(chuàng)建方式
1.just()
語法：
Observable.just(1, 2, 3)
注意事項：
（1）只能發(fā)送10個以內(nèi)事件
Observable.just(1, 2, 3)
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
//1
Log.e(TAG,"onSubscribe");
}
@Override
public void onNext(@NonNull Integer integer) {
//2
Log.e(TAG,"onNext收到："+integer);
}
@Override
public void onError(@NonNull Throwable e) {
}
@Override
public void onComplete() {
//3
Log.e(TAG,"onComplete");
}
});
打印為：
E/CaoZuoFuActivity: onSubscribe
E/CaoZuoFuActivity: onNext收到：1
E/CaoZuoFuActivity: onNext收到：2
E/CaoZuoFuActivity: onNext收到：3
E/CaoZuoFuActivity: onComplete
2.fromArray()
語法：
Observable.fromArray(nums)
注意事項：
（1）可發(fā)送超過10個事件
（2）傳輸對象為數(shù)組
Integer[] nums={1,2,3};
Observable.fromArray(nums)
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
//1
Log.e(TAG,"onSubscribe");
}
@Override
public void onNext(@NonNull Integer integer) {
//2
Log.e(TAG,"onNext收到："+integer);
}
@Override
public void onError(@NonNull Throwable e) {
}
@Override
public void onComplete() {
//3
Log.e(TAG,"onComplete");
}
});
E/CaoZuoFuActivity: onSubscribe
E/CaoZuoFuActivity: onNext收到：1
E/CaoZuoFuActivity: onNext收到：2
E/CaoZuoFuActivity: onNext收到：3
E/CaoZuoFuActivity: onComplete
3.fromIterable()
語法：
Observable.fromIterable(list)
注意事項：
（1）可發(fā)送超過10個事件
（2）傳輸對象為集合
List<Integer> list = new ArrayList<>();
list.add(1);
list.add(2);
list.add(3);
Observable.fromIterable(list)
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
//1
Log.e(TAG,"onSubscribe");
}
@Override
public void onNext(@NonNull Integer integer) {
//2
Log.e(TAG,"onNext收到："+integer);
}
@Override
public void onError(@NonNull Throwable e) {
}
@Override
public void onComplete() {
//3
Log.e(TAG,"onComplete");
}
});
E/CaoZuoFuActivity: onSubscribe
E/CaoZuoFuActivity: onNext收到：1
E/CaoZuoFuActivity: onNext收到：2
E/CaoZuoFuActivity: onNext收到：3
E/CaoZuoFuActivity: onComplete
三.延遲創(chuàng)建
1.defer()
語法：
Observable.defer(new Callable<ObservableSource<? extends Integer>)
注意事項：
（1）只有當被觀察者注冊時苛蒲，才會創(chuàng)建被觀察者
Integer i = 5;
Observable<Integer> observable = Observable.defer(new Callable<ObservableSource<? extends Integer>>() {
@Override
public ObservableSource<? extends Integer> call() throws Exception {
return Observable.just(i);
}
});
i = 1;
observable.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
}
@Override
public void onNext(@NonNull Integer integer) {
Log.e(TAG, "收到：" + integer);
}
@Override
public void onError(@NonNull Throwable e) {
}
@Override
public void onComplete() {
}
});
收到：1
2.timer()
語法：
Observable.timer(2, TimeUnit.SECONDS).subscribe()
注意事項：
（1）延遲指定時間后，發(fā)送1個0滥玷。相當于延遲加載多少執(zhí)行，執(zhí)行某個事情
（2）timer操作符默認運行在一個新線程上
Observable.timer(2, TimeUnit.SECONDS).subscribe(new Observer<Long>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
}
@Override
public void onNext(@NonNull Long aLong) {
Log.e(TAG, "接收到了事件"+ aLong );
}
@Override
public void onError(@NonNull Throwable e) {
}
@Override
public void onComplete() {
}
});
E/TimeActivity: 接收到了事件0
（時間為2秒后）
三.操作符
1.map()
定義：
將發(fā)送的事件類型轉(zhuǎn)換為指定的類型
語法：
map(new Function<轉(zhuǎn)換前類型, 轉(zhuǎn)換后類型>()
注意事項：
（1）map操作符中重寫的apply方法锌杀，返回要寫裙盾，不能使用默認的null
Observable.create(new ObservableOnSubscribe<Integer>() {
// 1. 被觀察者發(fā)送事件 = 參數(shù)為整型 = 1、2择诈、3
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
}
// 2. 使用Map變換操作符中的Function函數(shù)對被觀察者發(fā)送的事件進行統(tǒng)一變換：整型變換成字符串類型
}).map(new Function<Integer, String>() {
@Override
public String apply(Integer integer) throws Exception {
return integer+"轉(zhuǎn)換后" ;
}
}).subscribe(new Consumer<String>() {
// 3. 觀察者接收事件時辟犀，是接收到變換后的事件 = 字符串類型
@Override
public void accept(String s) throws Exception {
Log.e(TAG, s);
}
});
E/MapActivity: 1轉(zhuǎn)換后
E/MapActivity: 2轉(zhuǎn)換后
E/MapActivity: 3轉(zhuǎn)換后
2.flatMap()
定義：
將發(fā)送的事件進行拆分俏竞，每個拆分后的事件再單獨轉(zhuǎn)換，再合并成新的事件堂竟。所有新的事件再組合成新的事件魂毁，發(fā)送給觀察者
語法：
flatMap(new Function<Integer, ObservableSource<String>>()
注意事項：
（1）新合并后的事件是無序的，不與拆分前順序一致
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
}
}).flatMap(new Function<Integer, ObservableSource<String>>() {
@Override
public ObservableSource<String> apply(@NonNull Integer integer) throws Exception {
//先拆分出嘹，然后轉(zhuǎn)換為String席楚，然后組合成集合
final List<String> list = new ArrayList<>();
for (int i = 0; i < 3; i++) {
list.add("我是事件 " + integer + "拆分后的子事件" + i);
}
return Observable.fromIterable(list);
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.e(TAG, s);
}
});
E/MapActivity: 我是事件 1拆分后的子事件0
E/MapActivity: 我是事件 1拆分后的子事件1
E/MapActivity: 我是事件 1拆分后的子事件2
E/MapActivity: 我是事件 2拆分后的子事件0
E/MapActivity: 我是事件 2拆分后的子事件1
E/MapActivity: 我是事件 2拆分后的子事件2
E/MapActivity: 我是事件 3拆分后的子事件0
E/MapActivity: 我是事件 3拆分后的子事件1
E/MapActivity: 我是事件 3拆分后的子事件2
2.ConcatMap()
用法和概念和flatMap()類似，區(qū)別在于concatMap是重新合并后的順序與原來順序一致税稼。
