好吧。在不知道该做什么的时候，就跟随这别人的脚步走吧；
学习资料 ：
给初学者的RxJava2.0教程
强烈推荐这一系列教程，简单易懂，分析得面面俱到，虽然，我特么还是不会实战。
废物！
讲道理，我这保守落后的大清少年，在这之前我连这玩意是干什么的都不知道，现在也是半懵逼状态 。
在使用RxJava2.0之前
compile 'io.reactivex.rxjava2:rxjava:2.0.1'
compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
说有个观察者，被观察者，他俩通过订阅来进行连接
Observer、Observable 、subscribe()
观察者通过这个订阅方法，告诉被观察者，你要做事情时啊，你就发送信息告诉我一声，
然后就是这样，被观察者呢，他就在自己进行操作的时候，发送信息给观察者，我要干事情喽。
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
e.onNext(1);
e.onNext(2);
e.onNext(3);
e.onComplete();
//onComplete之后的数据，被观察者照常发送，但是观察者并不会接收到。
e.onNext(4);
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.i("TAG-->", "subscribe");
}
@Override
public void onNext(Integer value) {
Log.i("TAG-->", value + "");
}
@Override
public void onError(Throwable e) {
Log.i("TAG-->", "error");
}
@Override
public void onComplete() {
Log.i("TAG-->", "complete");
}
});
还有RxJava可以很轻松的切换线程。
- Schedulers.io() 代表io操作的线程, 通常用于网络,读写文件等io密集型的操作
- Schedulers.computation() 代表CPU计算密集型的操作, 例如需要大量计算的操作
- Schedulers.newThread() 代表一个常规的新线程
- AndroidSchedulers.mainThread() 代表Android的主线程
//RxJava内置的线程调度器，可以轻松的切换线程；
Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
Log.i("TAG-->", "Observable thread is :" + Thread.currentThread().getName());
e.onNext(1);
}
});
Consumer consumer = new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.d("TAG-->", "Observer thread is :" + Thread.currentThread().getName());
}
};
// observable.subscribe(consumer);
observable.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(consumer);
通过map转换数据类型
//Map , 我们在上游发送integer类型数据，通过map ，将数据转换为 string ,
// 使得下游接受的数据为string
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
}
}).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.i("TAG-->", s);
}
});
zip函数整合数据
//打包数据 ，通过一个函数，将多个 Observable 发送的事件结合到一起，然后发送这些组合到一起的事件
Observable observable1 = Observable.create(new ObservableOnSubscribe() {
@Override
public void subscribe(ObservableEmitter e) throws Exception {
e.onNext(1);
e.onNext(2);
e.onNext(3);
e.onNext(4);
e.onComplete();
}
}).subscribeOn(Schedulers.io());
Observable observable2 = Observable.create(new ObservableOnSubscribe() {
@Override
public void subscribe(ObservableEmitter e) throws Exception {
e.onNext("A");
e.onNext("B");
e.onNext("C");
e.onComplete();
}
}).subscribeOn(Schedulers.io());
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() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Object value) {
Log.i("TAG-->", value + "");
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
网友评论