前言
rxjava 可以很方便的进行线程切换, 那么rxjava是如何进行线程切换的呢?阅读本文可以了解下rxjava 是如何进行线程切换的及线程切换的影响点。
一个简单的代码:
Observable.create(new ObservableOnSubscribe() {
@Override
public void subscribe(ObservableEmitter e) throws Exception {
Log.d("WanRxjava ", "subscrib td ==" + Thread.currentThread().getName());
e.onNext("我在发送next");
e.onComplete();
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer() {
@Override
public void onSubscribe(Disposable d) {
Log.d("WanRxjava ", "onSubscribe td ==" + Thread.currentThread().getName());
}
@Override
public void onNext(String value) {
Log.d("WanRxjava ", "onNext td ==" + Thread.currentThread().getName());
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
Log.d("WanRxjava ", "onComplete td ==" + Thread.currentThread().getName());
}
});
如上代码,实现了线程切换和观察者被观察者绑定的逻辑。我们分四部分看上述代码逻辑create、subscribeOn、observeOn、subscribe
1.create
create 顾名思议是 创建被观察者,这里有一个参数是 ObservableOnSubscribe,这是个接口类,我们看下create 的源码:
@SchedulerSupport(SchedulerSupport.NONE)
public static Observable create(ObservableOnSubscribe source) {
ObjectHelper.requireNonNull(source, "source is null");
return RxJavaPlugins.onAssembly(new ObservableCreate(source));
}
将ObservableOnSubscribe 传入后 又调用了 new ObservableCreate(source)
public final class ObservableCreate extends Observable {
final ObservableOnSubscribe source;
public ObservableCreate(ObservableOnSubscribe source) {
this.source = source;
}
}
ObservableCreate 有一个变量是 source,这里只是将传入的ObservableOnSubscribe 赋值给source,也就是做了一层包装,然后返回。
2.subscribeOn
调用完create后返回了 ObservableCreate(Observable),然后继续调用subscribeOn,传入了一个变量 Schedulers.io()
@SchedulerSupport(SchedulerSupport.CUSTOM)
public final Observable subscribeOn(Scheduler scheduler) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
return RxJavaPlugins.onAssembly(new ObservableSubscribeOn(this, scheduler));
}
我们看到调用了new ObservableSubscribeOn(this, scheduler) 将自身和 scheduler 传入
public final class ObservableSubscribeOn extends AbstractObservableWithUpstream {
final Scheduler scheduler;
public ObservableSubscribeOn(ObservableSource source, Scheduler scheduler) {
super(source);
this.scheduler = scheduler;
}
}
ObservableSubscribeOn 将scheduler 和 create 返回的对象又包装了一层 返回ObservableSubscribeOn
3.observeOn
有一个参数是 Scheduler
@SchedulerSupport(SchedulerSupport.CUSTOM)
public final Observable observeOn(Scheduler scheduler) {
return observeOn(scheduler, false, bufferSize());
}
@SchedulerSupport(SchedulerSupport.CUSTOM)
public final Observable observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
ObjectHelper.verifyPositive(bufferSize, "bufferSize");
return RxJavaPlugins.onAssembly(new ObservableObserveOn(this, scheduler, delayError, bufferSize));
}
ObservableSubscribeOn(observable)又调用了observeOn,然后调用了new ObservableObserveOn(this, scheduler, delayError, bufferSize)
public final class ObservableObserveOn extends AbstractObservableWithUpstream {
final Scheduler scheduler;
final boolean delayError;
final int bufferSize;
public ObservableObserveOn(ObservableSource source, Scheduler scheduler, boolean delayError, int bufferSize) {
super(source);
this.scheduler = scheduler;
this.delayError = delayError;
this.bufferSize = bufferSize;
}
}
又是一个包装,将ObservableSubscribeOn 和 scheduler 包装成 ObservableObserveOn
4.subscribe
上述最后一步即调用ObservableObserveOn.subscribe,传入参数是一个 observer
//ObservableObserveOn.java
@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);
// can't call onError because no way to know if a Disposable has been set or not
// can't call onSubscribe because the call might have set a Subscription already
RxJavaPlugins.onError(e);
NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
npe.initCause(e);
throw npe;
}
}
可以看到调用subscribe 后调用了subscribeActual(observer);将observer 传入
我们看下 subscribeActual(observer)
//ObservableObserveOn.java
@Override
protected void subscribeActual(Observer super T> observer) {
if (scheduler instanceof TrampolineScheduler) {
source.subscribe(observer);
} else {
Scheduler.Worker w = scheduler.createWorker();
source.subscribe(new ObserveOnObserver(observer, w, delayError, bufferSize));
}
}
上面的if 先不管,主要看下下面的逻辑,调用了 scheduler.createWorker(),这个scheduler 是 observeOn 传入的,然后调用
new ObserveOnObserver(observer, w, delayError, bufferSize);将worker /observer 又做了一次包装。
//ObservableObserveOn 内部类
static final class ObserveOnObserver extends BasicIntQueueDisposable
implements Observer, Runnable {
private static final long serialVersionUID = 6576896619930983584L;
final Observer super T> actual;
final Scheduler.Worker worker;
final boolean delayError;
final int bufferSize;
SimpleQueue queue;
Disposable s;
Throwable error;
volatile boolean done;
volatile boolean cancelled;
int sourceMode;
boolean outputFused;
ObserveOnObserver(Observer super T> actual, Scheduler.Worker worker, boolean delayError, int bufferSize) {
this.actual = actual;
this.worker = worker;
this.delayError = delayError;
this.bufferSize = bufferSize;
}
@Override
public void onSubscribe(Disposable s) {
if (DisposableHelper.validate(this.s, s)) {
this.s = s;
if (s instanceof QueueDisposable) {
@SuppressWarnings("unchecked")
QueueDisposable qd = (QueueDisposable) s;
int m = qd.requestFusion(QueueDisposable.ANY | QueueDisposable.BOUNDARY);
if (m == QueueDisposable.SYNC) {
sourceMode = m;
queue = qd;
done = true;
actual.onSubscribe(this);
schedule();
return;
}
if (m == QueueDisposable.ASYNC) {
sourceMode = m;
queue = qd;
actual.onSubscribe(this);
return;
}
}
queue = new SpscLinkedArrayQueue(bufferSize);
actual.onSubscribe(this);
}
}
@Override
public void onNext(T t) {
if (done) {
return;
}
if (sourceMode != QueueDisposable.ASYNC) {
queue.offer(t);
}
schedule();
}
@Override
public void onError(Throwable t) {
if (done) {
RxJavaPlugins.onError(t);
return;
}
error = t;
done = true;
schedule();
}
@Override
public void onComplete() {
if (done) {
return;
}
done = true;
schedule();
}
@Override
public void dispose() {
if (!cancelled) {
cancelled = true;
s.dispose();
worker.dispose();
if (getAndIncrement() == 0) {
queue.clear();
}
}
}
@Override
public boolean isDisposed() {
return cancelled;
}
void schedule() {
if (getAndIncrement() == 0) {
worker.schedule(this);
}
}
void drainNormal() {
int missed = 1;
final SimpleQueue q = queue;
final Observer super T> a = actual;
for (;;) {
if (checkTerminated(done, q.isEmpty(), a)) {
return;
}
for (;;) {
boolean d = done;
T v;
try {
v = q.poll();
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
s.dispose();
q.clear();
a.onError(ex);
return;
}
boolean empty = v == null;
if (checkTerminated(d, empty, a)) {
return;
}
if (empty) {
break;
}
a.onNext(v);
}
missed = addAndGet(-missed);
if (missed == 0) {
break;
}
}
}
void drainFused() {
int missed = 1;
for (;;) {
if (cancelled) {
return;
}
boolean d = done;
Throwable ex = error;
if (!delayError && d && ex != null) {
actual.onError(error);
worker.dispose();
return;
}
actual.onNext(null);
if (d) {
ex = error;
if (ex != null) {
actual.onError(ex);
} else {
actual.onComplete();
}
worker.dispose();
return;
}
missed = addAndGet(-missed);
if (missed == 0) {
break;
}
}
}
@Override
public void run() {
if (outputFused) {
drainFused();
} else {
drainNormal();
}
}
boolean checkTerminated(boolean d, boolean empty, Observer super T> a) {
if (cancelled) {
queue.clear();
return true;
}
if (d) {
Throwable e = error;
if (delayError) {
if (empty) {
if (e != null) {
a.onError(e);
} else {
a.onComplete();
}
worker.dispose();
return true;
}
} else {
if (e != null) {
queue.clear();
a.onError(e);
worker.dispose();
return true;
} else
if (empty) {
a.onComplete();
worker.dispose();
return true;
}
}
}
return false;
}
@Override
public int requestFusion(int mode) {
if ((mode & ASYNC) != 0) {
outputFused = true;
return ASYNC;
}
return NONE;
}
@Override
public T poll() throws Exception {
return queue.poll();
}
@Override
public void clear() {
queue.clear();
}
@Override
public boolean isEmpty() {
return queue.isEmpty();
}
}
包装完ObserveOnObserver后,调用了source.subscribe 这里的source 即ObservableSubscribeOn.subscribe,进而调用ObservableSubscribeOn.subscribeActual
//ObservableSubscribeOn.java
@Override
public void subscribeActual(final Observer super T> s) {
final SubscribeOnObserver parent = new scheduler(s);
s.onSubscribe(parent);
parent.setDisposable(scheduler.scheduleDirect(new Runnable() {
@Override
public void run() {
source.subscribe(parent);
}
}));
}
static final class SubscribeOnObserver extends AtomicReference implements Observer, Disposable {
private static final long serialVersionUID = 8094547886072529208L;
final Observer super T> actual;
final AtomicReference s;
SubscribeOnObserver(Observer super T> actual) {
this.actual = actual;
this.s = new AtomicReference();
}
@Override
public void onSubscribe(Disposable s) {
DisposableHelper.setOnce(this.s, s);
}
@Override
public void onNext(T t) {
actual.onNext(t);
}
@Override
public void onError(Throwable t) {
actual.onError(t);
}
@Override
public void onComplete() {
actual.onComplete();
}
@Override
public void dispose() {
DisposableHelper.dispose(s);
DisposableHelper.dispose(this);
}
@Override
public boolean isDisposed() {
return DisposableHelper.isDisposed(get());
}
void setDisposable(Disposable d) {
DisposableHelper.setOnce(this, d);
}
}
ObservableSubscribeOn.subscribeActual
首先将传入的观察者封装成 SubscribeOnObserver
然后触发了 onSubscribe,接着调用 scheduler.scheduleDirect(new Runnable() 这里的scheduler 是 subscribeOn 传入的
最后调用了 scheduler.setsetDisposable方法。
我们看到 run 的方法体即source.subscribe(parent);这里的source 即 ObservableCreate(ObservableOnSubscribe),传入了observer,然后调用 observer的OnNext 和 OnComplete 方法。
5.小结:
a. 调用Observer.OnSubscribe 方法是 不受线程调度影响的
b.subscribeOn 影响的是发送事件的线程
c.observerOn 影响的是观察者处理接受数据的线程,如果没有调用observeOn 则不会进行包装成 ObserveOnObserver,也就是说不会执行观察者的线程切换,和 发送者的线程一致
d.多次调用subscribeOn切换线程,每次都会new ObservableSubscribeOn,触发事件发送时会往上调用,也就是第一次调用的subscribeOn传入的线程 会执行发送事件,后面的线程切换无效
e.Observer.OnSubscribe 只会执行一次,因为调用DisposableHelper.setOnce(this.s, s)
f.处理完onComplete 或者onError 后就不会再发出事件,因为被观察者发送完这两个事件后 就会调用disposed
文章来源于互联网:Rxjava 线程切换原理
1、本站所有资源均从互联网上收集整理而来,仅供学习交流之用,因此不包含技术服务请大家谅解!
2、本站不提供任何实质性的付费和支付资源,所有需要积分下载的资源均为网站运营赞助费用或者线下劳务费用!
3、本站所有资源仅用于学习及研究使用,您必须在下载后的24小时内删除所下载资源,切勿用于商业用途,否则由此引发的法律纠纷及连带责任本站和发布者概不承担!
4、本站站内提供的所有可下载资源,本站保证未做任何负面改动(不包含修复bug和完善功能等正面优化或二次开发),但本站不保证资源的准确性、安全性和完整性,用户下载后自行斟酌,我们以交流学习为目的,并不是所有的源码都100%无错或无bug!如有链接无法下载、失效或广告,请联系客服处理!
5、本站资源除标明原创外均来自网络整理,版权归原作者或本站特约原创作者所有,如侵犯到您的合法权益,请立即告知本站,本站将及时予与删除并致以最深的歉意!
6、如果您也有好的资源或教程,您可以投稿发布,成功分享后有站币奖励和额外收入!
7、如果您喜欢该资源,请支持官方正版资源,以得到更好的正版服务!
8、请您认真阅读上述内容,注册本站用户或下载本站资源即您同意上述内容!
原文链接:https://www.dandroid.cn/21135,转载请注明出处。
评论0