RxJava
提到Rxjava最多人都是用来处理，线程调度，回调地狱，加上Retrofit又支持Rxjava，所以大部分开发者都只会在请求网络和需要切换线程的时候用到Rxjava，其实它有一个最重要的特性，它可以让数据的流向更加直观，代码更清晰。
举个栗子
比如说一个庞大的项目，一个事件传递的整个过程可能要经历很多方法，方法套方法，每个方法的位置七零八落，一个个方法跳进去看,跳过去跳过来很容易把脑袋弄晕，不够直观。但是Rxjava可以把所有逻辑用链式加闭包的方式呈现，做了哪些操作，谁在前谁在后非常直观，逻辑清晰，维护就会非常轻松。就算不是你写的你也可以很快的了解，你可以把它看作一条河流，整个过程就是对里面的水流做进行加工。懂了这个特性我们才知道在复杂的逻辑中运用Rxjava是多么的重要。
Gradle添加依赖
要在Android中使用RxJava2, 先添加Gradle配置。
compile "io.reactivex.rxjava2:rxjava:2.0.7"
compile "io.reactivex.rxjava2:rxandroid:2.0.1"
一个简单实例（分别创建被观察者，观察者和两种对象进行关联）
//创建被观察者
Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
//调用观察者的回调
emitter.onNext("我是");
emitter.onNext("RxJava");
emitter.onNext("简单示例");
emitter.onError(new Throwable("出错了"));
emitter.onComplete();
}
});
//创建观察者
Observer<String> observer = new Observer<String>() {
@Override
public void onError(Throwable e) {
Log.e(TAG,e.getMessage());
}
@Override
public void onComplete() {
Log.e(TAG,"onCompleted");
}
//onSubscribe()方法是最先调用的
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG,"subscribe");
}
@Override
public void onNext(String s) {
Log.d(TAG,s);
}
};
//注册，将观察者和被观察者关联，将会触发OnSubscribe.call方法
observable.subscribe(observer);
运行结果：
Observable是被观察者，创建后传入一个OnSubscribe对象，当Observable（观察者）调用subscribe进行注册观察者时，OnSubscribe的call方法会触发。
ObservableEmitter： Emitter 是发射器的意思，它可以发出三种类型的事件，与之对应的。
Observer有三个回调方法：
- onNext：接受到一个事件
- onCompleted：接受完事件后调用，只会调用一次
- onError ：发生错误时调用，并停止接受事件，调用一次
注：onCompleted和onError不会同时调用，只会调用其中之一
另一个简单实例（访问网络时，通过RxJava来进行快捷的主线程和子线程切换）
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
final TextView tv_return = (TextView) findViewById(R.id.tv_return);
//创建被观察者
Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
Log.d(TAG, "Observer thread is :" + Thread.currentThread().getName());
emitter.onNext(getResponse());
}
});
//创建观察者
Consumer<String> consumer = new Consumer<String>() {
@Override
public void accept(String mResponse) throws Exception {
Log.d(TAG, "Observer thread is :" + Thread.currentThread().getName());
tv_return.setText(mResponse);
}
};
//subscribeOn() 指定的是发送事件的线程, observeOn() 指定的是接收事件的线程.
observable.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(consumer);
}
//使用okhttp访问网上提供的接口，由于是同步get请求，需要在子线程进行
private String getResponse() {
String url = "http://v.juhe.cn/weather/index?cityname=%E6%9D%AD%E5%B7%9E&dtype=&format=&key=7970495dbf33839562c9d496156e13cc";
OkHttpClient client = new OkHttpClient();
Request request = new Request.Builder()
.url(url)
.build();
Response response;
try {
response = client.newCall(request).execute();
return response.body().string();
} catch (IOException e) {
return "error";
}
}
在RxJava中, 已经内置了很多线程选项供我们选择, 例如有
- Schedulers.io() 代表io操作的线程, 通常用于网络,读写文件等io密集型的操作
- Schedulers.computation() 代表CPU计算密集型的操作, 例如需要大量计算的操作
- Schedulers.newThread() 代表一个常规的新线程
- AndroidSchedulers.mainThread() 代表Android的主线程
这些内置的Scheduler已经足够满足我们开发的需求， 因此我们应该使用内置的这些选项，在RxJava内部使用的是线程池来维护这些线程，所有效率也比较高。
第三个简单实例（类型变换操作）
RxJava提供了类型变换操作，像上面的发送网络请求事件，得到网络请求数据，传递和接收的都是字符串。
而RxJava中的map操作符可以对事件进行处理，比如发送网络请求事件，通过map处理后，可以接收到网络请求的字符串的字节数，如下所示。
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
final TextView tv_return = (TextView) findViewById(R.id.tv_return);
//创建被观察者
Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
Log.d(TAG, "Observer thread is :" + Thread.currentThread().getName());
emitter.onNext(getResponse());
}
//通过map操作符对数据进行中间处理
}).map(new Function<String, Integer>() {
@Override
public Integer apply(@NonNull String response) throws Exception {
return response.length();
}
});
//创建观察者
Consumer<Integer> consumer = new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.d(TAG, "Observer thread is :" + Thread.currentThread().getName());
tv_return.setText("字数："+integer);
}
};
//绑定，指定线程
observable.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(consumer);
}
第四个简单实例（链式操作）
而RxJava一个引以为豪的地方就是它的链式操作了，可以把被观察者和观察者串起来。
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
final TextView tv_return = (TextView) findViewById(R.id.tv_return);
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
Log.d(TAG, "Observer thread is :" + Thread.currentThread().getName());
emitter.onNext(getResponse());
}
}).map(new Function<String, Integer>() {
@Override
public Integer apply(@NonNull String response) throws Exception {
return response.length();
}
}).subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<Integer>() {
@Override
public void accept(@NonNull Integer integer) throws Exception {
Log.d(TAG, "Observer thread is :" + Thread.currentThread().getName());
tv_return.setText("字数：" + integer);
}
});
}
这里是项目地址 其实也没啥东西
喜欢的同学可以点个赞
参考
https://xiaobailong24.me/2017/03/18/Android-RxJava2.x/
http://www.jianshu.com/p/6fd8640046f1
网友评论