Rxjava是什么
它做的工作就是异步，实现了简洁，清晰的异步工作
io.reactivex.Observable.create(new ObservableOnSubscribe<ProductResult>() {//使用.create()创建一个被观察者；
@Override
public void subscribe(ObservableEmitter<ProductResult> e) throws Exception {
/////////// 使用retrofit发起的网络请求
Retrofit retrofit=new Retrofit.Builder()
.baseUrl("http://suggest.taobao.com/")
.addConverterFactory(ScalarsConverterFactory.create())
.addConverterFactory(GsonConverterFactory.create())
.build();
GetRequest_Interface request=retrofit.create(GetRequest_Interface.class);
Map<String,String> maps=new HashMap<>();
maps.put("code","utf-8");
maps.put("q","卫衣");
Call<ProductResult> repos=request.listProduct((HashMap<String, String>) maps);
ProductResult productResult=repos.execute().body();
//////////
e.onNext(productResult);
}
}).subscribeOn(Schedulers.io())//绑定被观察者事件处理的线程
.observeOn(AndroidSchedulers.mainThread())//绑定观察者事件处理的线程
.subscribe(new Observer<ProductResult>() {
@Override
public void onSubscribe(Disposable d) {//这里用new Observer()创建了一个观察者，并且重写了事件的处理方法
}
@Override
public void onNext(ProductResult value) {
Log.i("mainThread", "onNext: "+android.os.Process.myTid());
Log.i("json", "onNext: "+value.toString());
textView1.setText(value.getResult().get(0).get(0));
textView2.setText(value.getResult().get(0).get(1));
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
上面这段代码，描述了发起一个网络请求，同时对UI进行更新。可以看到，我们既没有新建一个子线程用来发出请求，也没有使用handler跳转到主线程更新UI。也就是说，Rxjava帮我们做好了一切。
从上述代码的结构当中很容易看出，Rxjava是一个典型的观察者模式
- Observable 被观察者
它决定了事件什么时候被触发，以及触发怎样的事件。
io.reactivex.Observable<ProductResult> observable= io.reactivex.Observable.create(new ObservableOnSubscribe<ProductResult>() {
@Override
public void subscribe(ObservableEmitter<ProductResult> e) throws Exception {
//操作一
e.onNext(new ProductResult());
//操作二
e.onNext(new ProductResult());
//操作完成
e.onComplete();
}
});
.create()只是Observer创建的一种方式
io.reactivex.Observable<Integer> observable1= io.reactivex.Observable.just(1,2,3);
//等同于下面这段代码
io.reactivex.Observable observable1= io.reactivex.Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
e.onNext(1);
e.onNext(2);
e.onNext(3);
}
});
List<ProductResult> list=new ArrayList<>();
list.add(new ProductResult());
list.add(new ProductResult());
io.reactivex.Observable observable1= io.reactivex.Observable.fromArray(list);
//等同于如下方式
io.reactivex.Observable observable1= io.reactivex.Observable.create(new ObservableOnSubscribe<ProductResult>() {
@Override
public void subscribe(ObservableEmitter<ProductResult> e) throws Exception {
e.onNext(new ProductResult());
e.onNext(new ProductResult());
}
});
可以看到，创建Observable的过程中传入了一个ObservableOnSubscribe参数，从参数名就可以看出是作用在被观察者被订阅时的。
其中又有一个ObservableEmitter事件发射器，负责发送事件到Observer中响应的函数中进行处理。
综上所述，Observable在被Observer订阅后执行subscribe()方法，将事件发送到Observer中处理，是典型的观察者模式。
- Observer 观察者
它决定了事件触发后的执行的操作
Observer<ProductResult> observer=new Observer<ProductResult>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(ProductResult value) {
Log.i("ProductResult", "onNext: "+value.toString());
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
};
在Observable中发出的事件，在Observer中按顺序依次处理
- Subscribe 订阅
是连接Observer与Observable的重要部分，形式如下
observable.subscribe();
那么是如何实现Observer订阅时触发订阅事件的呢？
@SchedulerSupport(SchedulerSupport.NONE)
@Override
public final void subscribe(Observer<? super T> observer) {
ObjectHelper.requireNonNull(observer, "observer is null");
try {
observer = RxJavaPlugins.onSubscribe(this, observer);
ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");
subscribeActual(observer);//触发订阅事件的逻辑在这里面
} catch (NullPointerException e) { // NOPMD
throw e;
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
RxJavaPlugins.onError(e);
NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
npe.initCause(e);
throw npe;
}
}
@SchedulerSupport(SchedulerSupport.NONE)
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
ObjectHelper.requireNonNull(source, "source is null");
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
public ObservableCreate(ObservableOnSubscribe<T> source) {
this.source = source;
}
//以下就是上面方法的实现
@Override
protected void subscribeActual(Observer<? super T> observer) {
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
observer.onSubscribe(parent);//首先触发观察者的onSubscribe()方法
try {
source.subscribe(parent);//触发被观察内的subscribe()方法
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
以上就是触发订阅事件的实现，接下来我们看一下如何实现异步的线程控制，看看Rxjava帮我们做了哪些工作。
Scheduler 线程控制
Scheduler中的API
Schedler——调度，用于控制线程
- Scheduler.newThread()
每次都启用新线程，在新线程中执行操作 - Schedulers.single()
创建一个单例线程，以后的所有任务都在该线程运行 - Schedulers.trampoline()
在当前线程暂停正在执行的任务，先执行插入的任务 - Schedulers.io():
I/O 操作（读写文件、读写数据库、网络信息交互等）所使用的 Scheduler。行为模式和 newThread() 差不多，区别在于 io() 的内部实现是是用一个无数量上限的线程池，可以重用空闲的线程，因此多数情况下 io() 比 newThread() 更有效率。不要把计算工作放在 io() 中，可以避免创建不必要的线程。是最为常用的模式。
还有一个模式
- AndroidSchedulers.mainThread()
将操作运行在主线程，可以更新UI。
.subscribeOn(Schedulers.io())//绑定被观察者事件处理的线程为IO线程
.observeOn(AndroidSchedulers.mainThread())//绑定观察者事件处理的线程为主线程
网友评论