rxjava2.x原始碼學習隨筆

implementation 'io.reactivex.rxjava2:rxjava:2.2.0'
implementation 'io.reactivex.rxjava2:rxandroid:2.1.0'

Observable.create(new ObservableOnSubscribe<Response>() {
@Override
public void subscribe(ObservableEmitter<Response> e) throws Exception {
//獲取服務端的介面資料
Request.Builder builder = new Request.Builder()
.url("http://xxx.com")
.get();
Request request = builder.build();
Call call = new OkHttpClient().newCall(request);
Response response = call.execute();
e.onNext(response);
}
}).map(new Function<Response, Model>() {
@Override
public Model apply( Response response) throws Exception {
//將json資料轉化為對應的Model
if (response.isSuccessful()) {
ResponseBody body = response.body();
if (body != null) {
Log.e(TAG, "map:轉換前:" + response.body());
return new Gson().fromJson(body.string(), Model.class);
}
}
return null;
}
}).doOnNext(new Consumer<Model>() {
@Override
public void accept( Model s) throws Exception {
//對資料進行其他快取的處理
Log.e(TAG, "doOnNext: 儲存網路載入的資料：" + s.toString() + "\n");
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<Model>() {
@Override
public void accept(Model model) throws Exception {
//重新整理ui
Log.e(TAG, "成功重新整理介面:" + data.toString() + "\n");
}
}, new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
//進行失敗的異常提示
Log.e(TAG, "失敗處理異常：" + throwable.getMessage() + "\n");
}
});

public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
ObjectHelper.requireNonNull(source, "source is null");
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}

//構造方法會傳入ObservableOnSubscribe介面的引用，指定為該被觀察者的source。
 public ObservableCreate(ObservableOnSubscribe<T> source) {
this.source = source;
}

 //核心的subscribeActual 
@Override
protected void subscribeActual(Observer<? super T> observer) {
//建立了CreateEmitter發射器
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
observer.onSubscribe(parent);

try {
//執行了ObservableOnSubscribe的subscribe回撥方法，傳入了CreateEmitter物件
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}

 //持有下游的觀察者引用
CreateEmitter(Observer<? super T> observer) {
this.observer = observer;
}

@Override
public void onNext(T t) {
if (t == null) {
onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
return;
}
//如果沒有取消訂閱，則會執行下游的觀察者的onNext方法，達到鏈式呼叫的效果
if (!isDisposed()) {
observer.onNext(t);
}
}

@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
ObjectHelper.requireNonNull(mapper, "mapper is null");
return RxJavaPlugins.onAssembly(new ObservableMap<T, R>(this, mapper));
}

@Override
public void subscribeActual(Observer<? super U> t) {
//將上游的被觀察者訂閱MapObserver觀察者
source.subscribe(new MapObserver<T, U>(t, function));
}

//持有下游的觀察者和回撥函式mapper
MapObserver(Observer<? super U> actual, Function<? super T, ? extends U> mapper) {
super(actual);
this.mapper = mapper;
}

@Override
public void onNext(T t) {
if (done) {
return;
}

if (sourceMode != NONE) {
actual.onNext(null);
return;
}

U v;

try {
//map的核心執行程式碼，mapper.apply(t)會執行資料的轉換，並將轉換後的結果v繼續交由下游的觀察者執行
v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper function returned a null value.");
} catch (Throwable ex) {
fail(ex);
return;
}
//將轉換後的結果v繼續交由下游的觀察者執行
actual.onNext(v);
}

private Observable<T> doOnEach(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Action onAfterTerminate) {
ObjectHelper.requireNonNull(onNext, "onNext is null");
ObjectHelper.requireNonNull(onError, "onError is null");
ObjectHelper.requireNonNull(onComplete, "onComplete is null");
ObjectHelper.requireNonNull(onAfterTerminate, "onAfterTerminate is null");
return RxJavaPlugins.onAssembly(new ObservableDoOnEach<T>(this, onNext, onError, onComplete, onAfterTerminate));
}

@Override
public void subscribeActual(Observer<? super T> t) {
//例項化一個DoOnEachObserver的觀察者物件
source.subscribe(new DoOnEachObserver<T>(t, onNext, onError, onComplete, onAfterTerminate));
}

@Override
public void onNext(T t) {
if (done) {
return;
}
try {
//回撥accept方法
onNext.accept(t);
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
s.dispose();
onError(e);
return;
}
//繼續往下游呼叫觀察者的onNext
actual.onNext(t);
}

@CheckReturnValue
@SchedulerSupport(SchedulerSupport.CUSTOM)
public final Observable<T> subscribeOn(Scheduler scheduler) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));
}

 public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
super(source);
this.scheduler = scheduler;
}

@Override
public void subscribeActual(final Observer<? super T> s) {
//建立了SubscribeOnObserver的觀察者
final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);

s.onSubscribe(parent);

//這個是核心方法，呼叫了執行緒排程去的scheduleDirect方法，並傳入SubscribeTask任務
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() {
//執行上游被觀察的訂閱方法，這裡就是subscribeOn將上游的訂閱方法控制在scheduler指定執行緒執行的核心
source.subscribe(parent);
}
}

@Override
public void onNext(T t) {
actual.onNext(t);
}

 @CheckReturnValue
@SchedulerSupport(SchedulerSupport.CUSTOM)
public final Observable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
ObjectHelper.verifyPositive(bufferSize, "bufferSize");
return RxJavaPlugins.onAssembly(new ObservableObserveOn<T>(this, scheduler, delayError, bufferSize));
}

@Override
 protected void subscribeActual(Observer<? super T> observer) {
//TrampolineScheduler 如果是當前的執行緒 則直接將下游的觀察者與上游的被觀察訂閱
if (scheduler instanceof TrampolineScheduler) {
source.subscribe(observer);
} else {
//其他執行緒策略
Scheduler.Worker w = scheduler.createWorker();
//將執行緒策略的worker傳入ObserveOnObserver觀察者
source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
}
 }

@Override
public void onNext(T t) {
// 上一級的模式如果不是非同步的，加入佇列
if (sourceMode != QueueDisposable.ASYNC) {
queue.offer(t);
}
//進行執行緒排程
schedule();
}

void schedule() {
// 判斷當前正在執行的任務數目
if (getAndIncrement() == 0) {
worker.schedule(this);
}
}

@Override
public void run() {
//輸出結果是否融合
if (outputFused) {
drainFused();
} else {
drainNormal();
}
}

void drainNormal() {
int missed = 1;
final SimpleQueue<T> q = queue;
final Observer<? super T> a = actual;
//第一層迴圈
for (;;) {
// 檢查異常處理
if (checkTerminated(done, q.isEmpty(), a)) {
return;
}
//第二層迴圈
for (;;) {
boolean d = done;
T v;
//從佇列中獲取資料
v = q.poll();
boolean empty = v == null;
// 檢查異常
if (checkTerminated(d, empty, a)) {
return;
}
//如果沒有資料了，跳出
if (empty) {
break;
}
//執行下一次操作。
a.onNext(v);
}
//減掉執行的次數，並獲取剩於任務數量，然後再次迴圈
//直到獲取剩餘任務量為0，跳出迴圈
missed = addAndGet(-missed);
if (missed == 0) {
break;
}
}
}

@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;
}
}

@Override
public void subscribeActual(final Observer<? super T> s) {
final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);

s.onSubscribe(parent);

parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
}

@NonNull
 public Disposable scheduleDirect(@NonNull Runnable run) {
return scheduleDirect(run, 0L, TimeUnit.NANOSECONDS);
 }
 
@NonNull
 public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
//建立一個Worker物件
final Worker w = createWorker();


final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);

//DisposeTasky也是一個包裝類 繼承了Runnable介面
DisposeTask task = new DisposeTask(decoratedRun, w);

//這裡是關鍵的實現，執行了worker的schedule方法
w.schedule(task, delay, unit);

return task;
 }

static final class EventLoopWorker extends Scheduler.Worker {
private final CompositeDisposable tasks;
private final CachedWorkerPool pool;
private final ThreadWorker threadWorker;

final AtomicBoolean once = new AtomicBoolean();

EventLoopWorker(CachedWorkerPool pool) {
this.pool = pool;
this.tasks = new CompositeDisposable();
this.threadWorker = pool.get();
}

@Override
public void dispose() {
if (once.compareAndSet(false, true)) {
tasks.dispose();

// releasing the pool should be the last action
pool.release(threadWorker);
}
}

@Override
public boolean isDisposed() {
return once.get();
}

@NonNull
@Override
public Disposable schedule(@NonNull Runnable action, long delayTime, @NonNull TimeUnit unit) {
if (tasks.isDisposed()) {
// don't schedule, we are unsubscribed
return EmptyDisposable.INSTANCE;
}

return threadWorker.scheduleActual(action, delayTime, unit, tasks);
}
}

public ScheduledRunnable scheduleActual(final Runnable run, long delayTime, @NonNull TimeUnit unit, @Nullable DisposableContainer parent) {
Runnable decoratedRun = RxJavaPlugins.onSchedule(run);

ScheduledRunnable sr = new ScheduledRunnable(decoratedRun, parent);

if (parent != null) {
if (!parent.add(sr)) {
return sr;
}
}

Future<?> f;
try {
if (delayTime <= 0) {
f = executor.submit((Callable<Object>)sr);
} else {
f = executor.schedule((Callable<Object>)sr, delayTime, unit);
}
sr.setFuture(f);
} catch (RejectedExecutionException ex) {
if (parent != null) {
parent.remove(sr);
}
RxJavaPlugins.onError(ex);
}

return sr;
 }

EventLoopWorker(CachedWorkerPool pool) {
this.pool = pool;
this.tasks = new CompositeDisposable();
this.threadWorker = pool.get();
}

ThreadWorker get() {
if (allWorkers.isDisposed()) {
return SHUTDOWN_THREAD_WORKER;
}
//從已經release的work執行緒佇列中獲取快取
while (!expiringWorkerQueue.isEmpty()) {
ThreadWorker threadWorker = expiringWorkerQueue.poll();
//如果找到，返回複用的執行緒
if (threadWorker != null) {
return threadWorker;
}
}

// 如果沒有，則會建立一個新的ThreadWorker
ThreadWorker w = new ThreadWorker(threadFactory);
allWorkers.add(w);
return w;
}

@Override
protected void subscribeActual(Observer<? super T> observer) {
//如果指定當前執行緒 則不進行排程
if (scheduler instanceof TrampolineScheduler) {
source.subscribe(observer);
} else {
//建立Worker
Scheduler.Worker w = scheduler.createWorker();
//例項化ObserveOnObserver觀察者並傳入Worker
source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
}
}

@Override
public void onNext(T t) {
if (done) {
return;
}

if (sourceMode != QueueDisposable.ASYNC) {
queue.offer(t);
}
schedule();
}

void schedule() {
if (getAndIncrement() == 0) {
worker.schedule(this);
}
}

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 Disposables.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傳送訊息執行run介面
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 Disposables.disposed();
}

return scheduled;
}

private static final class MainHolder {
static final Scheduler DEFAULT
= new HandlerScheduler(new Handler(Looper.getMainLooper()), false);
}

public static <T> Flowable<T> create(FlowableOnSubscribe<T> source, BackpressureStrategy mode) {
ObjectHelper.requireNonNull(source, "source is null");
ObjectHelper.requireNonNull(mode, "mode is null");
return RxJavaPlugins.onAssembly(new FlowableCreate<T>(source, mode));
}

@Override
public void subscribeActual(Subscriber<? super T> t) {
BaseEmitter<T> emitter;
//根據不同的策略初始化不同的資料發射器
switch (backpressure) {
case MISSING: {
emitter = new MissingEmitter<T>(t);
break;
}
case ERROR: {
emitter = new ErrorAsyncEmitter<T>(t);
break;
}
case DROP: {
emitter = new DropAsyncEmitter<T>(t);
break;
}
case LATEST: {
emitter = new LatestAsyncEmitter<T>(t);
break;
}
default: {
emitter = new BufferAsyncEmitter<T>(t, bufferSize());
break;
}
}

t.onSubscribe(emitter);
try {
source.subscribe(emitter);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
emitter.onError(ex);
}
}

abstract static class BaseEmitter<T>
extends AtomicLong
implements FlowableEmitter<T>, Subscription {
private static final long serialVersionUID = 7326289992464377023L;

final Subscriber<? super T> actual;

final SequentialDisposable serial;

BaseEmitter(Subscriber<? super T> actual) {
this.actual = actual;
this.serial = new SequentialDisposable();
}
}

//這裡需要注意的是，Request最終會把n負責給AtomicLong
@Override
public final void request(long n) {
if (SubscriptionHelper.validate(n)) {
BackpressureHelper.add(this, n);
onRequested();
}
}
//省略其他若干方法

//長度是128
static final int BUFFER_SIZE;
static {
BUFFER_SIZE = Math.max(1, Integer.getInteger("rx2.buffer-size", 128));
}

 public static int bufferSize() {
return BUFFER_SIZE;
}

//在呼叫observeOn時，會將長度最後傳給emitter發射器，具體可以打斷的追蹤檢視呼叫鏈
public final Flowable<T> observeOn(Scheduler scheduler) {
return observeOn(scheduler, false, bufferSize());
}

@Override
public void onNext(T t) {
if (isCancelled()) {
return;
}

if (t != null) {
actual.onNext(t);
} else {
onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
return;
}

for (;;) {
long r = get();
if (r == 0L || compareAndSet(r, r - 1)) {
return;
}
}
}

@Override
public final void onNext(T t) {
if (isCancelled()) {
return;
}

if (t == null) {
onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
return;
}
//如果數量不為0則減1，通過上面的Request，可以知道get（）為Flowable的BUFFER_SIZE 128
if (get() != 0) {
actual.onNext(t);
BackpressureHelper.produced(this, 1);
} else {
//超出閾值 執行onOverflow
onOverflow();
}
}

static final class DropAsyncEmitter<T> extends NoOverflowBaseAsyncEmitter<T> {


private static final long serialVersionUID = 8360058422307496563L;

DropAsyncEmitter(Subscriber<? super T> actual) {
super(actual);
}

@Override
void onOverflow() {
// nothing to do
}

}

static final class ErrorAsyncEmitter<T> extends NoOverflowBaseAsyncEmitter<T> {


private static final long serialVersionUID = 338953216916120960L;

ErrorAsyncEmitter(Subscriber<? super T> actual) {
super(actual);
}

@Override
void onOverflow() {
onError(new MissingBackpressureException("create: could not emit value due to lack of requests"));
}

}

@Override
public void onNext(T t) {
if (done || isCancelled()) {
return;
}

if (t == null) {
onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
return;
}
//加入佇列 queue為SpscLinkedArrayQueue佇列
queue.offer(t);
//通知消費
drain();
}

@Override
public void onNext(T t) {
if (done || isCancelled()) {
return;
}

if (t == null) {
onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
return;
}
//進行覆蓋 queue為AtomicReference
queue.set(t);
//通知消費
drain();
}