Rxjava2~defer~create~just~from~学

<pre>
static Observable<String> sampleObservable() {
return Observable.defer(new Callable<ObservableSource<? extends String>>() {
@Override
public ObservableSource<? extends String> call() throws Exception {
// Do some long running operation
SystemClock.sleep(2000);
return Observable.just("one", "two", "three", "four", "five");
}
});
}
</pre>
一上来我们就举个例子可以看出来两个操作符。 有了前面的经验我们来简单的分析一下。这几个创建操作符的不同

<pre>
public static <T> Observable<T> defer(Callable<? extends ObservableSource<? extends T>> supplier) {
ObjectHelper.requireNonNull(supplier, "supplier is null");
return RxJavaPlugins.onAssembly(new ObservableDefer<T>(supplier));
}
</pre>

和以前一样Observable的静态方法。
<pre>
public final class ObservableDefer<T> extends Observable<T> {
final Callable<? extends ObservableSource<? extends T>> supplier;
public ObservableDefer(Callable<? extends ObservableSource<? extends T>> supplier) {
this.supplier = supplier;
}
@Override
public void subscribeActual(Observer<? super T> s) {
ObservableSource<? extends T> pub;
try {
pub = ObjectHelper.requireNonNull(supplier.call(), "null publisher supplied");
} catch (Throwable t) {
Exceptions.throwIfFatal(t);
EmptyDisposable.error(t, s);
return;
}

    pub.subscribe(s);
}

}

</pre>

这是完整ObservableDefer代码。可以看到subscribeActual这个方法不？看到这个我想大家一定马上就想到另一个方法Observable的静态方法subscribe,中最重要的部分 subscribeActual(observer);。吼吼 看到了吗。 只有当你调用的时候（订阅）才会整理数据。
<pre>
a = 12;
Observable<String> o2 =
Observable.defer(new Func0<Observable<String>>() {

@Override
public Observable<String> call() {
    return Observable.just("defer result: " + a);
}

});
a = 20;

o2.subscribe(new Action1<String>() {

@Override
public void call(String t) {
    System.out.println(t);
}

});

</pre>
这是一个网上例子。根据咱们的分析 。 但你订阅了subscribe。它才开始去找寻数据a=20;是不是很简单？ 好了 往下进行

<pre>
public void subscribeActual(Observer<? super T> s) {
ObservableSource<? extends T> pub;
try {
pub = ObjectHelper.requireNonNull(supplier.call(), "null publisher supplied");
} catch (Throwable t) {
Exceptions.throwIfFatal(t);
EmptyDisposable.error(t, s);
return;
}

    pub.subscribe(s);
}

</pre>

最重要的相信大家直接看到出来了。
pub = ObjectHelper.requireNonNull(supplier.call(), "null publisher supplied");
supplier.call()这个就是调用你的call方法.下面的方法。大家试着分析一下？
pub.subscribe(s);
又回到了
Observable的静态方法subscribe。那么他会调用谁的？subscribeActual，大家肯定猜到了。
ObservableFromArray的subscribeActual方法~ 为什么呢？ 看这里看这里
<pre>
public static <T> Observable<T> fromArray(T... items) {
ObjectHelper.requireNonNull(items, "items is null");
if (items.length == 0) {
return empty();
} else
if (items.length == 1) {
return just(items[0]);
}
return RxJavaPlugins.onAssembly(new ObservableFromArray<T>(items));
}
</pre>
这是just的方法哦，让我们继续
<pre>
public void subscribeActual(Observer<? super T> s) {
FromArrayDisposable<T> d = new FromArrayDisposable<T>(s, array);

    s.onSubscribe(d);

    if (d.fusionMode) {
        return;
    }

    d.run();
}

void run() {
T[] a = array;
int n = a.length;

        for (int i = 0; i < n && !isDisposed(); i++) {
            T value = a[i];
            if (value == null) {
                actual.onError(new NullPointerException("The " + i + "th element is null"));
                return;
            }
            actual.onNext(value);
        }
        if (!isDisposed()) {
            actual.onComplete();
        }
    }
}

</pre>
看到这里明白了吧！大家思考一下。这两个以后什么不同点？
3
2
1
defer 中文的意思是推迟。 所以他们的区别就是 在创建的时候 defer的数据都是最新的，因为他在订阅的时候才回去执行，其他的just 就不会。
让我们进行下一个
creat 这个原理其实差不多。 我们来简单的走一下流程

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

</pre>

ObservableCreate的subscribeActual方法

<pre>
protected void subscribeActual(Observer<? super T> observer) {
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
observer.onSubscribe(parent);

    try {
        source.subscribe(parent);
    } catch (Throwable ex) {
        Exceptions.throwIfFatal(ex);
        parent.onError(ex);
    }
}

</pre>

这里要稍微留意一下
source.subscribe(parent);
这里的方法调用的是
<pre>
Observable.create(new ObservableOnSubscribe<List<User>>() {
@Override
public void subscribe(ObservableEmitter<List<User>> e) throws Exception {
if (!e.isDisposed()) {
e.onNext(Utils.getUserListWhoLovesCricket());
e.onComplete();
}
}
</pre>
的subscribe方法，当调用onNext 或者onComplete的时候。他调用的就是ObservableCreate下的
onNext等方法
<pre>
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;
}
if (!isDisposed()) {
observer.onNext(t);
}
}

    @Override
    public void onError(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();
            }
        } else {
            RxJavaPlugins.onError(t);
        }
    }

    @Override
    public void onComplete() {
        if (!isDisposed()) {
            try {
                observer.onComplete();
            } finally {
                dispose();
            }
        }
    }

<
</pre>
看看是不是特别简单
最后一个喽
from
。。。。。这个貌似很长。我要做一个专门的专题。