RxJava2介绍

RxJava2 发布已经有一段时间了，是对 RxJava 的一次重大的升级，由于我的一个库cv4j使用了 RxJava2 来尝鲜，但是 RxJava2 跟 RxJava1 是不能同时存在于一个项目中的，逼不得已我得把自己所有框架中使用 RxJava 的地方以及
App 中使用 RxJava 的地方都升级到最新版本。所以我整理并记录了一些已经填好的坑。分享出来供大家参考学习，下面来看看详细的介绍：

填坑记录

1. RxJava1 跟 RxJava2 不能共存

如果，在同一个module中同时使用RxJava1和RxJava2，类似如下：

compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
compile 'io.reactivex.rxjava2:rxjava:2.0.7'
compile 'io.reactivex:rxandroid:1.2.0'
compile 'io.reactivex:rxjava:1.1.5'

那么，很不幸你会遇到这样的错误

同理，在 App 中如果使用了 Rxjava2，但是某个第三方的 library 还在使用 Rxjava1 也会遇到同样的错误。

上面的错误是因为 RxAndroid 2.0.1 本身依赖了 RxJava 2.0.1。我们尝试去掉对 RxJava 的依赖，只留下 RxAndroid 。还是会遇到问题。

compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
//compile 'io.reactivex.rxjava2:rxjava:2.0.7'
compile 'io.reactivex:rxandroid:1.2.0'
//compile 'io.reactivex:rxjava:1.1.5'

所以使用RxAndroid不能去掉对RxJava的依赖，我是这样使用的。

compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
compile 'io.reactivex.rxjava2:rxjava:2.0.7'

官方也是这样解释的

Because RxAndroid releases are few and far between, it is recommended you also
explicitly depend on RxJava's latest version for bug fixes and new features.

最后，我建议要升级到 RxJava2 的时候必须所有使用的地方都要升级，并且用最新的版本。

2. 新增Flowable

RxJava1 中 Observable 不能很好地支持 backpressure ，会抛出MissingBackpressureException。所以在 RxJava2 中 Oberservable 不再支持 backpressure ，而使用新增的 Flowable 来支持 backpressure 。

Flowable的用法跟原先的Observable是一样的。

3. 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> 取代。

4. 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);
   }
  });

现在的写法：

  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);
   }
  });

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

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

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

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

6. 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());
   }
  };
 }

现在的写法：

 /**
  * 跟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());
   }
  };
 }

7. Subscription 改名为 Disposable

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

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

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

8. first() 用法改变

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

以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();

9. 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();
 }
}

10. PublishSubject

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

总结

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

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

好了，以上就是这篇文章的全部内容了，希望本文的内容对大家的学习或者工作能带来一定的帮助，如果有疑问大家可以留言交流，谢谢大家对的支持。

# rxjava1升级rxjava2  # rxjava1和rxjava2  # rxjava的坑  # Android性能优化之利用Rxlifecycle解决RxJava内存泄漏详解  # RxJava入门之介绍与基本运用  # RxJava入门指南及其在Android开发中的使用示例  # Android中的Retrofit+OkHttp+RxJava缓存架构使用  # Android上传多张图片的实例代码(RxJava异步分发)  # RxJava2.x实现定时器的实例代码  # 简单谈谈RxJava和多线程并发  # Android 使用 RxJava2 实现倒计时功能的示例代码  # RxJava+Retrofit+OkHttp实现多文件下载之断点续传  # 你用不惯 RxJava  # 只因缺了这把钥匙(推荐)  # 升级到  # 使用了  # 文档  # 我是  # 是在  # 很好  # 是因为  # 成了  # 好了  # 也会  # 还在  # 都要  # 逼不得已  # 你会  # 是这样  # 还会  # 只需  # 线上  # 三种  # 来看看 

相关栏目： 【 网站优化151355 】 【 网络推广146373 】 【 网络技术251813 】 【 AI营销90571 】

相关推荐： Laravel如何升级到最新的版本_Laravel版本升级流程与兼容性处理  网站广告牌制作方法,街上的广告牌，横幅，用PS还是其他软件做的？  如何在阿里云ECS服务器部署织梦CMS网站？  米侠浏览器网页图片不显示怎么办 米侠图片加载修复  JS去除重复并统计数量的实现方法  专业型网站制作公司有哪些,我设计专业的，谁给推荐几个设计师兼职类的网站？  Laravel怎么清理缓存_Laravel optimize clear命令详解  Laravel如何使用缓存系统提升性能_Laravel缓存驱动和应用优化方案  公司网站制作需要多少钱,找人做公司网站需要多少钱？  Laravel如何理解并使用服务容器（Service Container）_Laravel依赖注入与容器绑定说明  php json中文编码为null的解决办法  如何快速选择适合个人网站的云服务器配置？  微信小程序 wx.uploadFile无法上传解决办法  Laravel怎么配置自定义表前缀_Laravel数据库迁移与Eloquent表名映射【步骤】  如何在 React 中条件性地遍历数组并渲染元素  Laravel怎么创建自己的包(Package)_Laravel扩展包开发入门到发布  如何在 Pandas 中基于一列条件计算另一列的分组均值  PHP 实现电台节目表的智能时间匹配与今日/明日轮播逻辑  广州网站制作公司哪家好一点,广州欧莱雅百库网络科技有限公司官网？  如何在万网自助建站平台快速创建网站？  香港服务器WordPress建站指南：SEO优化与高效部署策略  nodejs redis 发布订阅机制封装实现方法及实例代码  Android中AutoCompleteTextView自动提示  手机钓鱼网站怎么制作视频,怎样拦截钓鱼网站。怎么办？  Windows10电脑怎么设置虚拟光驱_Win10右键装载ISO镜像文件  HTML5打空格有哪些误区_新手常犯的空格使用错误【技巧】  专业企业网站设计制作公司,如何理解商贸企业的统一配送和分销网络建设？  Midjourney怎么调整光影效果_Midjourney光影调整方法【指南】  如何在阿里云完成域名注册与建站？  Laravel如何处理表单验证？（Requests代码示例）  如何快速上传建站程序避免常见错误？  如何在HTML表单中获取用户输入并用JavaScript动态控制复利计算循环  微信推文制作网站有哪些,怎么做微信推文，急？  如何快速搭建支持数据库操作的智能建站平台？  如何快速搭建高效WAP手机网站吸引移动用户？  Laravel如何生成API文档？（Swagger/OpenAPI教程）  Laravel用户密码怎么加密_Laravel Hash门面使用教程  北京企业网站设计制作公司,北京铁路集团官方网站？  企业网站制作这些问题要关注  邀请函制作网站有哪些,有没有做年会邀请函的网站啊？在线制作，模板很多的那种？  如何在云主机快速搭建网站站点？  标题：Vue + Vuex 项目中正确使用 JWT 进行身份认证的实践指南  简单实现Android验证码  Laravel怎么实现软删除SoftDeletes_Laravel模型回收站功能与数据恢复【步骤】  Laravel Eloquent访问器与修改器是什么_Laravel Accessors & Mutators数据处理技巧  如何快速搭建自助建站会员专属系统？  如何用wdcp快速搭建高效网站？  如何在香港服务器上快速搭建免备案网站？  Laravel如何实现多语言支持_Laravel本地化与国际化(i18n)配置教程  深圳防火门网站制作公司,深圳中天明防火门怎么编码？