使用
首先从代码层面来分析RxJava的每一步到底干了什么。
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Throwable {
emitter.onNext("A");
}
}).map(new Function<String, Bitmap>() {
@Override
public Bitmap apply(String s) throws Throwable {
return null;
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<Bitmap>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
disposable = d;
}
@Override
public void onNext(@NonNull Bitmap bitmap) {
}
@Override
public void onError(@NonNull Throwable e) {
}
@Override
public void onComplete() {
}
});
源码分析
new ObservableOnSubscribe
在这里ObservableOnSubscribe就是我们的被观察者
public interface ObservableOnSubscribe<@NonNull T> {
/**
* Called for each {@link Observer} that subscribes.
* @param emitter the safe emitter instance, never {@code null}
* @throws Throwable on error
*/
void subscribe(@NonNull ObservableEmitter<T> emitter) throws Throwable;
}
Observable.create
调用Observable.create的时候将被观察者传了进来并且创建了对象
public static <T> Observable<T> create(@NonNull ObservableOnSubscribe<T> source) {
Objects.requireNonNull(source, "source is null");
return RxJavaPlugins.onAssembly(new ObservableCreate<>(source));
}
Hook
hook的含义就是在函数执行的过程中，我们有一个钩子函数，可以优先执行我们的代码，然后再接着执行。
上面的方法中调用了RxJavaPlugins.onAssembly函数
RxJavaPlugins.onAssembly(new ObservableCreate<>(source))
/*在这里什么都没有做，直接将source返回了，所以我们在这里可以给
onObservableAssembly进行赋值，通过setOnObservableAssembly方法。
*/
public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {
Function<? super Observable, ? extends Observable> f = onObservableAssembly;
if (f != null) {
return apply(f, source);
}
return source;
}
//如果onObservableAssembly有值的话会执行apply方法
static <T, R> R apply(@NonNull Function<T, R> f, @NonNull T t) {
try {
//这里会执行我们的hook方法，将被观察者传递进来，在我们实现的hook方法中返回观察者
return f.apply(t);
} catch (Throwable ex) {
throw ExceptionHelper.wrapOrThrow(ex);
}
}
//自定义hook方法实现
RxJavaPlugins.setOnObservableAssembly(new Function<Observable, Observable>() {
@Override
public Observable apply(Observable observable) throws Throwable {
//为了不影响代码流程，我们需要将observable返回
return observable;
}
});
map(new Function<String, Bitmap>()
上面我们获取到了对象，再调用map肯定是调用了
对象的map方法,由于他继承自Observable，所以调用的还是Observable的map方法。
这里又生成了对象
public final <R> Observable<R> map(@NonNull Function<? super T, ? extends R> mapper) {
Objects.requireNonNull(mapper, "mapper is null");
return RxJavaPlugins.onAssembly(new ObservableMap<>(this, mapper));
}
subscribeOn
对象. subscribeOn
public final Observable<T> subscribeOn(@NonNull Scheduler scheduler) {
Objects.requireNonNull(scheduler, "scheduler is null");
return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<>(this, scheduler));
}
这里传入了一个Scheduler对象Schedulers.io()，并且创建了ObservableSubscribeOn对象
schedulers.io调用过程.png
最终拿到了一个线程池。
observeOn
AndroidSchedulers.mainThread()
生成了一个主线程Handler.png
observeOn创建了ObservableObserveOn对象
new Observer<Bitmap>()
创建了一个观察者
public interface Observer<@NonNull T> {
/**
* Provides the {@link Observer} with the means of cancelling (disposing) the
* connection (channel) with the {@link Observable} in both
* synchronous (from within {@link #onNext(Object)}) and asynchronous manner.
* @param d the {@link Disposable} instance whose {@link Disposable#dispose()} can
* be called anytime to cancel the connection
* @since 2.0
*/
void onSubscribe(@NonNull Disposable d);
/**
* Provides the {@link Observer} with a new item to observe.
* <p>
* The {@link Observable} may call this method 0 or more times.
* <p>
* The {@code Observable} will not call this method again after it calls either {@link #onComplete} or
* {@link #onError}.
*
* @param t
* the item emitted by the Observable
*/
void onNext(@NonNull T t);
/**
* Notifies the {@link Observer} that the {@link Observable} has experienced an error condition.
* <p>
* If the {@code Observable} calls this method, it will not thereafter call {@link #onNext} or
* {@link #onComplete}.
*
* @param e
* the exception encountered by the Observable
*/
void onError(@NonNull Throwable e);
/**
* Notifies the {@link Observer} that the {@link Observable} has finished sending push-based notifications.
* <p>
* The {@code Observable} will not call this method if it calls {@link #onError}.
*/
void onComplete();
}
层次图
分析到这里就有了代码嵌套层次了
嵌套层次.png
subscribe
subscribe方法内部调用了subscribeActual(observer);那我们可以看到调用的就是ObservableObserveOn的subscribeActual方法
@Override
protected void subscribeActual(Observer<? super T> observer) {
//主线程的HandlerScheduler明显不是TrampolineScheduler的子类
if (scheduler instanceof TrampolineScheduler) {
source.subscribe(observer);
} else {
Scheduler.Worker w = scheduler.createWorker();
source.subscribe(new ObserveOnObserver<>(observer, w, delayError, bufferSize));
}
}
//调用了createWorker
@Override
public Worker createWorker() {
return new HandlerWorker(handler, async);
}
/*source.subscribe(new ObserveOnObserver<>(observer, w, delayError, bufferSize));
这里的source就是$\color{#FF0000}{ObservableSubscribeOn}$对象，因为调用observeOn方法的是ObservableSubscribeOn，并且他将自己作为source传了进去
*/
public final Observable<T> observeOn(@NonNull Scheduler scheduler, boolean delayError, int bufferSize) {
Objects.requireNonNull(scheduler, "scheduler is null");
ObjectHelper.verifyPositive(bufferSize, "bufferSize");
return RxJavaPlugins.onAssembly(new ObservableObserveOn<>(this, scheduler, delayError, bufferSize));
}
new ObserveOnObserver
在上面的subscribeActual方法我们可以看到他创建了一个新的ObserveOnObserver对象
//这里接收了一个observer对象，而这个observer正是我们的观察者，也就是在这里对观察者又进行了一次封装
new ObserveOnObserver<>(observer, w, delayError, bufferSize)
source.subscribe，source就是ObservableSubscribeOn
//这里SubscribeOnObserver<T> parent = new SubscribeOnObserver<>(observer);又对observer封装了一层，并且调用了onSubscribe，最终会调到我们自己的观察者的onSubscribe方法
@Override
public void subscribeActual(final Observer<? super T> observer) {
final SubscribeOnObserver<T> parent = new SubscribeOnObserver<>(observer);
observer.onSubscribe(parent);
//将观察者放入线程池中执行，下面的SubscribeTask可以看到在子线程里面执行了source.subscribe(parent);
parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
}
final class SubscribeTask implements Runnable {
private final SubscribeOnObserver<T> parent;
SubscribeTask(SubscribeOnObserver<T> parent) {
this.parent = parent;
}
@Override
public void run() {
//执行了上一步的subscribe方法
source.subscribe(parent);
}
}
对观察者进行封装图
观察者封装图.png
一层一层向上调用subscribe(@NonNull Observer<? super T> observer)，--->subscribeActual(observer);
当调用到了最上层的时候source就是我们的被观察者了
public static <T> Observable<T> create(@NonNull ObservableOnSubscribe<T> source) {
Objects.requireNonNull(source, "source is null");
//注意他这里传的可不是this
return RxJavaPlugins.onAssembly(new ObservableCreate<>(source));
}
create的source.png
在我们自定义的被观察者里面的subscribe调用了emitter.onNext("A");
从外层向内层调用onNext.png
当执行到以后一层也就是我们自己调用subscribe方法的时候
他会执行ObserveOnObserver的onNext
@Override
public void onNext(T t) {
if (done) {
return;
}
if (sourceMode != QueueDisposable.ASYNC) {
queue.offer(t);
}
schedule();
}
void schedule() {
if (getAndIncrement() == 0) {
//这里的work就是我们创建出来的HandlerWork
worker.schedule(this);
}
}
//HandlerWork类的schedule方法
public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
if (run == null) throw new NullPointerException("run == null");
if (unit == null) throw new NullPointerException("unit == null");
if (disposed) {
return Disposable.disposed();
}
run = RxJavaPlugins.onSchedule(run);
ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);
Message message = Message.obtain(handler, scheduled);
message.obj = this; // Used as token for batch disposal of this worker's runnables.
if (async) {
message.setAsynchronous(true);
}
//使用了主线程的Handler发送消息
handler.sendMessageDelayed(message, unit.toMillis(delay));
// Re-check disposed state for removing in case we were racing a call to dispose().
if (disposed) {
handler.removeCallbacks(scheduled);
return Disposable.disposed();
}
return scheduled;
}
handler的message中添加了该类的对象.png
当执行上面的run方法的时候
@Override
public void run() {
if (outputFused) {
drainFused();
} else {
drainNormal();
}
}
void drainNormal() {
int missed = 1;
final SimpleQueue<T> q = queue;
final Observer<? super T> a = downstream;
for (;;) {
if (checkTerminated(done, q.isEmpty(), a)) {
return;
}
for (;;) {
boolean d = done;
T v;
try {
v = q.poll();
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
disposed = true;
upstream.dispose();
q.clear();
a.onError(ex);
worker.dispose();
return;
}
boolean empty = v == null;
if (checkTerminated(d, empty, a)) {
return;
}
if (empty) {
break;
}
//*************************************************
//在这里最终调用了onNext的方法
a.onNext(v);
}
missed = addAndGet(-missed);
if (missed == 0) {
break;
}
}
}
总结
1.执行过程
- 从上到下对被观察者进行封装
- 从下往上对观察者进行封装
- 然后再从上往下执行onNext
思考一个问题？对观察者和被观察者的封装层数是一样的，那么是不是可以看成是1个被观察者对应一个观察者
对应关系.png
2.观察者的onSubscribe是在ObservableSubscribeOn的subscribeActual方法中执行的，此时还没有进行线程的切换，那么就是说在那个线程使用的RXJava那么观察者的onSubscribe方法就执行在哪个线程
- subscribeOn只会负责上层的线程调度，observeOn只有在执行onNext的时候才起作用，也就是下层的线程调度
4.使用handler(getMainLooper)来保证主线程操作
调用流程.png
自上而下(左边的流程)->自下而上(右边的流程+subscribe的调用)->自上而下(onNext的调用)
