工作以后配阵，一直保持著學習學習再學習的心態(tài)馏颂，卻總是發(fā)現示血，學不完啊，我滴天~

當前Rxjava2版本2.2.2救拉，操作符快達到四百多個难审，學習rxjava2多半就是學習其中的操作符使用，所以使用博客來記錄學習亿絮！

一告喊、 創(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);
            }
        });
    }

// 打印結果
1
Process finished with exit code 0

defer

在訂閱之后，與之相關的數據才會生效

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("當前輸出值為： " + integer);
        }
    });
}
// 輸出結果
當前輸出值為： 20
Process finished with exit code 0

Empty

不會發(fā)送任何數據派昧，訂閱之后直接發(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()");
                }
            });
}
// 運行結果：
onSubscribe()
onComplete()

Process finished with exit code 0

Never

訂閱之后不會走任何回調黔姜，常用于測試

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()");
                }
            });
}

// 運行結果
onSubscribe()

Process finished with exit code 0

FromArray

將一些數組對象轉換成被觀察者，按照順序發(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);
                }
            });
}
// 運行結果
1
2
3
4

Process finished with exit code 0

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

間隔一段恒定的時間無限制循環(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, "當前為： " + aLong);
            }
            @Override
            public void onError(Throwable e) {
            }
            @Override
            public void onComplete() {
            }
        });
// 運行結果：只有當程序終止才會停止
D/RxjavaActivity: 當前為： 0
D/RxjavaActivity: 當前為： 1
D/RxjavaActivity: 當前為： 2
D/RxjavaActivity: 當前為： 3
D/RxjavaActivity: 當前為： 4
D/RxjavaActivity: 當前為： 5
D/RxjavaActivity: 當前為： 6
D/RxjavaActivity: 當前為： 7
D/RxjavaActivity: 當前為： 8
D/RxjavaActivity: 當前為： 9
D/RxjavaActivity: 當前為： 10
...
Application terminated.

Just

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

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);
                }
            });
}

// 運行結果
1
2
3

Process finished with exit code 0

Range

創(chuàng)建一個發(fā)出特定范圍的連續(xù)整數的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);
                }
            });
}
// 運行結果
10
11
12
13

Process finished with exit code 0

Repeat

創(chuàng)建一個多次發(fā)出特定項目的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);
                }
            });
}

// 運行結果
1
2
3
1
2
3

Process finished with exit code 0

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

創(chuàng)建一個Observable蒂萎，在給定的延遲后發(fā)出特定的數據

timer.c.png

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

// 運行結果
  
I/System.out: 0