目录
1. Rx简介
2. Rxjava基础知识
3. Rxjava创建操作符
4. Rxjava的线程操作
5. Rxjava变换过滤操作符
6. Rxjava合并、链接操作符
7. Rxjava条件和布尔操作符
(一):前言
Rxjava有很多操作符，正是因为这些操作符的作用，才能使Rxjava更加灵活多变。Rxjava中的而操作付包含如图。
(二):Rxjava创建操作符
| 操作符 | 作用 |
|---|---|
| just() | 将一个或多个对象转换成发射这个或这些对象的一个Observable |
| form() | 将一个Iterable,一个Future或者一个数组装换成一个Observable |
| create() | 使用一个函数从头创建一个observable |
| defer() | 只有当订阅者订阅才创建observable |
| range() | 创建一个发射指定范围的整数序列的Observable |
| interval() | 创建一个按照给定的时间间隔发射整数序列的Observable |
| timer() | 创建一个在给定的延迟之后发射单个数据的Observable |
| empty() | 创建一个什么都不做直接通知完成的Observable |
| error() | 创建一个什么都不做直接通知错误的Observable |
| never() | 创建一个不发送任何数据的Observable |
1:onCreate():
使用一个函数从头创建一个Observable，这个是最常用的一个，因为我们可以用这种方式，自己定义要发射什么数据。他的参数类里包含了一个发射器，我们可以利用发射器的函数onNext, onError, onComplete来发射我们想要发射的数据。
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
//emitter 就是发射器。
emitter.onNext("tian");
emitter.onComplete();
}
});
2:just
将单个数据装换发射，这里不用我们去构建发射步骤，直接发射just的参数。它可以接受1到10个参数，按照参数序列逐个发送。如果参数为null，他会报出null指针异常。
Observable.just("tian").subscribe(s -> System.out.println(s));
Observable.just(1,2,3,4,5,6,7).subscribe(s -> System.out.println(s));
3:from
from可以将其他种类的对象和数据类型转换为Observable，其中包括Furure, Iterable和数组。from有与数据对应的创建操作符，formArray() fromIterable, fromFuture().
//不管是多少个参数，observable都将他们当做一个数组，一项一项发送
Observable.fromArray("hello","tian").subscribe(s -> System.out.println(s));
List<Integer> items = new ArrayList<>();
for (int i = 0; i<10; i++) items.add(i);
//这里是处理标准的数组的。
Observable.fromIterable(items).subscribe(i -> System.out.println(i));
对于fromFuture，了解多线程的都知道，Future是多线程执行的返回，利用fromFuture创建的Observable，会发射future.get()方法的返回值。
public static void main(String[] args) {
ExecutorService executorService = Executors.newCachedThreadPool();
Future<String> future = executorService.submit(new MyCallBack());
//这里是使用，fromFuture还有两个可选参数，分别是时长和时间单位，
//就是规定的时长后future没有返回则Observable发射错误通知并终止。
Observable.fromFuture(future).subscribe(s -> System.out.println(s));
}
static class MyCallBack implements Callable<String> {
@Override
public String call() {
System.out.println("模拟一些任务");
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "ok";
}
}
4：repeat(int n) ； n:重复次数
创建一个发射特定数据重复多次的Observable，，它不是创建一个Observable，而是重复发射原始的Observable数据，
Observable.just("tian").repeat(3).subscribe(s -> System.out.println(s));
repeatWhen()
它不是缓存和重放原始的Observable,而是根据参数，有条件的重新订阅和发射原始数据。当原始Observable有终止通知时，repeatWhen就会决定是否重新订阅Observable，当RepeatWhen先执行完，就会等待与那时的Observable发射完成事件，在决定是否重新订阅。
Observable.range(0,9).repeatWhen(new Function<Observable<Object>, ObservableSource<?>>() {
@Override
public ObservableSource<?> apply(Observable<Object> objectObservable) throws Exception {
return Observable.timer(10, TimeUnit.SECONDS); //这里确定了是否重新订阅，timer后续介绍
}
}).subscribe(s -> System.out.println(s));
Thread.sleep(12000);
repeatUntil()
这个是2.0版本新的操作符，表示直到达到某一条件就不在重复发射数据，类似与循环的while条件语句。接受BooleanSupplier,返回false，重复发射数据。返回true，则终止发射数据。
long startTime = System.currentTimeMillis();
Observable.interval(500, TimeUnit.MILLISECONDS)
.take(5)
.repeatUntil(new BooleanSupplier() {
@Override
public boolean getAsBoolean() throws Exception {
return System.currentTimeMillis() - startTime > 5000;
}
}).subscribe(s-> System.out.println(s));
Thread.sleep(6000);
5：defer:知道有观察者订阅才创建Observable，并为每一个订阅者创建一个全新的Observable。
6：interval：创建一个按照固定时间间隔发射整数序列的Observable，从0开始，
Observable.interval(500, TimeUnit.MILLISECONDS) //500，时间间隔，第二个参数时间单位。
.subscribe(s-> System.out.println(s)); //输出0,1,2,3,4,5......
7:timer:
创建一个Observable，在一个给定的延迟后发射一个特殊的值。返回的特殊值为0
Observable.timer(2,TimeUnit.SECONDS)
.subscribe(s->System.out.println(s));
7:range(n,m)
Range操作符发射一个范围内的有序整数序列，你可以指定范围的起始和长度。
range函数，它接受两个参数，一个是范围的起始值，一个是范围的数据的数目
rxjava有许多创建操作符，合理的利用可以使Observable更灵活，本文的创建操作符，Flowable也都可以使用。</font>
