RxJava 响应式编程 | RxJava 的实际应用示范

在这篇文章中，我们会先分析一下 RxJava2 中的 Subject；然后，我们会使用 Subject 制作一个类似于 EventBus 的全局的通信工具。在了解本篇文章的内容之前，你需要先了解 RxJava2 中的一些基本的用法，比如 Observable 以及背压的概念。

1、Subject

Subject 可以同时代表 Observer 和 Observable，允许从数据源中多次发送结果给多个观察者。除了 onSubscribe()、onNext()、onError() 和 onComplete() 之外，所有的方法都是线程安全的。此外，你还可以使用 toSerialized() 方法，也就是转换成串行的，将这些方法设置成线程安全的。

如果你已经了解了 Observable 和 Observer ，那么也许直接看 Subject 的源码定义会更容易理解：

public abstract class Subject extends Observable implements Observer {

    // ...
}
复制代码

从上面看出，Subject 同时继承了 Observable 和 Observer 两个接口，说明它既是被观察的对象，同时又是观察对象，也就是可以生产、可以消费、也可以自己生产自己消费。所以，我们可以项下面这样来使用它。这里我们用到的是该接口的一个实现 PublishSubject ：

    public static void main(String...args) {
        PublishSubject subject = PublishSubject.create();
        subject.subscribe(System.out::println);

        Executor executor = Executors.newFixedThreadPool(5);
        Disposable disposable = Observable.range(1, 5).subscribe(i ->
                executor.execute(() -> {
                    try {
                        Thread.sleep(i * 200);
                        subject.onNext(i);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }));
    }
复制代码

根据程序的执行结果，程序在第 200, 400, 600, 800, 1000 毫秒依次输出了 1 到 5 的数字。

在这里，我们用 PublishSubject 创建了一个主题并对其监听，然后在线程当中又通知该主题内容变化，整个过程我们都只操作了 PublishSubject 一个对象。显然，使用 Subject 我们可以达到对一个指定类型的值的结果进行监听的目的——我们把值改变之后对应的逻辑写在 subscribe() 方法中，然后每次调用 onNext() 等方法通知结果之后就可以自动调用 subscribe() 方法进行更新操作。

同时，因为 Subject 实现了 Observer 接口，并且在 Observable 等的 subscribe() 方法中存在一个以 Observer 作为参数的方法（如下），所以，Subject 也是可以作为消费者来对事件进行消费的。

public final void subscribe(Observer observer) 
复制代码

以上就是 Subject 的两个主要的特性。

在 RxJava2 ，Subject 有几个默认的实现，下面我们对它们之间的区别做简单的说明：

1. AsyncSubject: 只有当 Subject 调用 onComplete() 方法时，才会将 Subject 中的最后一个事件传递给所有的 Observer。
2. BehaviorSubject: 该类有创建时需要一个默认参数，该默认参数会在 Subject 未发送过其他的事件时，向注册的 Observer 发送；新注册的 Observer 不会收到之前发送的事件，这点和 PublishSubject 一致。
3. PublishSubject: 不会改变事件的发送顺序；在已经发送了一部分事件之后注册的 Observer 不会收到之前发送的事件。
4. ReplaySubject: 无论什么时候注册 Observer 都可以接收到任何时候通过该 Observable 发射的事件。
5. UnicastSubject: 只允许一个 Observer 进行监听，在该 Observer 注册之前会将发射的所有的事件放进一个队列中，并在 Observer 注册的时候一起通知给它。

对比 PublishSubject 和 ReplaySubject，它们的区别在于新注册的 Observer 是否能够收到在它注册之前发送的事件。这个类似于 EventBus 中的 StickyEvent 即黏性事件，为了说明这一点，我们准备了下面两段代码：

    private static void testPublishSubject() throws InterruptedException {
        PublishSubject subject = PublishSubject.create();
        subject.subscribe(i -> System.out.print("(1: " + i + ") "));

        Executor executor = Executors.newFixedThreadPool(5);
        Disposable disposable = Observable.range(1, 5).subscribe(i -> executor.execute(() -> {
            try {
                Thread.sleep(i * 200);
                subject.onNext(i);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }));

        Thread.sleep(500);
        subject.subscribe(i -> System.out.print("(2: " + i + ") "));

        Observable.timer(2, TimeUnit.SECONDS).subscribe(i -> ((ExecutorService) executor).shutdown());
    }

    private static void testReplaySubject() throws InterruptedException {
        ReplaySubject subject = ReplaySubject.create();
        subject.subscribe(i -> System.out.print("(1: " + i + ") "));

        Executor executor = Executors.newFixedThreadPool(5);
        Disposable disposable = Observable.range(1, 5).subscribe(i -> executor.execute(() -> {
            try {
                Thread.sleep(i * 200);
                subject.onNext(i);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }));

        Thread.sleep(500);
        subject.subscribe(i -> System.out.print("(2: " + i + ") "));

        Observable.timer(2, TimeUnit.SECONDS).subscribe(i -> ((ExecutorService) executor).shutdown());
    }
复制代码

它们的输出结果依次是

• PublishSubject的结果：(1: 1) (1: 2) (1: 3) (2: 3) (1: 4) (2: 4) (1: 5) (2: 5)
• ReplaySubject的结果：(1: 1) (1: 2) (2: 1) (2: 2) (1: 3) (2: 3) (1: 4) (2: 4) (1: 5) (2: 5)

从上面的结果对比中，我们可以看出前者与后者的区别在于新注册的 Observer 并没有收到在它注册之前发送的事件。试验的结果与上面的叙述是一致的。

其他的测试代码这不一并给出了，详细的代码可以参考 Github - Java Advanced。

清楚了 Subject 的概念之后，让我们来做一个实践——用 RxJava 打造 EventBus。

我们先考虑用一个全局的 PublishSubject 来解决这个问题，当然，这意味着我们发送的事件不是黏性事件。不过，没关系，只要这种实现方式搞懂了，用 ReplaySubject 做一个发送黏性事件的 EventBus 也非难事。

考虑一下，如果要实现这个功能我们需要做哪些准备：

1. **我们需要发送事件并能够正确地接收到事件。**要实现这个目的并不难，因为 Subject 本身就具有发送和接收两个能力，作为全局的之后就具有了全局的注册和通知的能力。因此，不论你在什么位置发送了事件，任何订阅的地方都能收到该事件。
2. **首先，我们要在合适的位置对事件进行监听，并在合适的位置取消事件的监听。如果我们没有在适当的时机释放事件，会不会造成内存泄漏呢？这还是有可能的。**所以，我们需要对注册监听的观察者进行记录，并提供注册和取消注册的方法，给它们在指定的生命周期中进行调用。

好了，首先是全局的 Subject 的问题，我们可以实现一个静态的或者单例的 Subject。这里我们选择使用后者，所以，我们需要一个单例的方式来使用 Subject：

public class RxBus {

    private static volatile RxBus rxBus;

    private final Subject