乐趣区

RxJava20转载

jiezi

零、来源

来源：Carson_Ho- 简书

一、基础知识

角色 作用 类比
被观察者（Observable） 产生事件 顾客
观察者（Observer） 接收事件，并给出响应动作 厨房
订阅（Subscribe） 连接 被观察者 & 观察者 服务员
事件（Event） 被观察者 & 观察者 沟通的载体 菜式


二、基础使用

1. 导入连接 

implementation 'io.reactivex.rxjava2:rxjava:2.2.19'
implementation 'io.reactivex.rxjava2:rxandroid:2.1.1'

2. 创建被观察者 

   // 创建被观察者，产生事件
    public Observable<Integer> createObservable() {Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {e.onNext(1);
                e.onNext(2);
                e.onNext(3);
                e.onComplete();}
        });
        return observable;
    }

3. 创建观察者 

    // 创建观察者
    public Observer<Integer> createObserver() {Observer<Integer> observer = new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {Log.v("lanjiabinRx", "onSubscribe 连接");
            }

            @Override
            public void onNext(Integer value) {Log.v("lanjiabinRx", "onNext" + value + "事件");
            }

            @Override
            public void onError(Throwable e) {Log.v("lanjiabinRx", "onError 事件");
            }

            @Override
            public void onComplete() {Log.v("lanjiabinRx", "onComplete 事件");
            }
        };
        return observer;
    }

4. 建立 subscribe() 连接 

     // 观察者订阅被观察者
    public void createSubscribe() {createObservable().subscribe(createObserver());
    }

5. 调用和结果 

 createSubscribe();

6. 链式调用 

     // 链式调用
    public void chainCall() {Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {e.onNext(1);
                e.onNext(2);
                e.onNext(3);
                e.onComplete();}
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {Log.v("lanjiabinRx", "onSubscribe 连接");
            }

            @Override
            public void onNext(Integer value) {Log.v("lanjiabinRx", "onNext" + value + "事件");
            }

            @Override
            public void onError(Throwable e) {Log.v("lanjiabinRx", "onError 事件");
            }

            @Override
            public void onComplete() {Log.v("lanjiabinRx", "onComplete 事件");
            }
        });
    }

6. 切断连接

即观察者 无法继续 接收 被观察者的事件，但被观察者还是可以继续发送事件 

Disposable mDisposable; //1. 定义

@Override
public void onSubscribe(Disposable d) {
       mDisposable=d; //2. 赋值
       Log.v("lanjiabinRx", "onSubscribe 连接");
}

@Override
public void onNext(Integer value) {if (value==2) mDisposable.dispose(); //3. 在第二个 next 事件断开连接
      Log.v("lanjiabinRx", "onNext" + value + "事件");
}

三、创建操作符

0. 总图


1. create（基础发送）

最基础的创建 


    //1.create
    public void chainCall() {Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {e.onNext(1);
                e.onNext(2);
                e.onNext(3);
                e.onComplete();}
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {Log.v("lanjiabinRx", "onSubscribe 连接");
            }

            @Override
            public void onNext(Integer value) {Log.v("lanjiabinRx", "onNext" + value + "事件");
            }

            @Override
            public void onError(Throwable e) {Log.v("lanjiabinRx", "onError 事件");
            }

            @Override
            public void onComplete() {Log.v("lanjiabinRx", "onComplete 事件");
            }
        });
    }

2. just（立刻发送 10 以下）

  • 快速创建 1 个被观察者对象（Observable）
  • 发送事件的特点：直接发送传入的事件
  • 最多只能发送十个参数
  • 应用场景：快速创建 被观察者对象（Observable）& 发送 10 个以下事件 

    //2.just
    public void justDo() {Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {Log.d("lanjiabinRx", "开始采用 subscribe 连接");
            }

            @Override
            public void onNext(Integer integer) {Log.d("lanjiabinRx", "接受的事件 onNext =" + integer);
            }

            @Override
            public void onError(Throwable e) {Log.d("lanjiabinRx", "接受的事件 onError =" + e.getMessage());
            }

            @Override
            public void onComplete() {Log.d("lanjiabinRx", "接受的事件 onComplete");
            }
        });
    }

结果：

3. fromArray（数组发送）

  • 快速创建 1 个被观察者对象（Observable）
  • 发送事件的特点：直接发送 传入的数组数据
  • 会将数组中的数据转换为 Observable 对象

应用场景：
1. 快速创建 被观察者对象（Observable）& 发送 10 个以上事件（数组形式）
2. 数组元素遍历 


    //3.fromArray
    public void fromArrayDo() {Integer[] items = {0, 1, 2, 3, 4};
        Observable.fromArray(items).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {Log.v("lanjiabinRx", "开始采用 subscribe 连接");
            }

            @Override
            public void onNext(Integer integer) {Log.v("lanjiabinRx", "接受的事件 onNext =" + integer);
            }

            @Override
            public void onError(Throwable e) {Log.v("lanjiabinRx", "接受的事件 onError =" + e.getMessage());
            }

            @Override
            public void onComplete() {Log.v("lanjiabinRx", "接受的事件 onComplete");
            }
        });
    }

结果：

4. fromIterable（集合发送）

  • 快速创建 1 个被观察者对象（Observable）
  • 发送事件的特点：直接发送 传入的集合 List 数据
  • 会将数组中的数据转换为 Observable 对象

应用场景：
1. 快速创建 被观察者对象（Observable）& 发送 10 个以上事件（集合形式）
2. 集合元素遍历 


    //4.fromIterable
    public void fromIterableDo(){List<Integer> list = new ArrayList<>();
        list.add(1);
        list.add(2);
        list.add(3);
        Observable.fromIterable(list).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {Log.v("lanjiabinRx", "开始采用 subscribe 连接");
            }

            @Override
            public void onNext(Integer integer) {Log.v("lanjiabinRx", "接受的事件 onNext =" + integer);
            }

            @Override
            public void onError(Throwable e) {Log.v("lanjiabinRx", "接受的事件 onError =" + e.getMessage());
            }

            @Override
            public void onComplete() {Log.v("lanjiabinRx", "接受的事件 onComplete");
            }
        });
    }

结果：

// 下列方法一般用于测试使用

<-- empty()  -->
// 该方法创建的被观察者对象发送事件的特点：仅发送 Complete 事件，直接通知完成
Observable observable1=Observable.empty(); 
// 即观察者接收后会直接调用 onCompleted（）<-- error()  -->
// 该方法创建的被观察者对象发送事件的特点：仅发送 Error 事件，直接通知异常
// 可自定义异常
Observable observable2=Observable.error(new RuntimeException())
// 即观察者接收后会直接调用 onError（）<-- never()  -->
// 该方法创建的被观察者对象发送事件的特点：不发送任何事件
Observable observable3=Observable.never();
// 即观察者接收后什么都不调用

5. defer（获取最新数据）

  • 直到有观察者（Observer）订阅时，才动态创建被观察者对象（Observable）& 发送事件
  • 通过 Observable 工厂方法创建被观察者对象（Observable）
  • 每次订阅后，都会得到一个刚创建的最新的 Observable 对象，这可以确保 Observable 对象里的数据是最新的

应用场景：
动态创建被观察者对象（Observable）& 获取最新的 Observable 对象数据 

   //5.defer
    Integer i = 10; // 第一次赋值

    public void deferDo() {Observable<Integer> observable = Observable.defer(new Callable<ObservableSource<? extends Integer>>() {

            @Override
            public ObservableSource<? extends Integer> call() throws Exception {return Observable.just(i);
            }
        });

        i = 15; // 第二次赋值
        observable.subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {Log.v("lanjiabinRx", "开始采用 subscribe 连接");
            }

            @Override
            public void onNext(Integer integer) {Log.v("lanjiabinRx", "接受的事件 onNext =" + integer);
            }

            @Override
            public void onError(Throwable e) {Log.v("lanjiabinRx", "接受的事件 onError =" + e.getMessage());
            }

            @Override
            public void onComplete() {Log.v("lanjiabinRx", "接受的事件 onComplete");
            }
        });
    }

结果：得到最新赋值的数字，说明取到了最新的数据

6. timer（延迟发送）

  • 快速创建 1 个被观察者对象（Observable）
  • 发送事件的特点：延迟指定时间后，发送 1 个数值 0（Long 类型）
  • 本质 = 延迟指定时间后，调用一次 onNext(0)

应用场景：
延迟指定事件，发送一个 0，一般用于检测 


    //6.timer
    public void timerDo() {
        // 注：timer 操作符默认运行在一个新线程上
        // 也可自定义线程调度器（第 3 个参数）：timer(long,TimeUnit,Scheduler)
        //TimeUnit.SECONDS 延迟 2s 后，发送一个 0
     
        /**
         * timer(long delay, TimeUnit unit)
         * delay 数值
         * unit 单位
         * 下面就是 2 数值，单位为秒，所以是 2 秒
         * */
        Observable.timer(2, TimeUnit.SECONDS).subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {Log.v("lanjiabinRx", "开始采用 subscribe 连接");
            }

            @Override
            public void onNext(Long aLong) {
                /*
                * 得到的结果为 0 一般用于检测
                * */
                Log.v("lanjiabinRx", "接受的事件 onNext =" + aLong);
            }

            @Override
            public void onError(Throwable e) {Log.v("lanjiabinRx", "接受的事件 onError =" + e.getMessage());
            }

            @Override
            public void onComplete() {Log.v("lanjiabinRx", "接受的事件 onComplete");
            }
        });
    }

结果：

7. interval（周期发送，无限）

  • 快速创建 1 个被观察者对象（Observable）
  • 发送事件的特点：每隔指定时间 就发送 事件
  • 发送的事件序列 = 从 0 开始、无限递增 1 的的整数序列 

    //7.interval
    public void intervalDo() {
        /**
         * 从 0 开始递增
         *
         * @param initialDelay（Long）*          初始延迟时间（第一次延迟时间）* @param period（Long）*          后续数字发射之间的时间间隔（一个周期时间）* @param unit
         *          时间单位
         * */
        Observable.interval(3, 2, TimeUnit.SECONDS).subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {Log.v("lanjiabinRx", "开始采用 subscribe 连接");
            }

            @Override
            public void onNext(Long aLong) {Log.v("lanjiabinRx", "接受的事件 onNext =" + aLong);
            }

            @Override
            public void onError(Throwable e) {Log.v("lanjiabinRx", "接受的事件 onError =" + e.getMessage());
            }

            @Override
            public void onComplete() {Log.v("lanjiabinRx", "接受的事件 onComplete");
            }
        });
    }

结果：

8. intervalRange（周期发送，有限，指定数据）

  • 作用类似于 interval（），但可指定发送的数据的数量 
     //8.intervalRange
    public void intervalRangeDo() {
        /**
         *
         * @param start 起始值
         * @param count 总共要发送的值的数量，如果为零，则运算符将在初始延迟后发出 onComplete
         * @param initialDelay 发出第一个值（开始）之前的初始延迟
         * @param period 后续值之间的时间段
         * @param unit 时间单位
         * */
        Observable.intervalRange(3, 10, 2, 1, TimeUnit.SECONDS)
                .subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {Log.v("lanjiabinRx", "开始采用 subscribe 连接");
            }

            @Override
            public void onNext(Long aLong) {Log.v("lanjiabinRx", "接受的事件 onNext =" + aLong);
            }

            @Override
            public void onError(Throwable e) {Log.v("lanjiabinRx", "接受的事件 onError =" + e.getMessage());
            }

            @Override
            public void onComplete() {Log.v("lanjiabinRx", "接受的事件 onComplete");
            }
        });
    }

结果：3-12 经历 10 个数

9. range (无延迟，Integer 类型指定数据)

  • 作用类似于 intervalRange（），但区别在于：无延迟发送事件 
    //9.range
    public void rangeDo(){
        /**
         * @param start
         *            序列中第一个 Integer 的值
         * @param count
         *           要生成的顺序整数的数量
         * */
        Observable.range(3,5).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {Log.v("lanjiabinRx", "开始采用 subscribe 连接");
            }

            @Override
            public void onNext(Integer integer) {Log.v("lanjiabinRx", "接受的事件 onNext =" + integer);
            }

            @Override
            public void onError(Throwable e) {Log.v("lanjiabinRx", "接受的事件 onError =" + e.getMessage());
            }

            @Override
            public void onComplete() {Log.v("lanjiabinRx", "接受的事件 onComplete");
            }
        });
    }

结果：

10. rangeLong（无延迟，Long 类型指定数据）


    //10.rangeLong
    public void rangeLongDo(){
        /**
         * @param start
         *            Long 类型，序列中第一个 Integer 的值
         * @param count
         *           Long 类型，要生成的顺序整数的数量
         * */
        Observable.rangeLong(3,8).subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {Log.v("lanjiabinRx", "开始采用 subscribe 连接");
            }

            @Override
            public void onNext(Long aLong) {Log.v("lanjiabinRx", "接受的事件 onNext =" + aLong);
            }

            @Override
            public void onError(Throwable e) {Log.v("lanjiabinRx", "接受的事件 onError =" + e.getMessage());
            }

            @Override
            public void onComplete() {Log.v("lanjiabinRx", "接受的事件 onComplete");
            }
        });
    }

结果：

编程中我们会遇到多少挫折？表放弃，沙漠尽头必是绿洲。

退出移动版