- 第一步：创建被观察者（observable）
Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onNext(4);
emitter.onComplete();
//ObservableEmitter为事件的发射器
}
});
- 第二步 创建观察者（observer）
Observer<Integer> observer = new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d("a","subscribe");
}
@Override
public void onNext(Integer integer) {
Log.d("a",""+integer);
}
@Override
public void onError(Throwable e) {
Log.d("a","error");
}
@Override
public void onComplete() {
Log.d("a","complete");
}
};
-
被观察者与观察者建立联系
observable.subscribe(observer);
Rxjava的链式操作
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onComplete();
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "subscribe");
}
@Override
public void onNext(Integer value) {
Log.d(TAG, "" + value);
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "error");
}
@Override
public void onComplete() {
Log.d(TAG, "complete");
}
});
-
RxJava的异步和链式编程
异步操作需要调用observeOn(AndroidSchedulers.mainThread())，observeOn是事件回调的线程，AndroidSchedulers.mainThread()一看就知道是主线程，
subscribeOn(Schedulers.io())，subscribeOn是事件执行的线程，Schedulers.io()是子线程，这里也可以用Schedulers.newThread()，只不过io线程可以重用空闲的线程，因此多数情况下 io()比newThread() 更有效率。
简单的来说, subscribeOn() 指定的是上游发送事件的线程, observeOn() 指定的是下游接收事件的线程.
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("连载1");
emitter.onNext("连载2");
emitter.onNext("连载3");
emitter.onComplete();
}
})
.observeOn(AndroidSchedulers.mainThread())//回调在主线程
.subscribeOn(Schedulers.io())//执行在io线程
.subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
Log.e(TAG,"onSubscribe");
}
@Override
public void onNext(String value) {
Log.e(TAG,"onNext:"+value);
}
@Override
public void onError(Throwable e) {
Log.e(TAG,"onError="+e.getMessage());
}
@Override
public void onComplete() {
Log.e(TAG,"onComplete()");
}
});
- 异步的举例2
没有任何嵌套，逻辑依然简洁
Observable.create(new ObservableOnSubscribe<Drawable>() {
@Override
public void subscribe(ObservableEmitter<Drawable> emitter) throws Exception {
for (int i=0;i<drawableRes.length;i++){
Drawable drawable=getResources().getDrawable(drawableRes[i]);
//第6个图片延时3秒后架子
if (i==5){
sleep(3000);
}
//复制第7张图片到sd卡
if (i==6){
Bitmap bitmap=((BitmapDrawable)drawable).getBitmap();
saveBitmap(bitmap,"test.png", Bitmap.CompressFormat.PNG);
}
//上传到网络
if (i==7){
updateIcon(drawable);
}
emitter.onNext(drawable);
}
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<Drawable>() {
@Override
public void accept(Drawable drawable) throws Exception {
//回调后在UI界面上展示出来
}
});
- 关于上下游所在线程的问题
多次指定上游的线程只有第一次指定的有效, 也就是说多次调用subscribeOn() 只有第一次的有效, 其余的会被忽略.
多次指定下游的线程是可以的, 也就是说每调用一次observeOn() , 下游的线程就会切换一次.
举个例子:
observable.subscribeOn(Schedulers.newThread())
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.observeOn(Schedulers.io())
.subscribe(consumer);
-
操作符
- Map
Observable.create(new ObservableOnSubscribe<Integer>() { @Override public void subscribe(ObservableEmitter<Integer> emitter) throws Exception { emitter.onNext(1); emitter.onNext(2); emitter.onNext(3); } }).map(new Function<Integer, String>() { @Override public String apply(Integer integer) throws Exception { return "This is result " + integer; } }).subscribe(new Consumer<String>() { @Override public void accept(String s) throws Exception { Log.d(TAG, s); } });
将上游的Integer类型数据转化为下游的字符串类型 - Map
flatMap
下游输出数据无序性concatMap 下游数据数据有序
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
}
}).concatMap(new Function<Integer, ObservableSource<String>>() {
@Override
public ObservableSource<String> apply(Integer integer) throws Exception {
final List<String> list = new ArrayList<>();
for (int i = 0; i < 3; i++) {
list.add("I am value " + integer);
}
return Observable.fromIterable(list).delay(10,TimeUnit.MILLISECONDS);
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.d(TAG, s);
}
});
- zip(将两个上游按照一定形式合并)同时可以结合切换线程（或者说指定线程）
Observable<Integer> observable1 = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
Log.d(TAG, "emit 1");
emitter.onNext(1);
Thread.sleep(1000);
Log.d(TAG, "emit 2");
emitter.onNext(2);
Thread.sleep(1000);
Log.d(TAG, "emit 3");
emitter.onNext(3);
Thread.sleep(1000);
Log.d(TAG, "emit 4");
emitter.onNext(4);
Thread.sleep(1000);
Log.d(TAG, "emit complete1");
emitter.onComplete();
}
});
Observable<String> observable2 = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
Log.d(TAG, "emit A");
emitter.onNext("A");
Thread.sleep(1000);
Log.d(TAG, "emit B");
emitter.onNext("B");
Thread.sleep(1000);
Log.d(TAG, "emit C");
emitter.onNext("C");
Thread.sleep(1000);
Log.d(TAG, "emit complete2");
emitter.onComplete();
}
});
Observable.zip(observable1, observable2, new BiFunction<Integer, String, String>() {
@Override
public String apply(Integer integer, String s) throws Exception {
return integer + s;
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "onSubscribe");
}
@Override
public void onNext(String value) {
Log.d(TAG, "onNext: " + value);
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError");
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete");
}
});
- sample
每隔指定的时间就从上游中取出一个事件发送给下游.但是这个方法有个缺点, 就是丢失了大部分的事件.
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
for (int i = 0; ; i++) {
emitter.onNext(i);
}
}
}).subscribeOn(Schedulers.io())
.sample(2, TimeUnit.SECONDS) //sample取样
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.d(TAG, "" + integer);
}
});
- interval操作符发送Long型的事件, 从0开始, 每隔指定的时间就把数字加1并发送出来
- onBackpressureBuffer()
onBackpressureDrop()
onBackpressureLatest()
Flowable.interval(1, TimeUnit.MICROSECONDS).onBackpressureDrop()
- 零碎知识点
- 上游可以发送无限个onNext, 下游也可以接收无限个onNext.
- 当上游发送了一个onComplete后, 上游onComplete之后的事件将会继续发送, 而下游收到onComplete事件之后将不再继续接收事件.
- 当上游发送了一个onError后, 上游onError之后的事件将继续发送, 而下游收到onError事件之后将不再继续接收事件.
- 上游可以不发送onComplete或onError.
- 最为关键的是onComplete和onError必须唯一并且互斥, 即不能发多个onComplete, 也不能发多个onError, 也不能先发一个onComplete, 然后再发一个onError, 反之亦然
