一、RxJava 简介
RxJava是一种基于观察者模式的响应式编程框架，其定义的角色有Observable事件流、ObservableOnSubscribe事件源头、Observer是事件订阅者。
二、基本使用
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("Android");
emitter.onNext("Kotlin");
emitter.onNext("Java");
emitter.onComplete();
}
}).map(new Function<String, String>() {
@Override
public String apply(String s) throws Exception {
return "I Love "+ s;
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String s) {
Log.d("test", "onNext: "+s);
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
Log.d("test", "[onComplete]");
}
});
运行结果：
test: onNext: I Love Android
test: onNext: I Love Kotlin
test: onNext: I Love Java
test: [onComplete]
三、原理剖析
3.1 Observable、Observer订阅原理
Observable.java
public abstract class Observable<T> implements ObservableSource<T> {
// 创建事件源
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
ObjectHelper.requireNonNull(source, "source is null");
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
// 订阅相关
public final void subscribe(Observer<? super T> observer) {
// ...
subscribeActual(observer);
}
protected abstract void subscribeActual(Observer<? super T> observer);
}
create方法创建事件源,其参数ObservableOnSubscribe 接口定义如下：
ObservableOnSubscribe.java
public interface ObservableOnSubscribe<T> {
void subscribe(@NonNull ObservableEmitter<T> emitter) throws Exception;
}
ObservableEmitter可以理解为事件发射器，继承Emitter里卖弄的基本发射方法，同时关连Disposable
Emitter.java
public interface Emitter<T> {
void onNext(@NonNull T value);
void onError(@NonNull Throwable error);
void onComplete();
}
ObservableEmitter.java
public interface ObservableEmitter<T> extends Emitter<T> {
void setDisposable(@Nullable Disposable d);
void setCancellable(@Nullable Cancellable c);
boolean isDisposed();
ObservableEmitter<T> serialize();
boolean tryOnError(@NonNull Throwable t);
}
Observable.create方法最终会返回ObservableCreate<T>对象，ObservableCreate类包含了ObservableOnSubscribe对象引用，且实现了subscribeActual订阅方法
ObservableCreate.java
public final class ObservableCreate<T> extends Observable<T> {
final ObservableOnSubscribe<T> source;
public ObservableCreate(ObservableOnSubscribe<T> source) {
this.source = source;
}
// 订阅的时候会执行该方法
@Override
protected void subscribeActual(Observer<? super T> observer) {
// 1 创建 CreateEmitter，也是一个适配器，可以将 Observer -> Disposable，CreateEmitter 中主要持有 observer 对象的引用，并且维护了 dispose 变量。
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
//2 onSubscribe（）参数是 Disposable。还有一点要注意的是 onSubscribe() 是在我们执行 subscribe() 这句代码的那个线程回调的，并不受线程调度影响。
// 给 observer 的一个回调，告诉它是否 dispose
observer.onSubscribe(parent);
try {
//3 将 ObservableOnSubscribe（源头）与 CreateEmitter（Observer，终点）联系起来，即完成订阅，此时 ObservableOnSubscribe 会向 observer 传送事件
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
static final class CreateEmitter<T> extends AtomicReference<Disposable> implements ObservableEmitter<T>, Disposable {
private static final long serialVersionUID = -3434801548987643227L;
final Observer<? super T> observer;
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;
}
// 如果没有被 dispose，会调用 Observer 的 onNext()方法
if (!isDisposed()) {
observer.onNext(t);
}
}
@Override
public void onError(Throwable t) {
if (!tryOnError(t)) {
RxJavaPlugins.onError(t);
}
}
@Override
public boolean tryOnError(Throwable t) {
if (t == null) {
t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
}
if (!isDisposed()) {
try {
observer.onError(t);
} finally {
dispose();
}
return true;
}
return false;
}
@Override
public void onComplete() {
// onComplete() 和 onError() 互斥只能执行一次
if (!isDisposed()) {
try {
observer.onComplete();
} finally {
dispose();
}
}
}
@Override
public void setDisposable(Disposable d) {
DisposableHelper.set(this, d);
}
@Override
public void setCancellable(Cancellable c) {
setDisposable(new CancellableDisposable(c));
}
@Override
public ObservableEmitter<T> serialize() {
return new SerializedEmitter<T>(this);
}
@Override
public void dispose() {
DisposableHelper.dispose(this);
}
@Override
public boolean isDisposed() {
return DisposableHelper.isDisposed(get());
}
@Override
public String toString() {
return String.format("%s{%s}", getClass().getSimpleName(), super.toString());
}
}
}
其中Observer订阅者定义如下：
public interface Observer<T> {
void onSubscribe(@NonNull Disposable d);
void onNext(@NonNull T t);
void onError(@NonNull Throwable e);
void onComplete();
}
3.2 map操作符原理
map函数的源码
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.java
public final class ObservableMap<T, U> extends AbstractObservableWithUpstream<T, U> {
final Function<? super T, ? extends U> function;
public ObservableMap(ObservableSource<T> source, Function<? super T, ? extends U> function) {
// super()将上游的Observable保存起来 ，用于subscribeActual()中用。
super(source);
// 将function变换函数类保存起来
this.function = function;
}
@Override
public void subscribeActual(Observer<? super U> t) {
// 用 MapObserver 订阅上游 Observable。
source.subscribe(new MapObserver<T, U>(t, function));
}
static final class MapObserver<T, U> extends BasicFuseableObserver<T, U> {
final Function<? super T, ? extends U> mapper;
MapObserver(Observer<? super U> actual, Function<? super T, ? extends U> mapper) {
super(actual);
this.mapper = mapper;
}
@Override
public void onNext(T t) {
// done在onError 和 onComplete以后才会是true，默认这里是false，所以跳过
if (done) {
return;
}
// 默认sourceMode是0，所以跳过
if (sourceMode != NONE) {
downstream.onNext(null);
return;
}
U v;
try {
// 这一步执行变换,将上游传过来的 T，利用 Function 转换成下游需要的 V
v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper function returned a null value.");
} catch (Throwable ex) {
fail(ex);
return;
}
// 变换后传递给下游Observer
downstream.onNext(v);
}
@Override
public int requestFusion(int mode) {
return transitiveBoundaryFusion(mode);
}
@Nullable
@Override
public U poll() throws Exception {
T t = qd.poll();
return t != null ? ObjectHelper.<U>requireNonNull(mapper.apply(t), "The mapper function returned a null value.") : null;
}
}
}
3.3 subscribeOn 线程调度
subscribeOn 函数源码：
public final Observable<T> subscribeOn(Scheduler scheduler) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
return RxJavaPlugins.onAssembly(
// 返回Observable，传入上游this和调度器scheduler
new ObservableSubscribeOn<T>(this, scheduler));
}
public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {
final Scheduler scheduler;
public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
super(source);
this.scheduler = scheduler;
}
@Override
public void subscribeActual(final Observer<? super T> downStream) {
final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(downStream);
// SubscribeOnObserver实现了Disposable，将其传递给下游的onSubscribe
downStream.onSubscribe(parent);
// 把source的订阅放在Runnable中，由scheduler调度
parent.setDisposable(scheduler.scheduleDirect(new Runnable() {
@Override
public void run() {
source.subscribe(parent);
}
}));
}
}
observeOn函数源码:
public final Observable<T> observeOn(Scheduler scheduler) {
return observeOn(scheduler, false, bufferSize());
}
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));
}
// 以上代码都很熟悉了，暂不赘述
public final class ObservableObserveOn<T> extends AbstractObservableWithUpstream<T, T> {
final Scheduler scheduler;
final boolean delayError;
final int bufferSize;
public ObservableObserveOn(ObservableSource<T> source, Scheduler scheduler, boolean delayError, int bufferSize) {
super(source);
this.scheduler = scheduler; // 线程调度器
this.delayError = delayError; // 出现错误是否立刻中断线程
this.bufferSize = bufferSize; //缓冲区大小，默认128
}
@Override
protected void subscribeActual(Observer<? super T> observer) {
if (scheduler instanceof TrampolineScheduler) {
// 默认线程则不做线程调度，直接在当前线程中调用
source.subscribe(observer);
} else {
Scheduler.Worker w = scheduler.createWorker();
source.subscribe(
// 用Worker、和相关参数装饰observer，得到新的Observer注入上游
new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
}
}
static final class ObserveOnObserver<T> extends BasicIntQueueDisposable<T> implements Observer<T>, Runnable {
//省略部分代码
@Override
public void onSubscribe(Disposable s) {
if (DisposableHelper.validate(this.s, s)) {
this.s = s;
//省略部分代码，创建缓冲队列
queue = new SpscLinkedArrayQueue<T>(bufferSize);
actual.onSubscribe(this);
}
}
@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);
}
}
@Override
public void run() {
// Fused 熔断机制，默认false
if (outputFused) {
drainFused();
} else {
drainNormal();
}
}
//该函数在Runnable所在的线程执行，从缓冲队列里拿出事件，向下游发射
void drainNormal() {
int missed = 1;
final SimpleQueue<T> q = queue;
final Observer<? super T> a = actual;
for (;;) {
// 如果设置了errorDelay，则不管队列是否为空，发生了错误都会中断发射，并调用observer的onError
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);
worker.dispose();
return;
}
boolean empty = v == null;
if (checkTerminated(d, empty, a)) {
return;
}
if (empty) {
break;
}
// 向下游发射数据
a.onNext(v);
}
missed = addAndGet(-missed);
if (missed == 0) {
break;
}
}
}
}
}
