take()
<pre>
Observable.just(1, 2, 3, 4, 5)
.subscribeOn(Schedulers.io())
// Be notified on the main thread
.observeOn(AndroidSchedulers.mainThread())
.take(3)
.subscribe(getObserver())
</pre>
輸出沒(méi)錯(cuò)是123
我們面來(lái)看看源碼
直接來(lái)看ObservableTake的subscribeActual钙勃,[不懂的同學(xué)請(qǐng)看我前面的學(xué)渣系列]
<pre>
protected void subscribeActual(Observer<? super T> observer) {
source.subscribe(new TakeObserver<T>(observer, limit));
}
</pre>
這個(gè)source是ObservableSource的對(duì)象。 那么我們?nèi)フ覍?shí)現(xiàn)他的Observable
好吧 又回到了碉就。
<pre>
public final void subscribe(Observer<? super T> observer)
subscribeActual(observer);
其他的省略了
</pre>
關(guān)鍵點(diǎn)一步底挫,這回調(diào)用了誰(shuí)的方法呢血久? 下面來(lái)揭曉
是ObservableObserveOn的subscribeActual
<pre>
@Override
protected void subscribeActual(Observer<? super T> observer) {
if (scheduler instanceof TrampolineScheduler) {
source.subscribe(observer);
} else {
Scheduler.Worker w = scheduler.createWorker();
source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
}
}
</pre>
看到了嗎 又會(huì)調(diào)用
source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
然后 又要調(diào)用的是ObservableSubscribeOn的subscribeActual
<pre>
@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 Runnable() {
@Override
public void run() {
source.subscribe(parent);
}
}));
}
</pre>
大家會(huì)好奇這兩個(gè)地方為什么會(huì)被調(diào)用呢?
下面我給大家看一個(gè)地方
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
O(∩_∩)O
你沒(méi)有看錯(cuò)
<pre>
public final Observable<T> observeOn(Scheduler scheduler) {
return observeOn(scheduler, false, bufferSize());
}
public final Observable<T> subscribeOn(Scheduler scheduler) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));
}
</pre>
大家可以看到不。 這兩個(gè)方法返回的也是Observable對(duì)象挖炬。 所以 他們會(huì)分別調(diào)用這兩個(gè)對(duì)象subscribeActual方法讯壶。好吧料仗,讓我們來(lái)像下進(jìn)行。
【下面是一個(gè)小擴(kuò)展 給大家一個(gè)小小的感覺(jué)】
<pre>
Observable.just(1, 2, 3, 4, 5)
.observeOn(AndroidSchedulers.mainThread())
.take(3)
.subscribe(getObserver())
看到有什么不同了嗎伏蚊? 我注釋掉了一個(gè)方法立轧。我為什么要這么干?我注視掉了那么
source.subscribe 會(huì)調(diào)用誰(shuí)呢躏吊? 我直接給出來(lái)答案氛改。大家可以思考一個(gè) 當(dāng)我直接注釋之后會(huì)調(diào)用just的subscribeActual
public final class ObservableFromArray<T> extends Observable<T> {
final T[] array;
public ObservableFromArray(T[] array) {
this.array = array;
}
@Override
public void subscribeActual(Observer<? super T> s) {
FromArrayDisposable<T> d = new FromArrayDisposable<T>(s, array);
s.onSubscribe(d);
if (d.fusionMode) {
return;
}
d.run();
}
相信大家看過(guò)我之前的應(yīng)該可以看懂。
</pre>
讓我們回歸正題當(dāng)執(zhí)行到ObservableSubscribeOn的subscribeActual的方法的時(shí)候
public void subscribeActual(final Observer<? super T> s) {
final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);
s.onSubscribe(parent);
parent.setDisposable(scheduler.scheduleDirect(new Runnable() {
@Override
public void run() {
source.subscribe(parent);
}
}));
source.subscribe(parent); 看到這個(gè)方法了嗎颜阐、首先它是異步的平窘。另外執(zhí)行
.source.subscribe(parent);的時(shí)候 ,實(shí)際上就執(zhí)行了ObservableFromArray的subscribeActual
<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();
</pre>
剩下的就好理解了凳怨,都是分別執(zhí)行onnext等方法瑰艘。
到這里task的大體思路介紹完畢
2下面開始timer 定時(shí)器
public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
final Worker w = createWorker();
final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
w.schedule(new Runnable() {
@Override
public void run() {
try {
decoratedRun.run();
} finally {
w.dispose();
}
}
}, delay, unit);
return w;
}
重復(fù)的就不貼了是鬼。 都是差不多重復(fù)的。 只是給大家貼上關(guān)鍵代碼
看到這里面了嗎紫新。delay 就是大家貼上的時(shí)間均蜜。 詳細(xì)這個(gè)大家都是可以看明白的。芒率,
3interval
做周期性操作囤耳,從翻譯上大家就應(yīng)該可以看明白
ComputationScheduler的schedulePeriodicallyDirect的方法
<pre>
public Disposable schedulePeriodicallyDirect(@NonNull Runnable run, long initialDelay, long period, TimeUnit unit) {
PoolWorker w = pool.get().getEventLoop();
return w.schedulePeriodicallyDirect(run, initialDelay, period, unit);
}
</pre>
<pre>
NewThreadWorker的schedulePeriodicallyDirect的方法
public Disposable schedulePeriodicallyDirect(final Runnable run, long initialDelay, long period, TimeUnit unit) {
Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
try {
Future<?> f = executor.scheduleAtFixedRate(decoratedRun, initialDelay, period, unit);
return Disposables.fromFuture(f);
} catch (RejectedExecutionException ex) {
RxJavaPlugins.onError(ex);
return EmptyDisposable.INSTANCE;
}
}
</pre>
分別設(shè)置了 什么時(shí)候開始。多長(zhǎng)時(shí)間執(zhí)行一次
4buffer
Observable<List<String>> buffered = getObservable().buffer(3, 2);
buffered.subscribe(getObserver());
ObservableBuffer的subscribeActual的方法
<pre>
protected void subscribeActual(Observer<? super U> t) {
if (skip == count) {
BufferExactObserver<T, U> bes = new BufferExactObserver<T, U>(t, count, bufferSupplier);
if (bes.createBuffer()) {
source.subscribe(bes);
}
} else {
source.subscribe(new BufferSkipObserver<T, U>(t, count, skip, bufferSupplier));
}
}
</pre>
好吧到了關(guān)鍵的地方 source.subscribe是調(diào)用誰(shuí)的地方
Observable.just("one", "two", "three", "four", "five");
所以是ObservableFromArray的subscribeActual方法
<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>
看到這個(gè)for方法了嗎 這個(gè)就是決定你跳過(guò)的數(shù)量的偶芍。
5filter
這個(gè)相信大家很熟悉充择,對(duì)就是過(guò)濾
<pre>
fromArray(1, 0, 6)
.filter(new Predicate<Integer>() {
@Override
public boolean test(Integer integer) throws Exception {
return integer.intValue() > 5;
}
})
</pre>
這里只是放出來(lái)關(guān)鍵代碼
ObservableFilter的onNext
<pre>
public void onNext(T t) {
if (sourceMode == NONE) {
boolean b;
try {
b = filter.test(t);
} catch (Throwable e) {
fail(e);
return;
}
if (b) {
actual.onNext(t);
}
} else {
actual.onNext(null);
}
}
</pre>
這個(gè)b就是你的過(guò)濾條件。 下面的就是判斷匪蟀。 不符合的就不執(zhí)行 actual.onNext(t);其實(shí)很簡(jiǎn)單的方式
6skip
和上面同理關(guān)鍵部分ObservableSkip的onNext方法
<pre>
public void onNext(T t) {
if (remaining != 0L) {
remaining--;
} else {
actual.onNext(t);
}
}
</pre>
7 scan
RxJava的scan()函數(shù)可以看做是一個(gè)累加器函數(shù)椎麦。scan()函數(shù)對(duì)原始Observable發(fā)射的每一項(xiàng)數(shù)據(jù)都應(yīng)用一個(gè)函數(shù),它將函數(shù)的結(jié)果填充回可觀測(cè)序列材彪,等待和下一次發(fā)射的數(shù)據(jù)一起使用观挎。
關(guān)鍵代碼
<pre>
@Override
public void onNext(T t) {
if (done) {
return;
}
final Observer<? super T> a = actual;
T v = value;
if (v == null) {
value = t;
a.onNext(t);
} else {
T u;
try {
u = ObjectHelper.requireNonNull(accumulator.apply(v, t), "The value returned by the accumulator is null");
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
s.dispose();
onError(e);
return;
}
value = u;
a.onNext(u);
}
}
</pre>
執(zhí)行的時(shí)候value 會(huì)累加。 a.onNext(u);在發(fā)射出去
8 replay
<pre>
PublishSubject<Integer> source = PublishSubject.create();
ConnectableObservable<Integer> connectableObservable = source.replay(2); // bufferSize = 3 to retain 3 values to replay
connectableObservable.connect(); // connecting the connectableObservable
connectableObservable.subscribe(getFirstObserver());
source.onNext(1);
source.onNext(2);
source.onNext(3);
source.onNext(4);
source.onComplete();
/*
* it will emit 2, 3, 4 as (count = 3), retains the 3 values for replay
*/
connectableObservable.subscribe(getSecondObserver());
</pre>
replay 這個(gè)是緩存操作段化。
第二次訂閱之后嘁捷,就是緩存后面兩個(gè)數(shù)據(jù)
9concat
<pre>
final String[] aStrings = {"A1", "A2", "A3", "A4"};
final String[] bStrings = {"B1", "B2", "B3"};
final Observable<String> aObservable = Observable.fromArray(aStrings);
final Observable<String> bObservable = Observable.fromArray(bStrings);
Observable.concat(aObservable, bObservable)
.subscribe(getObserver());
</pre>
他的過(guò)程是
<pre>
return RxJavaPlugins.onAssembly(new ObservableConcatMap(fromArray(sources), Functions.identity(), bufferSize(), ErrorMode.BOUNDARY));
</pre>
concat操作符肯定也是有序的,實(shí)際上fromArray(sources)這么一個(gè)過(guò)程显熏。
10merge
<pre>
final String[] aStrings = {"A1", "A2", "A3", "A4"};
final String[] bStrings = {"B1", "B2", "B3"};
final Observable<String> aObservable = Observable.fromArray(aStrings);
final Observable<String> bObservable = Observable.fromArray(bStrings);
Observable.merge(aObservable, bObservable)
.subscribe(getObserver());
</pre>
無(wú)序的合并
11distinct 去除重復(fù)的
<pre>
enum HashSetCallable implements Callable<Set<Object>> {
INSTANCE;
@Override
public Set<Object> call() throws Exception {
return new HashSet<Object>();
}
}
</pre>
HashSet中 是不允許重復(fù)元素的
12last
<pre>
private void doSomeWork() {
getObservable().last("A1") // the default item ("A1") to emit if the source ObservableSource is empty
.subscribe(getObserver());
}
private Observable<String> getObservable() {
return Observable.just("A1", "A2", "A3", "A4", "A5", "A6");
}
打印出來(lái)的是a6
</pre>
ObservableFromArray的run方法
<pre>
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>
ObservableLastSingle的onComplete
<pre>
public void onComplete() {
s = DisposableHelper.DISPOSED;
T v = item;
if (v != null) {
item = null;
actual.onSuccess(v);
} else {
v = defaultItem;
if (v != null) {
actual.onSuccess(v);
} else {
actual.onError(new NoSuchElementException());
}
}
</pre>
last方法會(huì)返回Single
13throttleFirst
<pre>
private void doSomeWork() {
getObservable()
.throttleFirst(500, TimeUnit.MILLISECONDS)
// Run on a background thread
.subscribeOn(Schedulers.io())
// Be notified on the main thread
.observeOn(AndroidSchedulers.mainThread())
.subscribe(getObserver());
}
private Observable<Integer> getObservable() {
return Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
// send events with simulated time wait
Thread.sleep(0);
emitter.onNext(1); // skip
emitter.onNext(2); // deliver
Thread.sleep(505);
emitter.onNext(3); // skip
Thread.sleep(99);
emitter.onNext(4); // skip
Thread.sleep(100);
emitter.onNext(5); // skip
emitter.onNext(6); // deliver
Thread.sleep(305);
emitter.onNext(7); // deliver
Thread.sleep(510);
emitter.onComplete();
}
});
}
</pre>
從這個(gè)可以理解到發(fā)送第一個(gè)之后雄嚣。剩下的500之后才會(huì)接受第二個(gè)
14throttleLast
從這個(gè)可以看出來(lái),這是在一段時(shí)間內(nèi)接受最后一個(gè)數(shù)據(jù)
<pre>
getObservable()
.throttleLast(500, TimeUnit.MILLISECONDS)
// Run on a background thread
.subscribeOn(Schedulers.io())
// Be notified on the main thread
.observeOn(AndroidSchedulers.mainThread())
.subscribe(getObserver());
}
private Observable<Integer> getObservable() {
return Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
// send events with simulated time wait
Thread.sleep(0);
emitter.onNext(1); // skip
emitter.onNext(2); // deliver
Thread.sleep(505);
emitter.onNext(3); // skip
Thread.sleep(99);
emitter.onNext(4); // skip
Thread.sleep(100);
emitter.onNext(5); // skip
emitter.onNext(6); // deliver
Thread.sleep(305);
emitter.onNext(7); // deliver
Thread.sleep(510);
emitter.onComplete();
}
});
}
</pre>
15debounce
<pre>
getObservable()
.debounce(500, TimeUnit.MILLISECONDS)
// Run on a background thread
.subscribeOn(Schedulers.io())
// Be notified on the main thread
.observeOn(AndroidSchedulers.mainThread())
.subscribe(getObserver());
}
private Observable<Integer> getObservable() {
return Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
// send events with simulated time wait
emitter.onNext(1); // skip
Thread.sleep(400);
emitter.onNext(2); // deliver
Thread.sleep(505);
emitter.onNext(3); // skip
Thread.sleep(100);
emitter.onNext(4); // deliver
Thread.sleep(605);
emitter.onNext(5); // deliver
Thread.sleep(510);
emitter.onComplete();
}
});
</pre>
這個(gè)接受的一一個(gè)時(shí)間跨度之內(nèi)的數(shù)據(jù)
16window
可以看出來(lái)大概 的意思就是截取被觀察者組成一個(gè)新的被觀察者