手把手教你写一个基于 RxJava 的扩展框架

背景

现在 RxJava 在 Android 开发中可谓时炽手可热，其受欢迎程度不言而喻，也因此在 github 上出现了一系列的基于 RxJava 的框架，如 RxBinding、RxPermissions、RxCache 等等，这些框架极大的方便了我们在开发中对 RxJava 的应用，那么怎样去写一个基于 RxJava 的框架呢？接下来我们就来分析一下 RxBinding 的源码，然后仿照它的源码写一款基于 RxJava 的 Toast 框架。

RxBinding

我们知道，RxBinding 是通过 RxView.clicks() 方法来订阅点击事件的，那么我们就从这个方法入手，打开 clicks() 方法可以看到

@CheckResult @NonNull
public static Observable<Object> clicks(@NonNull View view) {
    checkNotNull(view, "view == null");
    return new ViewClickObservable(view);
}

这个方法非常简单，首先调用 checkNotNull 方法来判断传入的 view 是不是 null，如果是 null，就抛出一个 NullPointerException，然后创建一个 ViewClickObservable 对象并返回，接下来看看这个 ViewClickObservable 的源码

final class ViewClickObservable extends Observable<Object> {
  private final View view;

  ViewClickObservable(View view) {
    this.view = view;
  }

  @Override protected void subscribeActual(Observer<? super Object> observer) {
    if (!checkMainThread(observer)) {
      return;
    }
    Listener listener = new Listener(view, observer);
    observer.onSubscribe(listener);
    view.setOnClickListener(listener);
  }

  static final class Listener extends MainThreadDisposable implements OnClickListener {
    private final View view;
    private final Observer<? super Object> observer;

    Listener(View view, Observer<? super Object> observer) {
      this.view = view;
      this.observer = observer;
    }

    @Override public void onClick(View v) {
      if (!isDisposed()) {
        observer.onNext(Notification.INSTANCE);
      }
    }

    @Override protected void onDispose() {
      view.setOnClickListener(null);
    }
  }
}

这个类也非常简单，首先它是一个 final 的，说明不能被继承，它继承自 Observable，并重写它的 subscribeActual 方法，进入 Observable 里面会发现这是一个抽象方法，然后找到它在 Observable 中被调用的位置

@SchedulerSupport(SchedulerSupport.NONE)
@Override
public final void subscribe(Observer<? super T> observer) {
    ObjectHelper.requireNonNull(observer, "observer is null");
    try {
        observer = RxJavaPlugins.onSubscribe(this, observer);

        ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");

        subscribeActual(observer);
    } catch (NullPointerException e) { // NOPMD
        throw e;
    } catch (Throwable e) {
        Exceptions.throwIfFatal(e);
        // can't call onError because no way to know if a Disposable has been set or not
        // can't call onSubscribe because the call might have set a Subscription already
        RxJavaPlugins.onError(e);

        NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
        npe.initCause(e);
        throw npe;
    }
}

发现这个方法在 subscribe 方法中被调用，相信这个方法大家都很熟悉了，也就是说在 Observable 被创建之后，当我们调用它的 subscribe 方法的时候会触发它的 subscribeActual 方法。

接着往下看，ViewClickObservable 中有一个内部类 Listener ,它继承自 MainThreadDisposable 并实现了 OnClickListener 接口，我们发现它重写了 onDispose 方法，进入 MainThreadDisposable 中发现这又是一个抽象方法，再来看看它是在哪里被调用的

@Override
public final void dispose() {
    if (unsubscribed.compareAndSet(false, true)) {
        if (Looper.myLooper() == Looper.getMainLooper()) {
            onDispose();
        } else {
            AndroidSchedulers.mainThread().scheduleDirect(new Runnable() {
                @Override public void run() {
                    onDispose();
                }
            });
        }
    }
}

我们发现它在 dispose 方法中被调用，相信这个方法大家也都很熟悉了，也就是说，在我们调用 dispose 方法来取消订阅的时候会触发 onDispose 方法。

另外由于 MainThreadDisposable 实现了 OnClickListener 接口，所以必须重写它的 onClick 方法，在 onClick 方法中，调用 observer 的 onNext 方法来发送一个事件，这里发送的对象 Notification.INSTANCE 是一个枚举的单例。

最后再来看一下 ViewClickObservable 的 subscribeActual 方法，首先调用 checkMainThread 方法，从这个方法的名字我们大致就能猜到它是用来判断当前线程是不是主线程的，看一下它的源码

public static boolean checkMainThread(Observer<?> observer) {
    if (Looper.myLooper() != Looper.getMainLooper()) {
        observer.onError(new IllegalStateException(
                "Expected to be called on the main thread but was " + Thread.currentThread().getName()));
        return false;
    }
    return true;
}

我们知道，每个线程中只能有一个 Looper 对象，通过 myLooper 获取到的是当前线程，让它与通过 getMainLooper 获取的主线程比较来判断当前是不是在主线程上，如果不是，发送一个 error 事件并返回 false，回到刚才的 subscribeActual 方法中，如果在主线程中，就创建一个 Listener 对象，并调用 observer 的 onSubscribe 方法和 view 的 setOnClickListener 方法。

接下来我们再对整个过程做一下总结，分析一下 view 的点击事件是怎么一步一步被订阅的。首先调用 clicks 方法的时候会创建一个 ViewClickObservable 对象并返回，通过 ViewClickObservable 的构造方法来传入 view 的对象，ViewClickObservable 是 Observable 的子类，当我们调用它的 subscribe 方法的时候，subscribeActual 方法会被触发，然后，判断一下当前是不是在主线程中，如果不是就直接发送一个 error 事件，这里为什么要判断是不是主线程呢，因为在接下来的操作中，需要调用 setOnClickListener 来绑定监听器，我们知道，这个操作是必须在主线程中的，然后调用 observer 的 onSubscribe 来告诉我们已经订阅成功，事实上订阅是在下一步的 setOnClickListener 中进行的，订阅成功之后，当我们点击到 view 的时候，就会触发 Listener 的 onClick 方法，然后调用 isDisposed 方法判断一下是不是已经取消订阅了，如果没有取消订阅，就通过 observer 的 onNext 方法发送事件，这时候，我们就会接收到这个事件，然后就可以对它进行处理，整个订阅过程就完成了。另外，当我们不需要再订阅 view 的点击事件的时候，调用 Disposable 的 dispose 方法，然后 onDispose 方法会被触发，这里设置 view 的 OnClickListener 为 null，再加上前面点击的时候有个 isDisposed 的判断，这样就保证了当订阅取消之后，再点击 view 的时候，就不会再接收到事件了。

RxBinding 中其他的方法的实现原理跟 clicks 的原理大致一样，就不再过多解释了。

RxToast

接下来我们仿照 RxBinding 的方法，写一个基于 RxJava 的 Toast 框架，就叫它 RxToast 吧。

相对于 RxBinding 来说，RxToast 更简单，因为我们在订阅之后直接发送事件就可以了，而在 RxBinding 中是首先订阅，然后每点击一次就发送一个事件。

首先写一个 RxToast 类

public class RxToast {

    private static RxToast rxToast;

    private Toast toast;

    private Context context;

    private RxToast(Context context) {
        this.context = context.getApplicationContext();
    }

    public static RxToast getInstance(Context context) {
        if (rxToast == null) {
            synchronized (RxToast.class) {
                if (rxToast == null) {
                    rxToast = new RxToast(context);
                }
            }
        }
        return rxToast;
    }

    private void initToast(CharSequence text) {
        if (toast == null) {
            toast = Toast.makeText(context, text, Toast.LENGTH_SHORT);
        } else {
            toast.setText(text);
        }
    }

    private void initToast(@StringRes int stringRes) {
        if (toast == null) {
            toast = Toast.makeText(context, stringRes, Toast.LENGTH_SHORT);
        } else {
            toast.setText(stringRes);
        }
    }

    public Observable<CharSequence> show(CharSequence text) {
        initToast(text);
        return new ToastObservable(toast, text);
    }

    public Observable<CharSequence> show(@StringRes int stringRes) {
        initToast(stringRes);
        return new ToastObservable(toast, context.getString(stringRes));
    }

}

这是一个单例的类，里面持有一个 toast 的引用，保证系统中只有一个 toast，在它的构造方法中，采用 ApplicationContext 是为了防止这个单例持有 Activity 的引用，导致内存泄漏，这个类中暴露给用户的是两个重载的 show 方法，分别对应 toast 的两个 makeText 方法，在这两个方法中，先对 toast 进行初始化，然后创建一个 ToastObservable 并返回。再来看一个这个 ToastObservable 类

final class ToastObservable extends Observable<CharSequence> {

    private Toast toast;

    private CharSequence text;

    ToastObservable(Toast toast, CharSequence text) {
        this.toast = toast;
        this.text = text;
    }

    @Override
    protected void subscribeActual(Observer<? super CharSequence> observer) {
        if (!checkMainThread(observer)) {
            return;
        }
        observer.onSubscribe(new Disposable(toast));
        observer.onNext(text);
        toast.show();
    }

    private boolean checkMainThread(Observer<? super CharSequence> observer) {
        if (Looper.myLooper() != Looper.getMainLooper()) {
            observer.onError(new IllegalStateException(
                    "Expected to be called on the main thread but was " + Thread.currentThread().getName()));
            return false;
        }
        return true;
    }

    private static final class Disposable extends MainThreadDisposable {

        private Toast toast;

        Disposable(Toast toast) {
            this.toast = toast;
        }

        @Override
        protected void onDispose() {
            toast.cancel();
        }
    }
}

跟前面的 ViewClickObservable 非常相似，不做过多的解释了，直接分析一遍流程，首先调用 RxToast 的 show 方法的时候会返回一个 ToastObservable 对象，然后调用它的 subscribe 方法，subscribeActual 方法被触发，检测一下当前的线程，如果是主线程，就调用 onSubscribe 通知我们已经订阅，然后直接发送一个事件，最后再弹出 toast，另外在它的内部类 Disposable 的 onDispose 方法中取消 toast，这样，当我们调用 dispose 方法取消订阅的时候，toast 就会取消显示。

综合应用

框架写好了，接下来试试好不好用，直接跟之前的 RxBinding 放在一起，当每点击一次 button 的时候，就弹出一个 toast，显示当前点击的次数

private void bindButton() {
    RxView.clicks(showButton).subscribe(new Consumer<Object>() {
        @Override
        public void accept(Object o) throws Exception {
            RxToast.getInstance(MainActivity.this).show("click " + (count++)).subscribe(new Consumer<CharSequence>() {
                @Override
                public void accept(CharSequence charSequence) throws Exception {

                }
            });
        }
    });
}

测试发现可以正常弹出每次点击 button 的时候都可以正常弹出 toast，并且显示当前点击的次数，但是代码的嵌套有点复杂，在 subscribe 的内部又创建了一个 Observable，那么怎样改进呢，这就是展示 RxJava 魅力的时候了，我们可以用一个 flatmap 来把弹出 toast 的事件放在外层的链上，代码如下

private void bindButton2() {
    RxView.clicks(showButton).flatMap(new Function<Object, ObservableSource<CharSequence>>() {
        @Override
        public ObservableSource<CharSequence> apply(Object o) throws Exception {
            return RxToast.getInstance(MainActivity.this).show("click " + (count++));
        }
    }).subscribe(new Consumer<CharSequence>() {
        @Override
        public void accept(CharSequence charSequence) throws Exception {

        }
    });
}

测试发现，点击的效果和刚才是一样的。

总结

其实写一个基于 RxJava 的框架还是很简单的，基本的思想就是暴露给用户一个返回 Observable 的方法，然后在合适的时候发送事件，当然，这只是最简单的情况，类似于 RxPermissions 这种有很多情况要考虑，会复杂很多。

代码已经上传到 github RxToast
也可以关注一下我的另个基于 RxJava 的高德地图定位的框架 RxLocation