RxJava 1与RxJava 2的API上的区别

作者: itbird01 | 来源:发表于2018-03-26 19:22 被阅读99次

    RxJava 2相比于RxJava 1,改动还是比较大的,这里我们来简单说一下API上的改变

    1.Flowable

    RxJava1 中 Observable 不能很好地支持 backpressure ,最常见的例如UI事件,而不处理backpressure有可能导致MissingBackpressureException的出现。所以在 RxJava2 中 Oberservable 不再支持 backpressure ,而使用新增的 Flowable 来支持 backpressure 。
    Flowable的用法跟原先的Observable是一样的。

    2.ActionN 和 FuncN 改名

    ActionN 和 FuncN 遵循Java 8的命名规则。
    其中,Action0 改名成Action,Action1改名成Consumer,而Action2改名成了BiConsumer,而Action3 - Action9都不再使用了,ActionN变成了Consumer<Object[]> 。

    同样,Func改名成Function,Func2改名成BiFunction,Func3 - Func9 改名成 Function3 - Function9,FuncN 由 Function<Object[], R> 取代。

    3.Observable.OnSubscribe 变成 ObservableOnSubscribe

    RxJava1的写法:

    Observable.create(new Observable.OnSubscribe<String>() {
     
     @Override
     public void call(Subscriber<? super String> subscriber) {
      subscriber.onNext("hello");
     }
     
    }).subscribe(new Action1<String>() {
     
     @Override
     public void call(String s) {
      System.out.println(s);
     }
    });
    

    RxJava2的写法:

    Observable.create(new ObservableOnSubscribe<String>() {
     
     @Override
     public void subscribe(ObservableEmitter<String> e) throws Exception {
      e.onNext("hello");
     }
     
    }).subscribe(new Consumer<String>() {
     
     @Override
     public void accept(String s) {
      System.out.println(s);
     }
    });
    

    4. ObservableOnSubscribe 中使用 ObservableEmitter 发送数据给 Observer

    结合上一条,ObservableOnSubscribe 不再使用 Subscriber 而是用 ObservableEmitter 替代。

    ObservableEmitter 可以理解为发射器,是用来发出事件的,它可以发出三种类型的事件,通过调用emitter的onNext(T value) 、onComplete()和onError(Throwable error)可以分别发出next事件、complete事件和error事件。 如果只关心next事件的话,只需单独使用onNext()即可。

    需要特别注意,emitter的onComplete()调用后,Consumer不再接收任何next事件。

    5. Observable.Transformer 变成 ObservableTransformer

    RxJava1的写法:

    /**
     * 跟compose()配合使用,比如ObservableUtils.wrap(obj).compose(toMain())
     * @param <T>
     * @return
     */
    public static <T> Observable.Transformer<T, T> toMain() {
     return new Observable.Transformer<T, T>() {
      @Override
      public Observable<T> call(Observable<T> tObservable) {
       return tObservable
         .subscribeOn(Schedulers.io())
         .observeOn(AndroidSchedulers.mainThread());
      }
     };
    }
    

    RxJava2的写法:

    /**
     * 跟compose()配合使用,比如ObservableUtils.wrap(obj).compose(toMain())
     * @param <T>
     * @return
     */
    public static <T> ObservableTransformer<T, T> toMain() {
     
     return new ObservableTransformer<T, T>() {
     
      @Override
      public ObservableSource<T> apply(Observable<T> upstream) {
       return upstream.subscribeOn(Schedulers.io())
         .observeOn(AndroidSchedulers.mainThread());
      }
     };
    }
    

    由于新增了Flowable,同理也增加了FlowableTransformer

    public static <T> FlowableTransformer<T, T> toMain() {
     
     return new FlowableTransformer<T, T>() {
     
      @Override
      public Publisher<T> apply(Flowable<T> upstream) {
       return upstream.subscribeOn(Schedulers.io())
         .observeOn(AndroidSchedulers.mainThread());
      }
     };
    }
    

    6.Subscription 改名为 Disposable

    在 RxJava2 中,由于已经存在了 org.reactivestreams.subscription 这个类,为了避免名字冲突将原先的 rx.Subscription 改名为 io.reactivex.disposables.Disposable 。

    刚开始不知道,在升级 RxJava2 时发现 org.reactivestreams.subscription 这个类完全没法做原先 rx.Subscription 的事情:(

    顺便说下,Disposable必须单次使用,用完就要销毁。

    7. first() 用法改变

    官方文档是这么描述的first()的用法

    1.x 2.x
    first() RC3 renamed to firstElement and returns Maybe<T>
    first(Func1) dropped, use filter(predicate).first()
    firstOrDefault(T) renamed to first(T) and RC3 returns Single<T>
    firstOrDefault(Func1, T) renamed to first(T) and RC3 returns Single<T>

    first(Func1)为例,first(Func1)后面还使用了push() ,原先 Rxjava1会这样写

    ConnectableObservable<Data> connectableObservable = Observable
       .concat(Observable.from(list))
       .first(new Func1<Data, Boolean>() {
        @Override
        public Boolean call(Data data) {
         return DataUtils.isAvailable(data);
        }
       }).publish();
    

    RxJava2 改成这样

    ConnectableObservable<Data> connectableObservable = Observable
       .concat(Observable.fromIterable(list))
       .filter(new Predicate<Data>() {
     
        @Override
        public boolean test(@NonNull Data data) throws Exception {
         return DataUtils.isAvailable(data);
        }
       }).firstElement().toObservable().publish();
    

    8. toBlocking().y 被 blockingY() 取代

    在我的框架中存在着一个Optional类,它跟Java 8的Optional作用差不多,原先是使用RxJava1来编写的。

    import rx.Observable;
     
    /**
     * 使用方法:
     *   String s = null;
     *   Optional.ofNullable(s).orElse("default")); // 如果s为null,则显示default,否则显示s的值
     * @author Tony Shen
     *
     */
    public class Optional<T> {
     
     Observable<T> obs;
     
     public Optional(Observable<T> obs) {
      this.obs = obs;
     }
     
     public static <T> Optional<T> of(T value) {
      if (value == null) {
       throw new NullPointerException();
      } else {
       return new Optional<T>(Observable.just(value));
      }
     }
     
     public static <T> Optional<T> ofNullable(T value) {
      if (value == null) {
       return new Optional<T>(Observable.<T>empty());
      } else {
       return new Optional<T>(Observable.just(value));
      }
     }
     
     public T get() {
      return obs.toBlocking().single();
     }
     
     public T orElse(T defaultValue) {
      return obs.defaultIfEmpty(defaultValue).toBlocking().single();
     }
    }
    

    升级到RxJava2之后,get()orElse()方法都会报错,修改之后是这样的。

    import io.reactivex.Observable;
     
    /**
     * 使用方法:
     *   String s = null;
     *   Optional.ofNullable(s).orElse("default"); // 如果s为null,则显示default,否则显示s的值
     * @author Tony Shen
     *
     */
    public class Optional<T> {
     
     Observable<T> obs;
     
     public Optional(Observable<T> obs) {
      this.obs = obs;
     }
     
     public static <T> Optional<T> of(T value) {
      if (value == null) {
       throw new NullPointerException();
      } else {
       return new Optional<T>(Observable.just(value));
      }
     }
     
     public static <T> Optional<T> ofNullable(T value) {
      if (value == null) {
       return new Optional<T>(Observable.<T>empty());
      } else {
       return new Optional<T>(Observable.just(value));
      }
     }
     
     public T get() {
     
      return obs.blockingSingle();
     }
     
     public T orElse(T defaultValue) {
     
      return obs.defaultIfEmpty(defaultValue).blockingSingle();
     }
    }
    
    

    9. PublishSubject

    包括 PublishSubject 以及各种 Subject(ReplaySubject、BehaviorSubject、AsyncSubject) 都不再支持backpressure。

    总结

    RxJava2 所带来的变化远远不止这些,以后遇到的话还会继续整理和总结,毕竟我使用的 RxJava2 还是很少的一部分内容。

    RxJava2 最好到文档依然是官方文档。如果是新项目到话,可以毫不犹豫地使用RxJava2,如果是在线上已经成熟稳定的项目,可以再等等。对于新手的话,可以直接从 RxJava2 学起,RxJava1 就直接略过吧。对于老手,RxJava2 还是使用原来的思想,区别不大,从 RxJava1 迁移到 Rxjava2 也花不了多少工夫。

    相关文章

      网友评论

        本文标题:RxJava 1与RxJava 2的API上的区别

        本文链接:https://www.haomeiwen.com/subject/cetdcftx.html