前言

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

1、Subject

1.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 作为参数的方法（如下）&#x

参考链接

评论可见，请评论后查看内容，谢谢！！！
 您阅读本篇文章共花了：