前言正文总结

前言

上一节简单概述了RxJava的用法，下面来介绍下Rxjava2的线程控制

正文

首先，我们需要明确Rxjava执行操作的线程，通过下面这段代码可以看到

Subscriber<String> subscriber = new Subscriber<String>() { @Override public void onSubscribe(Subscription s) { s.request(Long.MAX_VALUE); } @Override public void onNext(String s) { Log.d(TAG, "Subscriber thread is : " + Thread.currentThread().getName()); Log.d(TAG, "Finish"); } @Override public void onError(Throwable t) { } @Override public void onComplete() { } }; Flowable<String> flowable = Flowable.create(new FlowableOnSubscribe<String>() { @Override public void subscribe(FlowableEmitter<String> e) throws Exception { Log.d(TAG, "Flowable thread is : " + Thread.currentThread().getName()); Log.d(TAG, "Turn to Subscriber"); } }, BackpressureStrategy.BUFFER); flowable.subscribe(subscriber);low able.subscribe(subscriber);

这段的输出结果是这样的:

D/Rxjava: Flowable thread is : main D/Rxjava: Turn to Subscriber D/Rxjava: Subscriber thread is : main D/Rxjava: Finish

这样就会有一个问题：如果我的操作中有网络请求，那么会直接在main线程中执行，这样的操作会导致报错。

因此，我们需要更改下代码：

Subscriber<String> subscriber = new Subscriber<String>() { @Override public void onSubscribe(Subscription s) { s.request(Long.MAX_VALUE); } @Override public void onNext(String s) { Log.d(TAG, "Subscriber thread is : " + Thread.currentThread().getName()); Log.d(TAG, "Finish"); } @Override public void onError(Throwable t) { } @Override public void onComplete() { } }; Flowable<String> flowable = Flowable.create(new FlowableOnSubscribe<String>() { @Override public void subscribe(FlowableEmitter<String> e) throws Exception { Log.d(TAG, "Flowable thread is : " + Thread.currentThread().getName()); Log.d(TAG, "Turn to Subscriber"); e.onNext("test"); } }, BackpressureStrategy.BUFFER); flowable.subscribeOn(AndroidSchedulers.mainThread()) .observeOn(Schedulers.io()) .subscribe(subscriber);

输出结果如下:

D/Rxjava: Flowable thread is : main D/Rxjava: Turn to Subscriber D/Rxjava: Subscriber thread is : RxCachedThreadScheduler-2 D/Rxjava: Finish

在RxJava中, 已经内置了很多线程选项供我们选择, 例如 - Schedulers.io() 代表io操作的线程,通常用于网络,读写文件等io密集型的操。 - Schedulers.computation() 代表CPU计算密集型的操作,例如需要大量计算的操作 - Schedulers.newThread() 代表一个常规的新线程 - Schedulers.single() 代表一个单线程。 - Schedulers.from(@NonNull Executor executor) 代表执行一个Executor - AndroidSchedulers.mainThread() 代表android的主线程。

这些内置的Scheduler已经足够满足我们开发的需求, 因此我们应该使用内置的这些选项, 在RxJava内部使用的是线程池来维护这些线程, 所有效率也比较高

另外多次调用subscribeOn() 只有第一次的有效, 其余的会被忽略. 但是每调用一次observeOn() , 线程就会切换一次.

总结

总而言之，RxJava2的线程控制是通过subscribeOn和observeOn来实现的，我们只需要调用相应的方法就可以实现线程的跳转，下一篇将是操作符的使用，运用操作符可以很轻松的实现一些特殊效果