RxJava 的基本概念和基本实现

Posted 2021-11-11 wdmxzf

文章目录

• 四个基本概念
• 回调方法
• 基本实现
• • Observer 观察者
  • Subscriber 订阅者
  • Observable 被观察者
  • Subscribe （订阅）
  • Action
• 总结

RxJava 是一个在 Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库。总之一句话就是异步。
它是一种扩展的观察者模式。

四个基本概念

• Observable：被观察者
• Observer：观察者
• subscribe：订阅
• 事件

回调方法

• onNext()：相当于 onClick() / onEvent()
• onCompleted()：当不会有新的 onNext()方法触发时调用
• onError()：时间队列异常，同时队列自动终止，且和onCompleted() 方法只能触发其中一个。

基本实现

Observer 观察者

Observer<String> observer = new Observer<String>() { 
    @Override 
    public void onNext(String s) { Log.d(tag, "Item: " + s); } 
    @Override 
    public void onCompleted() { Log.d(tag, "Completed!"); } 
    @Override 
    public void onError(Throwable e) { Log.d(tag, "Error!"); }
};

Subscriber 订阅者

Subscriber 对 Observer 接口进行了一些扩展，但他们的基本使用方式是完全一样的

Subscriber<String> subscriber = new Subscriber<String>() { 
    @Override 
    public void onNext(String s) { Log.d(tag, "Item: " + s); } 
    @Override 
    public void onCompleted() { Log.d(tag, "Completed!"); } 
    @Override 
    public void onError(Throwable e) { Log.d(tag, "Error!"); }
};

额外方法：

• onStart()：这是 Subscriber 新增方法，会在 subscribe 刚开始，而事件还未发送之前被调用，可以用于做一些准备工作，如果有线程要求，则不适用，因为总是在subscibe发生的线程被调用。
• unsubscribe()：用于取消订阅。在这个方法被调用后，Subscriber将不再接收事件；可用 isUnsubscribed() 判断是否订阅。在 subscribe() 之后， Observable 会持有 Subscriber 的引用，所以在合适的地方调用该方法，防止内存泄漏。

Observable 被观察者

• create()创建

Observable observable = Observable.create(new Observable.OnSubscribe<String>() { 
    @Override 
    public void call(Subscriber<? super String> subscriber) {
         subscriber.onNext("Hello"); 
        subscriber.onNext("Hi"); 
        subscriber.onNext("Aloha"); 
        subscriber.onCompleted(); 
    }
});
//这里传入一个 OnSubscribe 对象作为参数，OnSubscribe 会被存储在返回的
// Observable 对象中，相当一个计划表。Observable 被订阅的时候，
// OnSubscribe 的 call() 方法会自动被调用，事件序列就会依照设定依次触发。

• just(T...)将传入的参数依次发送过来

Observable observable = Observable.just("Hello", "Hi", "Aloha");
// 将会依次调用：
// onNext("Hello");
// onNext("Hi");
// onNext("Aloha");
// onCompleted();

• from(T[])/from(Iterable<? extends T>) 将传入的数组或 Iterable 拆分成具体对象后，依次发送出来

String[] words = {"Hello", "Hi", "Aloha"};
Observable observable = Observable.from(words);
// 将会依次调用：
// onNext("Hello");
// onNext("Hi");
// onNext("Aloha");
// onCompleted();

其中 just(T…) 和 from(T[]) 和 create(OnSubscribe)是等价的。

Subscribe （订阅）

创建 Observable 和 Observer 之后，再用 subscribe() 方法链接起来

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

被观察者，订阅观察者。当被订阅的时候，是 subscribe() 方法执行的时候

除了 subscribe(Observer) 和 subscribe(Subscriber) ，subscribe() 还支持不完整定义的回调：

Action

Action1<String> onNextAction = new Action1<String>() {
    // onNext()
    @Override
    public void call(String s) {
        Log.d(tag, s);
    }
};
Action1<Throwable> onErrorAction = new Action1<Throwable>() {
    // onError()
    @Override
    public void call(Throwable throwable) {
        // Error handling
    }
};
Action0 onCompletedAction = new Action0() {
    // onCompleted()
    @Override
    public void call() {
        Log.d(tag, "completed");
    }
};

// 自动创建 Subscriber ，并使用 onNextAction 来定义 onNext()
observable.subscribe(onNextAction);
// 自动创建 Subscriber ，并使用 onNextAction 和 onErrorAction 来定义 onNext() 和 onError()
observable.subscribe(onNextAction, onErrorAction);
// 自动创建 Subscriber ，并使用 onNextAction、 onErrorAction 和 onCompletedAction 来定义 onNext()、 onError() 和 onCompleted()
observable.subscribe(onNextAction, onErrorAction, onCompletedAction);

• Action0
  只有 call() 一个回调方法，且是无参无返回值的。

onCompleted() 是无参无返回的函数，所以 Action0 可以定义不完整回调

• Action1
  只有 call(T param) 一个回调方法，且是有一个参数 and 无返回值的。

onNext(T obj) 和 onError(Throwable error) 是有参数的，所以 Action1 可以定义不完整回调

总结

RXjava 是个好东西，但是学习起来还是有些难度，想要使用好它，只有在实战中不断的去实践。
以上这些是我以前整理的知识，希望对大家有用。最后推荐一个博客： 干货集中营