rxjava2.x源码学习随笔
ReactiveX是Reactive Extensions的缩写，一般简写为Rx，最初是LINQ的一个扩展，由微软的架构师Erik Meijer领导的团队开发，在2012年11月开源，Rx是一个编程模型，目标是提供一致的编程接口，帮助开发者更方便的处理异步数据流，Rx库支持.NET、JavaScript和C++，Rx近几年越来越流行了，现在已经支持几乎全部的流行编程语言了，Rx的大部分语言库由ReactiveX这个组织负责维护，比较流行的有RxJava/RxJS/Rx.NET。
rxjava
rxjava是ReactiveX在java平台的一个实现。是一个编程模型，以观察者模式提供链式的接口调用，动态控制线程的切换，使得可以简便的处理异步数据流。
简介
Github: rxjava
中文文档： ReactiveX/RxJava文档中文版
官网： reactivex
特点
- 链式调用，使用简单
- 简化逻辑
- 灵活的线程调度
- 提供完善的数据操作符，功能强大
观察者模式
观察者模式定义对象间一种一对多的依赖关系，使得每当一个对象改变状态，则所以依赖于它的对象都会得到通知并被自动更新。rxjava的核心设计就是采用观察者模式。Observable是被观察者，Observer是观察者，通过subscribe方法进行订阅。
- 优点
观察者和被观察者之间是抽象解耦，应对业务变化
增强系统灵活性、可扩展性
具体代码示例可参考设计模式-观察者模式
- 缺点
在应用观察者模式时需要考虑一下开发效率和运行效率问题，程序中包括一个被观察者、多个观察者、开发和调试等内容会比较复杂，而且在Java中消息的通知默认是顺序执行，一个观察者卡顿，会影响整体的执行效率，在这种情况下，一般考虑采用异步的方式
rxjava怎么用？
gradle引入版本
implementation 'io.reactivex.rxjava2:rxjava:2.2.0' implementation 'io.reactivex.rxjava2:rxandroid:2.1.0'
接着举一个常用的rxjava使用的例子，我们在项目经常需要请求服务端接口，然后获取数据，将数据进行缓存，然后处理ui上的显示。示例的代码如下：
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");
}
});
本文主要对rxjava的源码进行梳理分析，关于rxjava操作符的使用，推荐参考中文的文档，以及下面的博文介绍。
这可能是最好的RxJava 2.x 教程（完结版）
rxjava核心执行流程是怎样？
rxjava主要是采用观察者模式进行设计，当执行相关的操作符是会生成新的Observable及Observer。Observable会持有上游被观察者，Observer会持有下游的观察者。当执行subscribe订阅方法的时候，通过持有上游的被观察者对象，会往上游逐步执行订阅方法。当执行到起始的被观察者回调方法时，如果执行ObservableEmitter的onNext方法时，由于Observer会持有下游的Observer对象，会逐步调用下游的onNext方法，直到最终subscribe传入的观察者实例。这是rxjava链式调用的核心执行流程。
当然rxjava还涉及到线程的调度及数据的背压处理，关于这些实现的原理会再后续进行梳理。但rxjava的链式调用的核心执行流程都是一致。下面我们将通过2个部分来梳理rxjava的核心执行流程，包含一些关键类的说明，及通过示例的代码相关的执行流程图进行梳理。
关键类功能说明
| 类 | 说明 |
|---|---|
| ObservableSource | 接口类，只有一个subscribe方法，参数是Observer对象 |
| Observer | 接口类，观察者。有onSubscribe、onNext、onError、onComplete方法 |
| Consumer | 接口有，观察者。只有一个accept方法，在被订阅时最终也会转换成Observer，设计这个类是为了简化调用 |
| Observable | 抽象类，继承了ObservableSource接口，操作符的实现都是继承与它。内部封装了大量的操作符调用方法，主要是有一个核心的抽象方法abstract void subscribeActual(Observer<? super T> observer)，用于实现相关的订阅分发逻辑。 |
| AbstractObservableWithUpstream | 继承于Observable，构造方法需要传入ObservableSource
|
| ObservableCreate | 继承于AbstractObservableWithUpstream，source为ObservableOnSubscribe。subscribeActual方法会实例化一个CreateEmitter对象，执行ObservableOnSubscribe的subscribe方法 |
| ObservableMap | 继承于AbstractObservableWithUpstream，订阅会新生产一个观察者MapObserver |
| MapObserver | ObservableMap的内部类，onNext方法会触发mapper.apply(t)回调，然后执行下游观察者的onNext方法 |
| ObservableDoOnEach | 继承于AbstractObservableWithUpstream，订阅会新生产一个观察者DoOnEachObserver |
| DoOnEachObserver | ObservableDoOnEach的内部类，onNext会执行onNext.accept(t)方法，然后执行下游观察者的onNext方法 |
| ObservableSubscribeOn | 继承于AbstractObservableWithUpstream，被观察者线程调度控制。subscribeActual会执行scheduler.scheduleDirect(new SubscribeTask(parent))，SubscribeTask的run方法会执行source.subscribe(parent)。ObservableSubscribeOn根据线程调度器的策略去执行上游的订阅方法实现。 |
| ObservableObserveOn | 继承于AbstractObservableWithUpstream，观察者线程调度控制。subscribeActual方法会判断scheduler是否为TrampolineScheduler。若是则执行下游的观察者，否会创建新的ObserveOnObserver，并传入schedule的work。 |
| ObserveOnObserver | ObservableObserveOn内部类，onNext会触发执行schedule（）方法，根据worker去控制下游观察者的回调线程 |
代码执行流程
首先我们根据上面demo例子，梳理出rxjava的简单执行流程，如下图：
通过流程图可知，rxjava当执行相关的操作符是会生成新的Observable及Observer。Observable会持有上游被观察者，Observer会持有下游的观察者。当执行subscribe订阅方法的时候，通过持有上游的被观察者对象，会往上游逐步执行订阅方法。当执行到起始的被观察者回调方法时，如果执行ObservableEmitter的onNext方法时，由于Observer会持有下游的Observer对象，会逐步调用下游的onNext方法，直到最终subscribe传入的观察者实例。
了解了rxjava大致的执行流程，下面我们来详细的看看源码的执行流程。首先还是先上一下整体的流程图，由于图片较大，建议结合上述的demo及rxjava的源码进行查看。
下面我们分配通过几个操作符来看看rxjava源码具体的实现。
create
create的操作符会返回一个ObservableCreate的被观察者。
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
ObjectHelper.requireNonNull(source, "source is null");
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
接下来看看ObservableCreate对象的关键实现代码，如下：
//构造方法会传入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);
}
}
当我们在业务代码执行了ObservableEmitter的onNext方法，我们看一下CreateEmitter的onNext的实现代码，如下：
//持有下游的观察者引用
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);
}
}
map
map的操作符会返回一个ObservableMap的被观察者。
@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));
}
接下来看看ObservableMap对象的关键实现代码，如下：
@Override
public void subscribeActual(Observer<? super U> t) {
//将上游的被观察者订阅MapObserver观察者
source.subscribe(new MapObserver<T, U>(t, function));
}
接下来主要看看MapObserver的onNext方法，该方法会在ObservableEmitter的onNext方法触发后被调用，如下：
//持有下游的观察者和回调函数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);
}
doOnNext
doOnNext的操作符会返回一个ObservableDoOnEach的被观察者。
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));
}
接下来看看ObservableDoOnEach对象的关键实现代码，如下：
@Override
public void subscribeActual(Observer<? super T> t) {
//实例化一个DoOnEachObserver的观察者对象
source.subscribe(new DoOnEachObserver<T>(t, onNext, onError, onComplete, onAfterTerminate));
}
这里核心我们还是要看DoOnEachObserver的onNext对于数据的处理，如下：
@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);
}
subscribeOn
subscribeOn的操作符会返回一个ObservableSubscribeOn的被观察者，并传入scheduler线程调度参数。
@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));
}
接下来看看ObservableSubscribeOn对象的关键实现代码，如下：
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)));
}
接下来我们看看SubscribeTask的实现，如下：
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);
}
}
最后看下SubscribeOnObserver的onNext方法，比较简单，直接执行下游观察者的onNext方法，如下：
@Override
public void onNext(T t) {
actual.onNext(t);
}
关于scheduler的具体实现，在后续的线程原理进行分析。这里我们只需要知道上游的被观察者的订阅在指定的scheduler线程策略中执行就可以了。
observerOn
observerOn 的操作符会返回一个ObservableObserveOn的被观察者，并传入scheduler线程调度参数。
@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));
}
接下来看看ObservableObserveOn对象的关键实现代码，如下：
@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));
}
}
接下来关键还是看ObserveOnObserver的实现，如下：
@Override
public void onNext(T t) {
// 上一级的模式如果不是异步的，加入队列
if (sourceMode != QueueDisposable.ASYNC) {
queue.offer(t);
}
//进行线程调度
schedule();
}
void schedule() {
// 判断当前正在执行的任务数目
if (getAndIncrement() == 0) {
worker.schedule(this);
}
}
ObserveOnObserver本身继承了Runnable接口，run方法实现如下：
@Override
public void run() {
//输出结果是否融合
if (outputFused) {
drainFused();
} else {
drainNormal();
}
}
我们先进入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;
}
}
}
关于scheduler的具体实现，在后续的线程原理进行分析。这里我们只需要知道下游的观察者的onNext在指定的scheduler线程策略中执行就可以了。
subscribe
@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;
}
}
最后的订阅方法在做了非空检查后，会调用subscribeActual方法，开始往上游逐层执行订阅。
被观察者Observable是如何发送数据？
通过上面的流程分析，我们可以知道。如果使用create创建了Observable，在ObservableOnSubscribe的subscribe方法中会通过ObservableEmitter的onNext去发送数据，onNext会触发开始往下游观察者传递数据。当然rxjava的创建型操作符还有很多，如just、from等，本质最后都是触发下游观察者的onNext进行数据的发送。
观察者Observer是如何接收到数据的？
通过源码分析，每一个链层的Observer都会持有相邻下游的Observer对象，当开始发送数据时，会依次链式执行Observer的onNext方法，最后执行到subscribe方法中创建的Observer对象。
被观察者和观察者之间是如何实现订阅？
每一个链层的Observable 都会持有相邻上游的Observable对象，在subscribe方法开始调用后，最后会执行到subscribeActual方法，在subscribeActual方法中会将观察者与上游的被观察执行订阅。
rxjava是如何进行线程的调度？
rxjava的Scheduler有很多种实现，下面我们介绍Scheduler的相关说明，然后通过最常用的.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())来分析具体的线程调度流程。
Scheduler
我们在调用subscribeOn与observeOn时，都会传入Scheduler对象，首先我们先看一下Scheduler的种类及其功能
| Scheduler种类 | 说明 |
|---|---|
| Schedulers.io( ) | 用于IO密集型的操作，例如读写SD卡文件，查询数据库，访问网络等，具有线程缓存机制，在此调度器接收到任务后，先检查线程缓存池中，是否有空闲的线程，如果有，则复用，如果没有则创建新的线程，并加入到线程池中，如果每次都没有空闲线程使用，可以无上限的创建新线程 |
| Schedulers.newThread( ) | 在每执行一个任务时创建一个新的线程，不具有线程缓存机制，因为创建一个新的线程比复用一个线程更耗时耗力，虽然使用Schedulers.io( )的地方，都可以使用Schedulers.newThread( )，但是，Schedulers.newThread( )的效率没有Schedulers.io( )高 |
| Schedulers.computation() | 用于CPU 密集型计算任务，即不会被 I/O 等操作限制性能的耗时操作，例如xml,json文件的解析，Bitmap图片的压缩取样等，具有固定的线程池，大小为CPU的核数。不可以用于I/O操作，因为I/O操作的等待时间会浪费CPU |
| Schedulers.trampoline() | 在当前线程立即执行任务，如果当前线程有任务在执行，则会将其暂停，等插入进来的任务执行完之后，再将未完成的任务接着执行 |
| Schedulers.single() | 拥有一个线程单例，所有的任务都在这一个线程中执行，当此线程中有任务执行时，其他任务将会按照先进先出的顺序依次执行 |
| Scheduler.from(Executor executor) | 指定一个线程调度器，由此调度器来控制任务的执行策略 |
| AndroidSchedulers.mainThread() | 在Android UI线程中执行任务，为Android开发定制 |
subscribeOn(Schedulers.io())
根据上面的分析，subscribeOn()方法最后会执行到subscribeActual方法，SubscribeTask上面分析了，继承了Runnable接口， run方法最后会执行source.subscribe(parent)方法。
@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)));
}
这里我们主要要分析scheduler.scheduleDirect()方法。
@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;
}
Worker的schedule是一个抽象的方法，Schedulers.io()对应的Worker实现为EventLoopWorker。我们看看EventLoopWorker的schedule实现如下：
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);
}
}
这里会执行到 threadWorker的scheduleActual方法，继续往下看
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;
}
在这里会使用executor最终去执行run方法。当然看到这里有一个疑问IoScheduler在这里是怎么实现线程的复用呢？我们看看threadWorker在IoScheduler中的线程的创建，如下：
EventLoopWorker(CachedWorkerPool pool) {
this.pool = pool;
this.tasks = new CompositeDisposable();
this.threadWorker = pool.get();
}
这里会通过维护一个Worker的线程池来达到线程复用的效果，具体我们看看CachedWorkerPool的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;
}
observeOn(AndroidSchedulers.mainThread())
@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));
}
}
这里我们主要需要分析ObserveOnObserver对象，onNext实现如下：
@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);
}
}
关键还是执行了worker的schedule，AndroidSchedulers的实现主要为HandlerScheduler，HandlerScheduler中关于Worker的实现为HandlerWorker，我们看下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 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;
}
关于handler的实例，我们看AndroidSchedulers中的创建如下：
private static final class MainHolder {
static final Scheduler DEFAULT
= new HandlerScheduler(new Handler(Looper.getMainLooper()), false);
}
综上可知AndroidSchedulers.mainThread()是通过消息将run方法的实现交由主线程Looper进行处理，达到将观察者的数据处理在主线程中执行的效果
rxjava背压策略实现原理是怎样的？
背压（backpressure）
当上下游在不同的线程中，通过Observable发射，处理，响应数据流时，如果上游发射数据的速度快于下游接收处理数据的速度，这样对于那些没来及处理的数据就会造成积压，这些数据既不会丢失，也不会被垃圾回收机制回收，而是存放在一个异步缓存池中，如果缓存池中的数据一直得不到处理，越积越多，最后就会造成内存溢出，这便是响应式编程中的背压（backpressure）问题。
背压处理机制
rxjava2.x使用Flowable来支持背压的机制，调用create方法时需要传入BackpressureStrategy策略。
| Strategy | 作用 |
|---|---|
| MISSING | 此策略表示，通过Create方法创建的Flowable没有指定背压策略，不会对通过OnNext发射的数据做缓存或丢弃处理，需要下游通过背压操作符（onBackpressureBuffer()/onBackpressureDrop()/onBackpressureLatest()）指定背压策略 |
| ERROR | 在此策略下，如果放入Flowable的异步缓存池中的数据超限了，则会抛出MissingBackpressureException异常 |
| BUFFER | 此策略下，Flowable的异步缓存池同Observable的一样，没有固定大小，可以无限制向里添加数据，不会抛出MissingBackpressureException异常，但会导致OOM |
| DROP | 在此策略下，如果Flowable的异步缓存池满了，会丢掉上游发送的数据 |
| LATEST | 与Drop策略一样，如果缓存池满了，会丢掉将要放入缓存池中的数据，不同的是，不管缓存池的状态如何，LATEST都会将最后一条数据强行放入缓存池中 |
实现原理
首先看看Flowable的create实现
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));
}
这里会创建一个FlowableCreate对象，并传入指定的BackpressureStrategy策略。接着看看FlowableCreate的订阅方法
@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);
}
}
BaseEmitter
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();
}
}
//省略其他若干方法
通过上面的结束我们知道Flowable有一个缓冲池，那个这个大小是多少，在哪里进行复制给发射器呢？
//长度是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());
}
MissingEmitter
不会对通过OnNext发射的数据做缓存或丢弃处理
@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;
}
}
}
NoOverflowBaseAsyncEmitter
DropAsyncEmitter和ErrorAsyncEmitter继承了NoOverflowBaseAsyncEmitter
@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();
}
}
DropAsyncEmitter
如果Flowable的异步缓存池满了，会丢掉上游发送的数据
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
}
}
ErrorAsyncEmitter
如果Flowable的异步缓存池满了，会抛出异常
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"));
}
}
BufferAsyncEmitter
Flowable的异步缓存池同Observable的一样，没有固定大小，可以无限制向里添加数据
@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();
}
LatestAsyncEmitter
Flowable的异步缓存池同Observable的一样，没有固定大小，可以无限制向里添加数据
@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();
}
正文到此结束
- 本文标签： cache 本质 tab 线程 微软 CTO 组织 build message plugin 模型 源码 数据库 IDE CEO value consumer 线程池 mapper id tag 定制 数据 设计模式 XML 开源 调度器 ORM 服务端 stream java js 调试 响应式 json final IOS Action 垃圾回收 缓存 rmi https 开发者 时间 实例 开发 http IO 构造方法 git UI 代码 解析 一对多 ask JavaScript 图片 架构师 ip map GitHub Android 参数 src Atom executor token mina cat 领导 App REST client queue
- 版权声明： 本文为互联网转载文章，出处已在文章中说明(部分除外)。如果侵权，请联系本站长删除，谢谢。
- 本文海报： 生成海报一 生成海报二
热门推荐
相关文章
近期评论
-
文章和留言都翻到11页了 没有OOM
-
-
朋友，翻页到11页，及以后，会出现OOM，无法访问
-
-
搞个gitee的项目
-
-
版本号是多少，你可以下载哪个代码仓库，jdk选1.8 直接跑就行
-
-
-
Loading...
![[HBLOG]公众号](https://www.liuhaihua.cn/img/qrcode_gzh.jpg)
