工作以后，一直保持着学习学习再学习的心态，却总是发现，学不完啊，我滴天~

当前Rxjava2版本2.2.2，操作符快达到四百多个，学习rxjava2多半就是学习其中的操作符使用，所以使用博客来记录学习！

一、 创建型操作符

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

不会发送任何数据，订阅之后直接发送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

将一些数组对象转换成被观察者，按照顺序发射

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环境下运行）

间隔一段恒定的时间无限制循环执行

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

发出特定数据的被观察者

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

创建一个发出特定范围的连续整数的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

创建一个多次发出特定项目的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环境下运行）

创建一个Observable，在给定的延迟后发出特定的数据

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