工作以后，一直保持著學(xué)習(xí)學(xué)習(xí)再學(xué)習(xí)的心態(tài)，卻總是發(fā)現(xiàn)，學(xué)不完啊，我滴天~

當(dāng)前Rxjava2版本2.2.2，操作符快達(dá)到四百多個(gè)，學(xué)習(xí)rxjava2多半就是學(xué)習(xí)其中的操作符使用，所以使用博客來記錄學(xué)習(xí)！

一、 創(chuàng)建型操作符

create

create.c.png

    @Test
    public void testCreate() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                System.out.println(integer);
            }
        });
    }

// 打印結(jié)果
1
Process finished with exit code 0

defer

在訂閱之后，與之相關(guān)的數(shù)據(jù)才會(huì)生效

defer.c.png

int i = 10;
@Test
public void testDefer() {
    Observable<Integer> deferObservable = 
        Observable.defer(new Callable<ObservableSource<Integer>>() {
        @Override
        public ObservableSource<Integer> call() throws Exception {
            return new ObservableSource<Integer>() {
                @Override
                public void subscribe(Observer<? super Integer> observer) {
                    observer.onNext(i);
                }
            };
        }
    });
    i = 20;
    deferObservable.subscribe(new Consumer<Integer>() {
        @Override
        public void accept(Integer integer) throws Exception {
            System.out.println("當(dāng)前輸出值為： " + integer);
        }
    });
}
// 輸出結(jié)果
當(dāng)前輸出值為： 20
Process finished with exit code 0

Empty

不會(huì)發(fā)送任何數(shù)據(jù)，訂閱之后直接發(fā)送onComplete事件

empty.c.png

@Test
public void testEmpty() {
    Observable.empty()
            .subscribe(new Observer<Object>() {
                @Override
                public void onSubscribe(Disposable d) {
                    System.out.println("onSubscribe()");
                }
                @Override
                public void onNext(Object o) {
                    System.out.println("onNext()");
                }
                @Override
                public void onError(Throwable e) {
                    System.out.println("onError()");
                }
                @Override
                public void onComplete() {
                    System.out.println("onComplete()");
                }
            });
}
// 運(yùn)行結(jié)果：
onSubscribe()
onComplete()

Process finished with exit code 0

Never

訂閱之后不會(huì)走任何回調(diào)，常用于測試

never.c.png

@Test
public void testNever() {
    Observable.never()
            .subscribe(new Observer<Object>() {
                @Override
                public void onSubscribe(Disposable d) {
                    System.out.println("onSubscribe()");
                }
                @Override
                public void onNext(Object o) {
                    System.out.println("onNext()");
                }
                @Override
                public void onError(Throwable e) {
                    System.out.println("onError()");
                }
                @Override
                public void onComplete() {
                    System.out.println("onComplete()");
                }
            });
}

// 運(yùn)行結(jié)果
onSubscribe()

Process finished with exit code 0

FromArray

將一些數(shù)組對象轉(zhuǎn)換成被觀察者，按照順序發(fā)射

from.c.png

@Test
public void testFrom() {
    Integer[] arr = {1, 2, 3, 4};
    Observable.fromArray(arr)
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    System.out.println(integer);
                }
            });
}
// 運(yùn)行結(jié)果
1
2
3
4

Process finished with exit code 0

Interval（在Android環(huán)境下運(yùn)行）

間隔一段恒定的時(shí)間無限制循環(huán)執(zhí)行

interval.c.png

Observable.interval(10, 1, TimeUnit.SECONDS)
        .subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {
            }
            @Override
            public void onNext(Long aLong) {
                Log.d(TAG, "當(dāng)前為： " + aLong);
            }
            @Override
            public void onError(Throwable e) {
            }
            @Override
            public void onComplete() {
            }
        });
// 運(yùn)行結(jié)果：只有當(dāng)程序終止才會(huì)停止
D/RxjavaActivity: 當(dāng)前為： 0
D/RxjavaActivity: 當(dāng)前為： 1
D/RxjavaActivity: 當(dāng)前為： 2
D/RxjavaActivity: 當(dāng)前為： 3
D/RxjavaActivity: 當(dāng)前為： 4
D/RxjavaActivity: 當(dāng)前為： 5
D/RxjavaActivity: 當(dāng)前為： 6
D/RxjavaActivity: 當(dāng)前為： 7
D/RxjavaActivity: 當(dāng)前為： 8
D/RxjavaActivity: 當(dāng)前為： 9
D/RxjavaActivity: 當(dāng)前為： 10
...
Application terminated.

Just

發(fā)出特定數(shù)據(jù)的被觀察者

just.c.png

@Test
public void testJust() {
    Observable.just(1, 2, 3)
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    System.out.println(integer);
                }
            });
}

// 運(yùn)行結(jié)果
1
2
3

Process finished with exit code 0

Range

創(chuàng)建一個(gè)發(fā)出特定范圍的連續(xù)整數(shù)的Observable

range.c.png

@Test
public void testRange() {
    Observable.range(/*start*/10, /*count*/4)
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    System.out.println(integer);
                }
            });
}
// 運(yùn)行結(jié)果
10
11
12
13

Process finished with exit code 0

Repeat

創(chuàng)建一個(gè)多次發(fā)出特定項(xiàng)目的Observable

repeat.c.png

@Test
public void testRepeat() {
    Observable.just(1, 2, 3)
            .repeat(/*times*/2)
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    System.out.println(integer);
                }
            });
}

// 運(yùn)行結(jié)果
1
2
3
1
2
3

Process finished with exit code 0

Timer （在Android環(huán)境下運(yùn)行）

創(chuàng)建一個(gè)Observable，在給定的延遲后發(fā)出特定的數(shù)據(jù)

timer.c.png

Observable.timer(1, TimeUnit.SECONDS)
        .subscribe(new Consumer<Long>() {
            @Override
            public void accept(Long aLong) throws Exception {
                System.out.println(aLong);
            }
        });

// 運(yùn)行結(jié)果
  
I/System.out: 0