零基础理解RxJava和响应式编程

Posted 2021-04-30 Android每日一讲

这是一篇能帮助你在10分钟内从零到一理解RxJava和响应式编程的文章。

RxJava发展到现在已经在2016年推出了第二代。可能你听说过很多人讲起RxJava，但是很少在实际项目开发中用到它。
原因很简单，RxJava虽然很好用，但是它有一定的学习成本。很多人只是知道这么个东西，但是没有真正的去学习和推动RxJava。毕竟会觉得即使没有RxJava也一样能写好代码。
其实它的学习成本和带来的收益对比的话，是非常值得花点时间去学的。当你切换到Rx编程思维之后，会发现很多以前难以处理的问题在响应式编程下都变得易如反掌。
而很多公司没有推进RxJava的原因，主要在于船大难掉头。笔者见过一个上亿日活的项目，至今还在用ant构建。可想而知还有许多新技术受限于项目的历史原因没法应用。
另一个推动RxJava困难的原因在于开发团队水平层次不齐。如果你的团队里有成员连并发和线程都搞不清楚的话，RxJava可能只能带来负面效果。

那为什么还是越来越多的人在谈论RxJava呢，这要从响应式编程说起。

响应式编程

只要你有一些开发经验就有体会，不管是什么系统，几乎都不可能用同步操作来编写代码。放在android上更是这样。
用户交互，屏幕旋转，网络请求，这些都是异步源。
异步源的存在让传统的命令式编程变的复杂，虽然Java是变向对象语言，但某种意义上它还是命令式编程。想象一下这么个场景，你需要等到用户输入搜索key后，开始网络请求，并把返回结果显示到UI上，而此时用户可能还会旋转屏幕。要完成这么个需求，你可以想象下要写多少个handler去做线程间的切换。而这些handler就是我们开发者需要去介入的代码逻辑，也是我们头痛的地方。

响应式编程的英文是Reactive Programming。回想一下开发中的listener，可以认为是响应式编程的例子，某个按钮被点击，listener回调，然后发起网络请求，网络请求可以看成是按钮点击的响应。而网络请求返回数据，到数据更新到UI上，也是一种响应。

RxJava的核心思想在这里，通常我们需要编写大量代码来控制响应的逻辑，有时候还需要处理一些异常情况，比如请求失败，数据为空的处理，也有可能用户已经切换出当前界面，这时候就需要处理一下把数据更新到一个已经被回收的UI上可能发生的问题。这些网络请求，UI互动，数据库，磁盘IO，都可以看成异步源，而Android本身是也是一个大异步源，像Push通知，屏幕旋转，锁屏开屏这些操作都会触发异步行为。Reactive编程认为我们不应该介入异步源之间的响应和交互，而应该直接的把异步源和响应绑定起来。

用响应式的思路写代码

我们从一个简单的例子开始，UserManager用来定义用户名字/年纪，也可以获取用户对象

interface UserManager {
    User getUser();
    void setName(String name);
    void setAge(int age);
}

UserManager um = new UserManager();
System.out.println(um.getUser());

um.setName("Jane Doe");
System.out.println(um.getUser());

这段代码没什么问题，但当它以异步的形式发展的话问题就来了。假设setName是个异步操作，它往服务器提交了个String。作为开发者可以选择相信服务器每次提交都是成功的(并不可能)，提交完成后照样println输出。也可以选择靠谱的方式，在服务器提交成功后才输出本地。
我们可以在每次提交的时候设置个runnable作为成功的回调，这是很简单的响应式编程思想，确保网络成功的时候才去更新数据。

interface UserManager {
    User getUser();
    void setName(String name, Runnable callback);A
    void setAge(int age, Runnable callback);B
}
UserManager um = new UserManager();
System.out.println(um.getUser());

um.setName("Jane Doe", new Runnable() {
    @Override public void run() {
        System.out.println(um.getUser());
    }
});

但这里并没处理网络失败的情况，更完善的代码需要像下面这样子，增加对错误处理的响应，

um.setName("Jane Doe", new UserManager.Listener() {
    @Override public void success() {
        System.out.println(um.getUser());
    }

    @Override public void failure(IOException e) {
        // TODO show the error...
    }
});

你作为开发者可以选择在failure的时候做响应，比如弹出提示，自动重新请求。在传统的命令式编程中不可避免地需要在单线程逻辑中混入这些异步代码。

而且在Android开发中这些都发生在Context里，比方Activity，我们把这段代码放到Activity里去看看，

public final class UserActivity extends Activity {
    private final UserManager um = new UserManager();

    @Override
  protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);

        setContentView(R.layout.user);
        TextView tv = (TextView) findViewById(R.id.user_name);
        tv.setText(um.getUser().toString());

        um.setName("Jane Doe", new UserManager.Listener() {
          @Override
      public void success() {
              tv.setText(um.getUser().toString());
          }
          @Override
      public void failure(IOException e) {
              // TODO show the error...
              }
        });
    }
}

我们现在还没考虑线程的问题。像更新UI的操作，众所周知需要在UI线程进行，上面的网络请求代码 setName 可能在子线程返回，我们需要让这些地方跳回到主线程去。

public final class UserActivity extends Activity {
    private final UserManager um = new UserManager();

    @Override
  protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);

        setContentView(R.layout.user);
        TextView tv = (TextView) findViewById(R.id.user_name);
        tv.setText(um.getUser().toString());

        um.setName("Jane Doe", new UserManager.Listener() {
            @Override
      public void success() {
                runOnUiThread(new Runnable() {
                    @Override public void run() {
                        if (isDestroyed()) {
                            tv.setText(um.getUser().toString());
                        }
                    }
                });
            }
            @Override
      public void failure(IOException e) {
                // TODO show the error...
            }
        });
    }
}

到这里我们还没有考虑更深的问题，比如TextView在回调中是否被回收了。因为这是个异步操作，不能指望网络请求结束时还处于刚才的界面。
你看这堆东西，我们用一大段的代码只是保证了一个简单的UI更新操作，并且还没有考虑实际场景中的问题。一旦把它投入到实际应用中，你就需要处理一堆繁琐的代码。
传统的响应式代码繁琐，而且阅读麻烦，却又不可避免。但你只需要记住这种异步源+响应的思路，就能轻松地切换到RxJava的世界。

Observable和Observer

RxJava中的其中一个重点对象是Observable-被观察者。如果用上面的例子来理解，网络请求就是一个Observable。没错，你可以把所有的异步源都看成Observable。
Reactive编程中另外一个重点对象是 Observer，它要观察，和响应Observable发生的事情，进而做出对应的响应。在上面的例子中，println事件就是一种响应，而UI更新，tv.setText也是一种响应。

概括起来RxJava就是用Observer和Observable把响应和异步源包裹起来，然后通过 subscribe(订阅)，将两者进行绑定。一段绑定异步源和响应的代码像下面这样，不过下面这段代码没有任何的现实意义，只是用来展示绑定的过程，

Observable.create(new Observable.OnSubscribe<String>() {
      @Override
      public void call(final Subscriber<? super String> subscriber) {
           //TODO
      })
      .subscribe(new Observer<String>() {
            @Override
            public void onCompleted() {
                //TODO
            }

            @Override
            public void onError(Throwable e) {
                //TODO
            }

            @Override
            public void onNext(String s) {
                //TODO
            }
        });

我们这里的代码用的是RxJava作为例子，在RxJava2的时候相关的接口改了一些，虽然名字变了，不过思路还是一样。

如果你是第一次接触链式调用，可能这段代码看起来不容易理解，我们转换成下面这样

Observable observable = Observable.create(new Observable.OnSubscribe(){...});
Observer observer = new Obvserver(){...};
observable.subscribe(observer);

在RxJava中，异步源作为被观察者被 Observable包括起来，它的逻辑要写到 call 方法里，当 subscribe被调用的时候会触发 call。
observer则是响应。这里在call方法的参数里出现了一个新的对象 subscriber，这其实就是 Observer，Observer是个接口，subscriber是实现了这个接口的类。Observer有三个方法分别用来作为异步操作完成后的回调，

public interface Observer<T> {
    void onCompleted();

    void onError(Throwable var1);

    void onNext(T var1);
}

从名字上立马能理解的是onError，那就是异步失败了。
onCompleted() 则是用来表示异步操作结束，
onNext(T var1) 用来把异步操作结果回传给 observer的 onNext(T value) 方法。
这里有个点要注意，onComplted调用之后再调用 onNext的话，observer的onNext也不会被调动，在RxJava中对于onCompleted的定义就是操作已经完成，此时应该要回收掉observer。有兴趣的话可以深入去看看RxJava的源码。

RxJava进阶

关于RxJava还有很多点没有说，比如变换的概念，因为篇幅的关系这些我们下一次再继续讲。
这篇文章的重点在于理解Reactive编程思维，将我们的命令式编程的习惯切换到响应式编程上来。如果你已经能轻松的用Reactive思维看代码的话，恭喜你已经入门了，RxJava写的代码现在你也能轻易地理解。只要再花点时间，就能轻松的在你的项目中应用上RxJava了。