RxJava操作符 ~ defer_rxjava defer

RxJava的 defer 操作符作以确保Observable代码在被订阅后才执行,而不是创建后立即执行

@Test
    public void defer() {

        Disposable disposable = Observable.defer(new Callable<ObservableSource<?>>() {
            @Override
            public ObservableSource<?> call() throws Exception {
                return getRxJavaCreateExampleData();
            }
        })

                .subscribeOn(Schedulers.io())
                .unsubscribeOn(Schedulers.io())
                .observeOn(Schedulers.io())
                .subscribe(new Consumer<Object>() {
                    @Override
                    public void accept(Object o) throws Exception {
                        System.out.println("rxJavaDeferExample:" + Thread.currentThread().getName() + " consumer --> " + o.toString());
                    }
                });
    }


    public Observable<Integer> getRxJavaCreateExampleData() {
        return Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) {
                System.out.println("getRxJavaCreateExampleData:" + Thread.currentThread().getName() + " --> " + 1);
                emitter.onNext(1);

                System.out.println("getRxJavaCreateExampleData:" + Thread.currentThread().getName() + " --> " + 2);
                emitter.onNext(2);

                System.out.println("getRxJavaCreateExampleData:" + Thread.currentThread().getName() + " --> " + 3);
                emitter.onNext(3);
                emitter.onComplete();

                System.out.println("getRxJavaCreateExampleData:" + Thread.currentThread().getName() + " --> " + 4);
                emitter.onNext(4);

            }
        });
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42