编程语言
首页 > 编程语言> > RxJava、RxBus学习

RxJava、RxBus学习

作者:互联网

 

使用介绍:

步骤1:创建Observable(被观察者),并生产事件

        // 1. 创建被观察者 Observable 对象
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
          // create() 是 RxJava 最基本的创造事件序列的方法
          // 此处传入了一个 OnSubscribe 对象参数
          // 当 Observable 被订阅时,OnSubscribe 的 call() 方法会自动被调用,即事件序列就会依照设定依次被触发
          // 即观察者会依次调用对应事件的复写方法从而响应事件
          // 从而实现被观察者调用了观察者的回调方法 & 由被观察者向观察者的事件传递,即观察者模式

        // 2. 在复写的subscribe()里定义需要发送的事件
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                // 通过 ObservableEmitter类对象产生事件并通知观察者
                // ObservableEmitter类介绍
                    // a. 定义:事件发射器
                    // b. 作用:定义需要发送的事件 & 向观察者发送事件
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        });

<--扩展:RxJava 提供了其他方法用于 创建被观察者对象Observable -->
// 方法1:just(T...):直接将传入的参数依次发送出来
  Observable observable = Observable.just("A", "B", "C");
  // 将会依次调用:
  // onNext("A");
  // onNext("B");
  // onNext("C");
  // onCompleted();

// 方法2:from(T[]) / from(Iterable<? extends T>) : 将传入的数组 / Iterable 拆分成具体对象后,依次发送出来
  String[] words = {"A", "B", "C"};
  Observable observable = Observable.from(words);
  // 将会依次调用:
  // onNext("A");
  // onNext("B");
  // onNext("C");
  // onCompleted();

  

步骤2:创建Observer(观察者),用于响应事件

Observer接口、Subscribe抽象类对象都可以

 

    <--方式1:采用Observer 接口 -->
        // 1. 创建观察者 (Observer )对象
        Observer<Integer> observer = new Observer<Integer>() {
        // 2. 创建对象时通过对应复写对应事件方法 从而 响应对应事件

            // 观察者接收事件前,默认最先调用复写 onSubscribe()
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "开始采用subscribe连接");
            }
            
            // 当被观察者生产Next事件 & 观察者接收到时,会调用该复写方法 进行响应
            @Override
            public void onNext(Integer value) {
                Log.d(TAG, "对Next事件作出响应" + value);
            }

            // 当被观察者生产Error事件& 观察者接收到时,会调用该复写方法 进行响应
            @Override
            public void one rror(Throwable e) {
                Log.d(TAG, "对Error事件作出响应");
            }
          
            // 当被观察者生产Complete事件& 观察者接收到时,会调用该复写方法 进行响应
            @Override
            public void onComplete() {
                Log.d(TAG, "对Complete事件作出响应");
            }
        };

<--方式2:采用Subscriber 抽象类 -->
// 说明:Subscriber类 = RxJava 内置的一个实现了 Observer 的抽象类,对 Observer 接口进行了扩展

// 1. 创建观察者 (Observer )对象
Subscriber<String> subscriber = new Subscriber<Integer>() {

// 2. 创建对象时通过对应复写对应事件方法 从而 响应对应事件
            // 观察者接收事件前,默认最先调用复写 onSubscribe()
            @Override
            public void onSubscribe(Subscription s) {
                Log.d(TAG, "开始采用subscribe连接");
            }

            // 当被观察者生产Next事件 & 观察者接收到时,会调用该复写方法 进行响应
            @Override
            public void onNext(Integer value) {
                Log.d(TAG, "对Next事件作出响应" + value);
            }

            // 当被观察者生产Error事件& 观察者接收到时,会调用该复写方法 进行响应
            @Override
            public void one rror(Throwable e) {
                Log.d(TAG, "对Error事件作出响应");
            }

            // 当被观察者生产Complete事件& 观察者接收到时,会调用该复写方法 进行响应
            @Override
            public void onComplete() {
                Log.d(TAG, "对Complete事件作出响应");
            }
        };


<--特别注意:2种方法的区别,即Subscriber 抽象类与Observer 接口的区别 -->
// 相同点:二者基本使用方式完全一致(实质上,在RxJava的 subscribe 过程中,Observer总是会先被转换成Subscriber再使用)
// 不同点:Subscriber抽象类对 Observer 接口进行了扩展,新增了两个方法:
    // 1. onStart():在还未响应事件前调用,用于做一些初始化工作
    // 2. unsubscribe():用于取消订阅。在该方法被调用后,观察者将不再接收 & 响应事件
    // 调用该方法前,先使用 isUnsubscribed() 判断状态,确定被观察者Observable是否还持有观察者Subscriber的引用,如果引用不能及时释放,就会出现内存泄露

  步骤3:通过Subscribe(订阅),连接观察者和被观察者

observable.subscribe(observer);
 // 或者 observable.subscribe(subscriber);

  

 

实际操作中:

// RxJava的链式操作
        Observable.create(new ObservableOnSubscribe<Integer>() {
        // 1. 创建被观察者 & 生产事件
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        }).subscribe(new Observer<Integer>() {
            // 2. 通过通过订阅(subscribe)连接观察者和被观察者
            // 3. 创建观察者 & 定义响应事件的行为
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "开始采用subscribe连接");
            }
            // 默认最先调用复写的 onSubscribe()

            @Override
            public void onNext(Integer value) {
                Log.d(TAG, "对Next事件"+ value +"作出响应"  );
            }

            @Override
            public void one rror(Throwable e) {
                Log.d(TAG, "对Error事件作出响应");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "对Complete事件作出响应");
            }

        });
    }
}

注:整体方法调用顺序:观察者.onSubscribe()> 被观察者.subscribe()> 观察者.onNext()>观察者.onComplete() 

  

参考:https://www.jianshu.com/nb/14302692

标签:Override,onNext,观察者,public,学习,RxBus,事件,RxJava,void
来源: https://www.cnblogs.com/meizixiong/p/13224503.html