Rxjava创建操作符

目录

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>