RxJava20转载

3次阅读

共计 8749 个字符，预计需要花费 22 分钟才能阅读完成。

来源：Carson_Ho- 简书

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

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

   // 创建被观察者，产生事件
    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;
    }

    // 创建观察者
    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;
    }

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

 createSubscribe();

     // 链式调用
    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 事件");
            }
        });
    }

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

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 + "事件");
}

最基础的创建


    //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 事件");
            }
        });
    }

快速创建 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");
            }
        });
    }

结果：

快速创建 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");
            }
        });
    }

结果：

快速创建 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();
// 即观察者接收后什么都不调用

直到有观察者（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");
            }
        });
    }

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

快速创建 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");
            }
        });
    }

结果：

快速创建 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");
            }
        });
    }

结果：

作用类似于 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 个数

作用类似于 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
    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");
            }
        });
    }

结果：

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

正文完

android

发表至： android

2020-07-07

0

关于android:Android-嵌套-Intent

关于android:iOS-零代码快速集成AGC崩溃服务

关于android:安卓逆向-35-安卓程序执行入口

关于android:Android-多返回栈技术详解

20200707-每天一个小技巧变形汉堡按钮Hamburger-Menu

RxJava20转载

零、来源

一、基础知识

二、基础使用

1. 导入连接

2. 创建被观察者

3. 创建观察者

4. 建立 subscribe() 连接

5. 调用和结果

6. 链式调用

6. 切断连接

三、创建操作符

0. 总图

1. create（基础发送）

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

3. fromArray（数组发送）

4. fromIterable（集合发送）

5. defer（获取最新数据）

6. timer（延迟发送）

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

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

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

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

站内搜索