线程控制 —— Scheduler
默认情况下， RxJava 遵循的是线程不变的原则，即：在哪个线程调用 subscribe()，就在哪个线程生产事件；在哪个线程生产事件，就在哪个线程消费事件。如果需要切换线程，就需要用到 Scheduler （调度器）。
- Schedulers.immediate(): 直接在当前线程运行，相当于不指定线程。这是默认的 Scheduler。
- Schedulers.trampoline():加入当前线程队列，待上一个工作完成后执行。
- Schedulers.newThread(): 总是启用新线程，并在新线程执行操作。
- Schedulers.io(): I/O 操作（读写文件、读写数据库、网络信息交互等）所使用的 Scheduler。行为模式和 newThread() 差不多，区别在于 io() 的内部实现是是用一个无数量上限的线程池，可以重用空闲的线程，因此多数情况下 io() 比 newThread() 更有效率。不要把计算工作放在 io() 中，可以避免创建不必要的线程。
- Schedulers.computation(): 计算所使用的 Scheduler。这个计算指的是 CPU 密集型计算，即不会被 I/O 等操作限制性能的操作，例如图形的计算。这个 Scheduler 使用的固定的线程池，大小为 CPU 核数。不要把 I/O 操作放在 computation() 中，否则 I/O 操作的等待时间会浪费 CPU。
- Android 还有一个专用的 AndroidSchedulers.mainThread()，它指定的操作将在 Android 主线程运行。
可以使用 subscribeOn() 和 observeOn() 两个方法来对线程进行控制。
- subscribeOn(): 指定 subscribe() 所发生的线程，即 Observable.OnSubscribe 被激活时所处的线程。或者叫做事件产生的线程。
- observeOn(): 指定 Subscriber 所运行在的线程。或者叫做事件消费的线程。
Observable.just(1,2,3)
.observeOn(Schedulers.io())
.subscribeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.d(tag,integer);
}
});
先看Observable的observeOn方法
public final Observable<T> observeOn(Scheduler scheduler) {
if (this instanceof ScalarSynchronousObservable) { // just参数为1个的时候为真
return ((ScalarSynchronousObservable<T>)this).scalarScheduleOn(scheduler);
}
//创建operator并把scheduler赋值给成员变量，调用lift方法，参数为operator
//注意！！又是lift方法！
return lift(new OperatorObserveOn<T>(scheduler));
}
// OperatorObserveOn类
public final class OperatorObserveOn<T> implements Operator<T, T> {
private final Scheduler scheduler;
public OperatorObserveOn(Scheduler scheduler) {
this.scheduler = scheduler;
}
}
//Observable的lift方法
public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {
//创建一个新的Observable对象并返回
return new Observable<R>(new OnSubscribe<R>() {
@Override
public void call(Subscriber<? super R> o) {
try {
// 返回传入的opetator对象，并调用它的call方法，参数是监听该Observable的subscriber
// 这里observeOn(scheduler)的参数如果是当前线程直接返回监听的subscriber对象
//如果线程改变，将scheduler和subscriber对象封装成ObserveOnSubscriber返回
Subscriber<? super T> st = hook.onLift(operator).call(o);
try {
st.onStart(); //调用subscriber的onstart();
onSubscribe.call(st);//调用Observeable.onSubscribe.call(subscriber)方法。参数是call返回的ObserveOnSubscriber。即调用了ObserveOnSubscriber的onNext方法
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
st.onError(e);
}
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
o.onError(e);
}
}
});
}
//hook的onLift方法 返回传入的Operator对象
public <T, R> Operator<? extends R, ? super T> onLift(final Operator<? extends R, ? super T> lift) {
return lift;
}
//opetator的call方法，如果运行在当前线程直接返回subscriber，其他线程将scheduler和subscriber包装成ObserveOnSubscriber返回
public Subscriber<? super T> call(Subscriber<? super T> child) {
if (scheduler instanceof ImmediateScheduler) {//运行在当前线程
return child;
} else if (scheduler instanceof TrampolineScheduler) {//运行在当前线程
return child;
} else {
ObserveOnSubscriber<T> parent = new ObserveOnSubscriber<T>(scheduler, child);
parent.init();
return parent;
}
}
//ObserveOnSubscriber构造方法
public ObserveOnSubscriber(Scheduler scheduler, Subscriber<? super T> child) {
this.child = child;
this.recursiveScheduler = scheduler.createWorker();
if (UnsafeAccess.isUnsafeAvailable()) {
queue = new SpscArrayQueue<Object>(RxRingBuffer.SIZE);
} else {
queue = new SynchronizedQueue<Object>(RxRingBuffer.SIZE);
}
this.scheduledUnsubscribe = new ScheduledUnsubscribe(recursiveScheduler);
}
//ObserveOnSubscriber init方法
void init() {
child.add(scheduledUnsubscribe);
child.setProducer(new Producer() {
@Override
public void request(long n) {
BackpressureUtils.getAndAddRequest(requested, n);
schedule();
}
});
child.add(recursiveScheduler);
child.add(this);
}
//ObserveOnSubscriber 的onStart方法
@Override
public void onStart() {
// signal that this is an async operator capable of receiving this many
request(RxRingBuffer.SIZE);
}
//ObserveOnSubscriber 的onNext方法 调用schedule
public void onNext(final T t) {
if (isUnsubscribed()) {
return;
}
if (!queue.offer(on.next(t))) {
onError(new MissingBackpressureException());
return;
}
schedule();
}
protected void schedule() {
if (counter.getAndIncrement() == 0) {
recursiveScheduler.schedule(action); //最终在schedule线程调用action的call方法
}
}
final Action0 action = new Action0() {
@Override
public void call() {
pollQueue();
}
};
// 最终在schedule线程调用subscriber的onNext方法
void pollQueue() {
int emitted = 0;
final AtomicLong localRequested = this.requested;
final AtomicLong localCounter = this.counter;
do {
localCounter.set(1);
long produced = 0;
long r = localRequested.get();
for (;;) {
if (child.isUnsubscribed())
return;
Throwable error;
if (finished) {
if ((error = this.error) != null) {
// errors shortcut the queue so
// release the elements in the queue for gc
queue.clear();
child.onError(error);
return;
} else
if (queue.isEmpty()) {
child.onCompleted();
return;
}
}
if (r > 0) {
Object o = queue.poll();
if (o != null) {
child.onNext(on.getValue(o));
r--;
emitted++;
produced++;
} else {
break;
}
} else {
break;
}
}
if (produced > 0 && localRequested.get() != Long.MAX_VALUE) {
localRequested.addAndGet(-produced);
}
} while (localCounter.decrementAndGet() > 0);
if (emitted > 0) {
request(emitted);
}
}
subscribeOn
public final Observable<T> subscribeOn(Scheduler scheduler) {
if (this instanceof ScalarSynchronousObservable) {
return ((ScalarSynchronousObservable<T>)this).scalarScheduleOn(scheduler);
}
return nest().lift(new OperatorSubscribeOn<T>(scheduler)); // 熟悉的lift方法
}
public final Observable<Observable<T>> nest() {
return just(this);
}
public final static <T> Observable<T> just(final T value) {
return ScalarSynchronousObservable.create(value);
}
//Observable的lift方法
public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {
//创建一个新的Observable对象并返回
return new Observable<R>(new OnSubscribe<R>() {
@Override
public void call(Subscriber<? super R> o) {
try {
// 调用OperatorSubscribeOn的call方法
// 这里observeOn(scheduler)的参数如果是当前线程直接返回监听的subscriber对象
//如果线程改变，将scheduler和subscriber对象封装成ObserveOnSubscriber返回
Subscriber<? super T> st = hook.onLift(operator).call(o);
try {
st.onStart(); //调用subscriber的onstart();
onSubscribe.call(st);//调用Observeable.onSubscribe.call(subscriber)方法。参数是call返回的ObserveOnSubscriber
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
st.onError(e);
}
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
o.onError(e);
}
}
});
}
public class OperatorSubscribeOn<T> implements Operator<T, Observable<T>> {
private final Scheduler scheduler;
public OperatorSubscribeOn(Scheduler scheduler) {
this.scheduler = scheduler;
}
@Override
public Subscriber<? super Observable<T>> call(final Subscriber<? super T> subscriber) {
final Worker inner = scheduler.createWorker();
subscriber.add(inner);
return new Subscriber<Observable<T>>(subscriber) {
@Override
public void onCompleted() {
// ignore because this is a nested Observable and we expect only 1 Observable<T> emitted to onNext
}
@Override
public void onError(Throwable e) {
subscriber.onError(e);
}
@Override
public void onNext(final Observable<T> o) {
inner.schedule(new Action0() {
@Override
public void call() {
final Thread t = Thread.currentThread();
o.unsafeSubscribe(new Subscriber<T>(subscriber) {
@Override
public void onCompleted() {
subscriber.onCompleted();
}
@Override
public void onError(Throwable e) {
subscriber.onError(e);
}
@Override
public void onNext(T t) {
subscriber.onNext(t);
}
@Override
public void setProducer(final Producer producer) {
subscriber.setProducer(new Producer() {
@Override
public void request(final long n) {
if (Thread.currentThread() == t) {
// don't schedule if we're already on the thread (primarily for first setProducer call)
// see unit test 'testSetProducerSynchronousRequest' for more context on this
producer.request(n);
} else {
inner.schedule(new Action0() {
@Override
public void call() {
producer.request(n);
}
});
}
}
});
}
});
}
});
}
};
}
}
