参考的博文
看过好多rxjava的文章。。感觉这个最好本文例子基本都是copy里面的，只是为了方便自己记忆。当做工具
这个是rxjava 1.x版本的特别特别好的
例子 1 普通的订阅
image.png
/**
* 1 上下级订阅 链式操作
*/
//上游
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> observableEmitter) throws Exception {
observableEmitter.onNext("1");
observableEmitter.onNext("2");
observableEmitter.onNext("3");
observableEmitter.onComplete();
}
//订阅下游
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable disposable) {
}
@Override
public void onNext(String s) {
}
@Override
public void onError(Throwable throwable) {
}
@Override
public void onComplete() {
}
});
例子2 去除下游不需要实现的太多方法
image.png
/**
* 2 上下级订阅 简化下游实现方法太多
*/
//上游
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> observableEmitter) throws Exception {
observableEmitter.onNext("1");
observableEmitter.onNext("2");
observableEmitter.onNext("3");
observableEmitter.onComplete();
}
//订阅下游
}).subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
}
}, new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
}
});
例子3 上游代码在其他线程执行,(例如请求网络的操作),下游代码在主线程执行
image.png
/**
* 3 上游代码在其他线程执行,(例如请求网络的操作),下游代码在主线程执行
*/
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> observableEmitter) throws Exception {
observableEmitter.onNext("1");
observableEmitter.onNext("2");
observableEmitter.onNext("3");
observableEmitter.onComplete();
}
}).subscribeOn(Schedulers.newThread()).observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
}
});
例子4 rxjava和retrofit的联合使用
例子5 rxjava中map的使用
map--将observable中的原始的数据格式进行转换并且返回一个observable的类型例如开始传入的类型是String类型的。。经过中游(map)的转换。。下游就可以接收到int类型的数据而不是之前的String。
image.png
/**
* 5 map observable的格式转换
*/
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> observableEmitter) throws Exception {
observableEmitter.onNext("1");
observableEmitter.onNext("2");
observableEmitter.onNext("3");
observableEmitter.onComplete();
}
}).map(new Function<String, Integer>() {
@Override
public Integer apply(String s) throws Exception {
return Integer.parseInt(s);
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
}
});
例子6 rxjava中flatmap和concatMap的使用
flatmap 不一定会按照上游发送的顺序给下游发送。。concatMap会严格按照上游发送的顺序发送事件
image.png
/**
* 6 flapMap observable的格式转换
*/
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> observableEmitter) throws Exception {
observableEmitter.onNext("1");
observableEmitter.onNext("2");
observableEmitter.onNext("3");
observableEmitter.onComplete();
}
}).flatMap(new Function<String, ObservableSource<Integer>>() {
@Override
public ObservableSource<Integer> apply(String s) throws Exception {
return new ObservableSource<Integer>() {
@Override
public void subscribe(Observer<? super Integer> observer) {
observer.onNext(1);
observer.onNext(2);
observer.onNext(3);
observer.onComplete();
}
};
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
}
});
例子7 Observable 创建的其他快捷方式
/**
* 7 Observable 创建的其他快捷方式
*/
Observable.just("1","2").subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
}
});
Observable.fromIterable(Arrays.asList("1","2")).subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
}
});
Observable.fromArray("1","2","3").subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
}
});
例子8 zip操作符
/**
* 8 zip操作符 可用在例如多个网络操作，显示loading，当多个网络操作都结束的时候 在最后关闭loading、
*/
Observable observable1 = Observable.just(1);
Observable observable2 = Observable.just("1", "2");
Observable.zip(observable1, observable2, new BiFunction<Integer, String, String>() {
@Override
public String apply(Integer integer, String s) throws Exception {
return integer + s;
}
}).subscribe(new Consumer() {
@Override
public void accept(Object o) throws Exception {
}
});
例子9 filter 过滤操作符
/**
* 9 filter 过滤操作符
*/
Observable.just(1,2,3).filter(new Predicate<Integer>() {
@Override
public boolean test(Integer integer) throws Exception {
return true;
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
}
});
