开发笔记:浅谈RxJava源码解析(观察者),创建(createfromjust)，变换(MapflatMap)线程调度

Observable.create(new Observable.OnSubscribe() {

@Override
public void call(Subscriber subscriber) {
subscriber.onNext("Hello");
subscriber.onNext("RxJava");
subscriber.onCompleted();
}
}).subscribe(new Observer() {
@Override
public void onCompleted() {
Log.i(TAG, "onCompleted");
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
Log.i(TAG, "OnNext=" + s);
}
});
Subscriber 对 Observer 接口进行了一些扩展，但他们的基本使用方式是完全一样的，只是多了onStart()方法，作为异步调用之前的操作：

.subscribe(new Subscriber() {
@Override
public void onStart() {
Log.i(TAG, "onStart");
}
@Override
public void onCompleted() {
Log.i(TAG, "onCompleted");
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Object o) {
Log.i(TAG, "OnNext=" + o);
}
});
*****************************************************源码解析订阅**************************************

public final Subscription subscribe(Subscriber subscriber) {
return Observable.subscribe(subscriber, this);
}
private static Subscription subscribe(Subscriber subscriber, Observable observable) {
// validate and proceed
if (subscriber == null) {
throw new IllegalArgumentException("observer can not be null");
}
if (observable.OnSubscribe== null) {
throw new IllegalStateException("onSubscribe function can not be null.");
/*
* the subscribe function can also be overridden but generally that‘s not the appropriate approach
* so I won‘t mention that in the exception
*/
}

// new Subscriber so onStart it
subscriber.onStart();

/*
* See https://github.com/ReactiveX/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls
* to user code from within an Observer"
*/
// if not already wrapped
if (!(subscriber instanceof SafeSubscriber)) {
// assign to `observer` so we return the protected version
subscriber = new SafeSubscriber(subscriber);
}
// The code below is exactly the same an unsafeSubscribe but not used because it would
// add a significant depth to already huge call stacks.
try {
// allow the hook to intercept and/or decorate
hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);
return hook.onSubscribeReturn(subscriber);
} catch (Throwable e) {
// special handling for certain Throwable/Error/Exception types
Exceptions.throwIfFatal(e);
// if an unhandled error occurs executing the onSubscribe we will propagate it
try {
subscriber.onError(hook.onSubscribeError(e));//捕获异常并回调onError()
} catch (Throwable e2) {
Exceptions.throwIfFatal(e2);
// if this happens it means the onError itself failed (perhaps an invalid function implementation)
// so we are unable to propagate the error correctly and will just throw
RuntimeException r = new RuntimeException("Error occurred attempting to subscribe [" + e.getMessage() + "] and then again while trying to pass to onError.", e2);
// TODO could the hook be the cause of the error in the on error handling.
hook.onSubscribeError(r);
// TODO why aren‘t we throwing the hook‘s return value.
throw r;
}
return Subscriptions.unsubscribed();
}
}

*****************************************************源码解析订阅**************************************

5、from()：

String[] arrays={"Hello","RxJava"};
Observable.from(arrays)
.subscribe(new Subscriber() {
@Override
public void onStart() {
Log.i(TAG, "onStart");
}
@Override
public void onCompleted() {
Log.i(TAG, "onCompleted");
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Object o) {
Log.i(TAG, "OnNext=" + o);
}
});

*****************************************************源码解析订阅**************************************

public final class{
　　...
　　//此处create()就不多做解释了，...call()...
　　public final static Observable from(Iterable iterable) {        
　　　　return create(new OnSubscribeFromIterable(iterable));    
　　}
　　...
}

public final class OnSubscribeFromIterable implements OnSubscribe {
final Iterable is;
public OnSubscribeFromIterable(Iterable iterable) {
if (iterable == null) {
throw new NullPointerException("iterable must not be null");
}
this.is = iterable;
}
@Override
public void call(final Subscriber o) {
final Iterator it = is.iterator();
if (!it.hasNext() && !o.isUnsubscribed())
o.onCompleted();
else
o.setProducer(new IterableProducer(o, it));//未执行完继续迭代
}
private static final class IterableProducer extends AtomicLong implements Producer {
/** */
private static final long serialVersiOnUID= -8730475647105475802L;
private final Subscriber o;
private final Iterator it;
private IterableProducer(Subscriber o, Iterator it) {
this.o = o;
this.it = it;
}
@Override
public void request(long n) {
if (get() == Long.MAX_VALUE) {
// already started with fast-path
return;
}
if (n == Long.MAX_VALUE && compareAndSet(0, Long.MAX_VALUE)) {
fastpath();
} else
if (n > 0 && BackpressureUtils.getAndAddRequest(this, n) == 0L) {
slowpath(n);
}
}
void slowpath(long n) {
// backpressure is requested
final Subscriber o = this.o;
final Iterator it = this.it;
long r = n;
while (true) {
/*
* This complicated logic is done to avoid touching the
* volatile `requested` value during the loop itself. If
* it is touched during the loop the performance is
* impacted significantly.
*/
long numToEmit = r;
while (true) {if (o.isUnsubscribed()) { return;} else if (it.hasNext()) { if (--numToEmit >= 0) { o.onNext(it.next()); } else break;} else if (!o.isUnsubscribed()) { o.onCompleted(); return;} else { // is unsubscribed return;}
}
r = addAndGet(-r);
if (r == 0L) {// we‘re done emitting the number requested so// returnreturn;
}
}
}
void fastpath() {
// fast-path without backpressure
final Subscriber o = this.o;
final Iterator it = this.it;
while (true) {
if (o.isUnsubscribed()) {return;
} else if (it.hasNext()) {o.onNext(it.next());
} else if (!o.isUnsubscribed()) {o.onCompleted();return;
} else {// is unsubscribedreturn;
}
}
}
}
}

*****************************************************源码解析订阅**************************************

6、just():

*****************************************************源码解析订阅**************************************

public final static Observable just(T t1, T t2) { //然而源码解析完之后你会觉得 同上from()
return from(Arrays.asList(t1, t2));
}

*****************************************************源码解析订阅**************************************

二、变换操作：(重点：当然还是要先创建被观察者)
1、Map(): (多输入，单输出的概念，用代理模式去理解map()方法执行过程，简单说就是Observable和OnSubscribe被新的取代了)

*****************************************************源码解析订阅**************************************

public class Observable {
...
　　//订阅，跟上面一样
private static Subscription subscribe(Subscriber subscriber, Observable observable) {
// validate and proceed
if (subscriber == null) {
throw new IllegalArgumentException("observer can not be null");
}
if (observable.OnSubscribe== null) {
throw new IllegalStateException("onSubscribe function can not be null.");
/*
* the subscribe function can also be overridden but generally that‘s not the appropriate approach
* so I won‘t mention that in the exception
*/
}
// new Subscriber so onStart it
subscriber.onStart();

/*
* See https://github.com/ReactiveX/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls
* to user code from within an Observer"
*/
// if not already wrapped
if (!(subscriber instanceof SafeSubscriber)) {
// assign to `observer` so we return the protected version
subscriber = new SafeSubscriber(subscriber);
}
// The code below is exactly the same an unsafeSubscribe but not used because it would
// add a significant depth to already huge call stacks.
try {
// allow the hook to intercept and/or decorate
hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);//发射OnSubscrible中的call()；注意，此时已替换了，用代理思维去理解。
return hook.onSubscribeReturn(subscriber);
} catch (Throwable e) {
// special handling for certain Throwable/Error/Exception types
Exceptions.throwIfFatal(e);
// if an unhandled error occurs executing the onSubscribe we will propagate it
try {
subscriber.onError(hook.onSubscribeError(e));
} catch (Throwable e2) {
Exceptions.throwIfFatal(e2);
// if this happens it means the onError itself failed (perhaps an invalid function implementation)
// so we are unable to propagate the error correctly and will just throw
RuntimeException r = new RuntimeException("Error occurred attempting to subscribe [" + e.getMessage() + "] and then again while trying to pass to onError.", e2);
// TODO could the hook be the cause of the error in the on error handling.
hook.onSubscribeError(r);
// TODO why aren‘t we throwing the hook‘s return value.
throw r;
}
return Subscriptions.unsubscribed();
}
}
public final Observable lift(final Operator operator) {
return new Observable(new OnSubscribe() { //重新创建了新的Observable和OnSubscribe
@Override
public void call(Subscriber o) {
try {Subscriber st = hook.onLift(operator).call(o);//回调替换的部分逻辑try { // new Subscriber created and being subscribed with so ‘onStart‘ it st.onStart(); onSubscribe.call(st);//发射新的} catch (Throwable e) { // localized capture of errors rather than it skipping all operators // and ending up in the try/catch of the subscribe method which then // prevents onErrorResumeNext and other similar approaches to error handling Exceptions.throwIfFatal(e); st.onError(e);}
} catch (Throwable e) {Exceptions.throwIfFatal(e);// if the lift function failed all we can do is pass the error to the final Subscriber// as we don‘t have the operator available to uso.onError(e);
}
}
});
}
...
}

*****************************************************源码解析订阅**************************************

2、flatMap():

 比较flatMap与map：

假设：如果要打印出每个学生所需要修的所有课程的名称呢？（需求的区别在于，每个学生只有一个名字，但却有多个课程。）首先可以这样实现：
Student[] students = ...;
Subscriber subscriber = new Subscriber() {
　　@Override
　　public void onNext(Student student) {
　　　　List courses = student.getCourses();
　　　　for (int i = 0; i 　　　　　　Course course = courses.get(i);
　　 Log.d(tag, course.getName());
　　 }
　　}
...
};
Observable.from(students)
　　.subscribe(subscriber);
而flatMap:
Student[] students = ...;
Subscriber subscriber = new Subscriber() {
　　@Override
　　public void onNext(Course course) {
Log.d(tag, course.getName());
}
...
};
Observable.from(students)
　　.flatMap(new Func1>() {
　　　　@Override
　　　　public Observable call(Student student) {
　　return Observable.from(student.getCourses());
}
})
.subscribe(subscriber);
3、Filter():

Observable.just(4, 2, 1, 7, 5)
.filter(new Func1() {
@Override
public Boolean call(Integer integer) {
return integer > 3;
}
})
.subscribe(new Observer() {
@Override
public void onCompleted() {
Log.i("wxl", "onCompleted");
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Integer integer) {
Log.i("wxl", "OnNext=" + integer);
}
});

***************************************************源码解析****************************************************************

public final Observable filter(Func1 predicate) {
return lift(new OperatorFilter(predicate));
}

public final class OperatorFilter implements Operator {
private final Func1 predicate;
public OperatorFilter(Func1 predicate) {
this.predicate = predicate;
}
@Override
public Subscriber call(final Subscriber child) {
return new Subscriber(child) {
@Override
public void onCompleted() {
child.onCompleted();
}
@Override
public void onError(Throwable e) {
child.onError(e);
}
@Override
public void onNext(T t) {
try {if (predicate.call(t)) {//判断 child.onNext(t);} else { // TODO consider a more complicated version that batches these request(1);}
} catch (Throwable e) {Exceptions.throwOrReport(e, child, t);
}
}
};
}
}

***************************************************源码解析****************************************************************

4、线程调度：

.subscribeOn(Schedulers.io()):

***************************************************源码解析****************************************************************

public final Observable subscribeOn(Scheduler scheduler) {
if (this instanceof ScalarSynchronousObservable) {
return ((ScalarSynchronousObservable)this).scalarScheduleOn(scheduler);
}
return nest().lift(new OperatorSubscribeOn(scheduler));
}

public final class Schedulers {
　　//
private final Scheduler computationScheduler;
private final Scheduler ioScheduler;
private final Scheduler newThreadScheduler;
private static final Schedulers INSTANCE = new Schedulers();
　　　....
}

public class OperatorSubscribeOn implements Operator> {
private final Scheduler scheduler;
public OperatorSubscribeOn(Scheduler scheduler) {
this.scheduler = scheduler;
}
@Override
public Subscriber> call(final Subscriber subscriber) {
final Worker inner = scheduler.createWorker();
subscriber.add(inner);
return new Subscriber>(subscriber) {
@Override
public void onCompleted() {
// ignore because this is a nested Observable and we expect only 1 Observable emitted to onNext
}
@Override
public void onError(Throwable e) {
subscriber.onError(e);
}
@Override
public void onNext(final Observable o) {
inner.schedule(new Action0() {//在选择的线程里做处理
@Overridepublic void call() { final Thread t = Thread.currentThread(); o.unsafeSubscribe(new Subscriber(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); } }); } }
}); }
});}
});
}
};
}
}

***************************************************源码解析****************************************************************

.observeOn(AndroidSchedulers.mainThread())：
参考OperatorObserveOn中源码；
基本思路：最好用代理思维去理解

(被观察者)Observable：(订阅者)OnSubscribe extends Action1 extends Action extends Function

——> 订阅 subscribe(Observer) 回调处理

(观察者)Observer：Subscriber implements Observer

 应用场景还没有，希望多交流多指正！