本文是 Piasy 独立翻译，发表于 https://blog.piasy.com/AdvancedRxJava/，请阅读原文支持原创 https://blog.piasy.com/AdvancedRxJava/2017/04/04/the-new-completable-api-part-2/
原文 The new Completable API (part 2)。
介绍
在本文中，我将展示如何实现 Completable 的操作符（包括数据源和转换操作符）。由于 Completable API 并不会发出数据，只有 onError 和 onCompleted 事件，所以它的操作符比 Observable 少得多。
Empty
我们首先实现 empty() 操作符，它会在被订阅时立即发出 onCompleted 事件。我们利用 CompletableOnSubscribe 来实现这一操作符：
Completable empty = Completable.create(completableSubscriber -> {
BooleanSubscription cancel = new BooleanSubscription(); // (1)
completableSubscriber.onSubscribe(cancel); // (2)
if (!cancel.isUnsubscribed()) {
cancel.unsubscribe(); // (3)
completableSubscriber.onCompleted(); // (4)
}
});
这里我们无需引入类型参数。Completable 的 API 设计遵循了 Reactive-Stream 规范，它通过调用 onSubscribe 传递一个 Subscription 给下游，使得下游可以取消上游。
我们需要先创建一个 Subscription 实例，这里我们用了一个 BooleanSubscription 用于检查下游是否已经取消订阅（1）。在发出终止事件之前，我们需要先调用 onSubscribe，并把 BooleanSubscription 对象传递给下游（2）。这一步是必须的，如果漏掉这一步，可能大多数操作符都会抛出 NullPointerException，或者是在转换为 Observable 时取消订阅将失效。
和 Reactive-Stream 规范中定义的一样，终止事件发生后，我们要把 Subscription 等同于已被取消订阅，所以我们在调用 onCompleted 之前自行取消订阅（3）。
上面的代码看起来对一个简单如 empty 这样的操作符未免有些复杂，我们可以稍作简化：
Completable empty = Completable.create(completableSubscriber -> {
completableSubscriber.onSubscribe(Subscriptions.unsubscribed());
completableSubscriber.onCompleted();
});
这里我们直接给 onSubscribe 传递了一个已经取消订阅的 Subscription，然后立即调用 onCompleted。这么做有以下几个原因，由于取消订阅本身就只是一个尽力而为的保证，那就意味着：a) 我们必须做好准备，以应对取消订阅之后到来的事件；b) onSubscribe 和 onCompleted 之间的时间窗口真的很小，所以对于绝大部分异步场景，isUnsubscribed 都会返回 false；c) Subscription 在 onCompleted 调用之前也应该被认为是已经取消了。
Empty delayed
假设我们需要在延迟一段时间之后发出 onCompleted 事件。为此我们需要一个 Scheduler.Worker 来进行延迟执行，但我们也得保证延迟执行的代码可以被取消。
public static Completable emptyDelayed(
long delay, TimeUnit unit, Scheduler scheduler) {
return Completable.create(cs -> {
Scheduler.Worker w = scheduler.createWorker();
cs.onSubscribe(w);
w.schedule(() -> {
try {
cs.onCompleted();
} finally {
w.unsubscribe();
}
}, delay, unit);
});
}
幸运的是，Scheduler.Worker 实现了 Subscription 接口，所以我们可以在调度延迟任务之前，直接把它传给 CompletableSubscriber。在调度的任务中，我们在调用 onCompleted 之后再取消订阅 worker，显然在调用 onCompleted 时 worker 并没有被取消订阅。首先即便 Reactive-Stream 规范规定，org.reactivestreams.Subscription 在终止事件到来时应该被认为已经取消订阅了，但现在我们之所以可以这样做，是因为我们无法检查 Subscription 是否已经被取消订阅。虽然在 RxJava 的实现中，我们可以检查是否已经取消订阅，但检查上游是否认为我们已经被取消订阅，并没有什么意义。其次之所以 onCompleted 和 unsubscribe 的顺序很重要，是因为如果先取消了 worker，我们在调用 onCompleted 时可能在下游被意外中断。
如果我们确实想要确保下游收到的 Subscription 已经被取消订阅，我们就得用 MultipleAssignmentSubscription 实现得没这么直观：
Scheduler.Worker w = scheduler.createWorker();
MultipleAssignmentSubscription mas =
new MultipleAssignmentSubscription();
mas.set(w);
cs.onSubscribe(mas);
w.schedule(() -> {
mas.set(Subscriptions.unsubscribed());
mas.unsubscribe();
try {
cs.onCompleted();
} finally {
w.unsubscribe();
}
}, delay, unit);
我们不直接把 worker 传给下游，而是用 MultipleAssignmentSubscription 包装一层，并在调用 onCompleted 之前把它的内容替换为一个已经被取消订阅的 subscription，然后取消订阅整个容器类。使用 MultipleAssignmentSubscription 容器，以及替换 MultipleAssignmentSubscription 的内容，都是为了防止过早取消订阅 worker。SerialSubscription 和 CompositeSubscription 都无法达成目标。
最后，调度 onCompleted 的执行时需要格外小心，尤其是使用 RxJava 2.x 的 scheduler 时。RxJava 2.x 的 scheduler 允许“直接调度”，也就是说我们无需创建 worker 并在使用完毕之后取消订阅 worker。如果我们只需要调度一次任务，那这种形式可以减小开销，当然这种形式不保证同一个 scheduler 上调度的任务执行的先后顺序（这一点在这里可以接收）。所以 emptyDelayed 也可以这样实现：
cs.onSubscribe(
scheduler.scheduleDirect(cs::onCompleted, delay, unit));
但是上面的代码存在竞争：有可能被调度的任务，以及任务中要发出的 onCompleted 事件，都会在 scheduleDirect 返回之前被执行（例如 delay 是 0），那这时 subscriber 在收到 onCompleted 之前就没有收到 onSubscribe 了。更坏的情况是 onError 和 onCompleted 可能同时发生，这违反了串行访问的原则。解决办法也是引入一个中间层：
MultipleAssignmentDisposable mad =
new MultipleAssignmentDisposable();
cs.onSubscribe(mad);
mad.set(
scheduler.scheduleDirect(cs::onCompleted, delay, unit));
注：由于命名冲突，RxJava 2.x 的资源管理类叫 Disposable 而不是 Subscription，我会很快发布一个关于 2.x 的系列文章。
上面这个例子也预示了 Completable API 的一个特性：资源管理需要操作符自己负责，这里并没有像 rx.Subscriber 一样的 add(Subscription) 方法。这一点小小的不便需要我们在涉及到任务调度，或者涉及到多个资源管理时，要使用Subscription 容器类。这一设定的好处就是那些不需要资源管理的操作符，就不需要和 rx.Subscriber 一样相关的代码了，这样能节省内存分配，提高性能。
First completed
在我小学的时候，老师们经常发起一些小的挑战，同学们谁第一个完成就能得到一份小奖励。这和 amb() 很类似：第一个结束的数据源是赢家。当然，第一个发生错误的数据源也是赢家，在现实生活中失败显然不会让我们胜出。
那如何为 Completable 实现一个 amb 操作符？显然，不像 Observable.amb，我们不需要保存胜出 Completable 的信息（因为它已经终止了），所以我们不需要使用 trikery 的索引以及其他麻烦事，使用一个简单的 AtomicBoolean 就够了。
public static Completable amb(Completable... students) {
return Completable.create(principal -> {
AtomicBoolean done = new AtomicBoolean(); // (1)
CompositeSubscription all =
new CompositeSubscription(); // (2)
CompletableSubscriber teacher =
new CompletableSubscriber() {
@Override
public void onSubscribe(Subscription s) {
all.add(s); // (3)
}
@Override
public void onCompleted() {
if (done.compareAndSet(false, true)) { // (4)
all.unsubscribe();
principal.onCompleted();
}
}
@Override
public void onError(Throwable e) {
if (done.compareAndSet(false, true)) { // (5)
all.unsubscribe();
principal.onError(e);
}
}
};
principal.onSubscribe(all); // (6)
for (Completable student : students) {
if (done.get() || all.isUnsubscribed()) { // (7)
return;
}
student.subscribe(teacher); // (8)
}
});
}
在前面小学挑战的例子中，只会有一个老师检查所有学生的进展。这是一个针对 Completable 很有趣的优化（Reactive-Stream 里面也有），而且技术上它是切实可行的：teacher 这个 CompletableSubscriber 是无状态的，它只负责转发 onXXX 事件，而不需要记住是谁发出的这个事件。这里我强调技术上，因为 Reactive-Stream 规范禁止用同一个 Subscriber 订阅到多个 Publisher 上，但这一点在我看来是太过严格了：库的编写者，他们很清楚地知道自己在做什么，应该允许他们这样做。毫无意外的是，我们只需要把 teacher 的创建移到（8）处即可，其他的代码都不需要变化；显然，这也说明了 Subscriber 是可以被复用的。尽管如此，让我们看看上面值得关注的几个地方：
- 我们创建了一个共享的 done 变量，它会在任何一个学生通知老师时（onCompleted 或者 onError）被设置为 true；
- 如果校长不喜欢这个挑战，他可以随时取消整个挑战；
- 老师需要在学生到来时进行注册；
- 并且确保第一个学生发出终止事件时，也通知校长（不幸的是，校长无法知道是哪个学生胜出）。此时其他的学生也就没必要继续了，所以我们取消整个挑战；
- 当然，如果发生了错误，处理方式也是一样的；
- 我们允许校长直接取消整个挑战；
- 我们把每挑战的材料交给每个学生（让老师订阅学生）。当然有可能我们还在订阅过程中挑战就已经结束了，那我们也就无需继续订阅了。
When all completed
继续我们上小学的例子，有时候学生的表现需要被集体评估，整个活动需要等到所有学生都完成评估后才能结束。如果发生了意外，我们可能需要终止整个活动，并且把受伤的学生一起送到救护车。
我希望这样的设定听起来很熟悉，如果不熟悉，它们其实分别对应着 merge 和 mergeDelayError 操作符，具体是哪个取决于学校的政策。
让我们先看看最简单的情况，我们知道学生的数量，而且任何错误都会终止整个活动：
public static Completable merge(Completable... students) {
return Completable.create(principal -> {
AtomicInteger remaining =
new AtomicInteger(students.length); // (1)
CompositeSubscription all =
new CompositeSubscription();
CompletableSubscriber evaluator =
new CompletableSubscriber() {
@Override
public void onSubscribe(Subscription s) {
all.add(s);
}
@Override
public void onCompleted() {
if (remaining.decrementAndGet() == 0) { // (2)
all.unsubscribe();
principal.onCompleted();
}
}
@Override
public void onError(Throwable e) {
if (remaining.getAndSet(0) > 0) { // (3)
all.unsubscribe();
principal.onError(e);
}
}
};
principal.onSubscribe(all);
for (Completable student : students) {
if (all.isUnsubscribed()
|| remaining.get() <= 0) { // (4)
return;
}
student.subscribe(evaluator);
}
});
}
上面的实现看起来和 amb() 比较类似，但也有以下几点不同：
- 我们需要原子地计数（递减）成功完成评估的学生数量；
- 一旦计数递减到 0，我们就通知校长评估完成；
- 如果有学生发出了错误，那我们就原子地把计数置为 0，如果此前计数不为 0，我们就取消掉其他所有学生，并向校长通报错误。注意错误的通报最多只会发生一次，因为如果有多个 onError 并发到来，只会有一个成功地把计数置为 0。任何后续的结束事件都会继续递减计数器；
- 如果发生了错误，或者活动被取消，我们就不再继续订阅了，以避免不必要的开销；
但如果我们不知道准确的学生数量，而且发生了错误之后，我们不希望结束整个活动，应该怎么做？这让我们的错误管理以及对学生的追踪变得复杂了一些：
public static Completable mergeDelayError(
Iterable<? extends Completable> students) {
return Completable.create(principal -> {
AtomicInteger wip = new AtomicInteger(1); // (1)
CompositeSubscription all =
new CompositeSubscription();
Queue<Throwable> errors =
new ConcurrentLinkedQueue<>(); // (2)
CompletableSubscriber evaluator =
new CompletableSubscriber() {
@Override
public void onSubscribe(Subscription s) {
all.add(s);
}
@Override
public void onCompleted() {
if (wip.decrementAndGet() == 0) { // (3)
if (errors.isEmpty()) {
principal.onCompleted();
} else {
principal.onError(
new CompositeException(errors));
}
}
}
@Override
public void onError(Throwable e) {
errors.offer(e); // (4)
onCompleted();
}
};
principal.onSubscribe(all);
for (Completable student : students) {
if (all.isUnsubscribed()) { // (5)
return;
}
wip.getAndIncrement(); // (6)
student.subscribe(evaluator);
}
evaluator.onCompleted(); // (7)
});
}
同样，结构看起来很类似，但不同的要求需要用不同的算法来实现：
- 首先我们的 wip 计数器初始值是 1。我们不知道会有多少个学生，但我们知道当 wip 变成 0 的时候，整个活动就结束了；
- 我们会把所有的异常都保存到一个并发的队列中；
- 终止条件在 evaluator 的 onCompleted 中检查：如果 wip 变成了 0，我们就检查异常队列是否为空，如果非空我们就把它们一起作为 CompositeException 发给校长；否则我们就告知校长活动成功结束；
- 由于异常并不会提前终止活动，所以发送异常的处理和正常结束一样，但在此之前我们需要把异常加入到队列中（注意，行为不当的 Completable 会导致出错，如果它发出了多个终止事件，就会导致提前终止）；
- 在订阅的循环中，我们只能检查校长是否需要取消活动，wip 的值在这里无法使用，因为在循环过程中，它至少是 1（因为初始值是 1）；
- 对每一个学生，我们先递增 wip，再订阅它；这保证了我们在循环过程中，wip 至少是 1；
- 最后我们手动调用一次
evaluator.onCompleted()，表示没有更多的 Completable 了，这使得 wip 可以被递减到 0 进而终止整个活动；
在 RxJava 1.x 的 Observable 体系中，很难出现代码如此紧凑，复用程度如此高的场景（甚至不可能出现）。
Transformative operators
我不认为能有多少 Completable 的“序列”是可以被变换的。大多数 Observable 的操作符都没有意义，所以在 Completable 的 API 中省略了。但还是有几个例子可以看看的。
首先我们可能想要忽略错误事件，由于这里不涉及到 onNext，所以错误发生时我们最好的选择就是自行调用 onCompleted：
CompletableOperator onErrorComplete = cs -> {
return new CompletableSubscriber() {
@Override
public void onSubscribe(Subscription s) {
cs.onSubscribe(s);
}
@Override
public void onCompleted() {
cs.onCompleted();
}
@Override
public void onError(Throwable e) {
cs.onCompleted();
}
};
};
source.lift(onErrorComplete).subscribe();
当然，我们也可以在错误发生时订阅到另一个 Completable：
public static Completable onErrorResumeNext(
Completable first,
Func1<Throwable, ? extends Completable> otherFactory) {
return first.lift(cs -> new CompletableSubscriber() {
final SerialSubscription serial =
new SerialSubscription(); // (1)
boolean once;
@Override
public void onSubscribe(Subscription s) {
serial.set(s); // (2)
}
@Override
public void onCompleted() {
cs.onCompleted();
}
@Override
public void onError(Throwable e) {
if (!once) { // (3)
once = true;
otherFactory.call(e).subscribe(this); // (4)
} else {
cs.onError(e); // (5)
}
}
});
}
在上面的例子中，我们把 CompletableOperator lift 到传入的 Completable 上。在操作符主体中，我们返回一个具有以下行为的 CompletableSubscriber：
- 由于我们可能会切换上游，所以我们需要把第一个 Completable 传入的 Subscription 更换为其他 Completable 传入的 Subscription。这里我们也可以使用 MultipleAssignmentSubscription。这和 Observable 操作符中常见的 ProducerArbiter 机制差不多，尽管简单了不少。此外，我们会在后来的 Completable 中复用这个 serial 对象，当然，如果它也出错了，我们就不会再尝试了；
- 我们把 Subscription 设置到 SerialSubscription 中，把老的 Subscription 淘汰掉；
- 如果 first 发生了错误，我们需要确保订阅到新的 Completable 只会发生一次；
- 这里我们复用了 CompletableSubscriber 对象，节省了一次内存分配。（2）处的代码确保了取消链条被正确维护。这里为了简洁，我省略了工厂方法调用周围的 try-catch；如果发生了异常，我们可以把它和 onError 的异常合并为一个 CompositeException，并调用
cs.onError； - 如果第二个 Completable 也发生了异常，我们就把错误发往下游；
考虑一下，如果我们需要在 onCompleted 发生后执行类似的逻辑（例如 andThen，endWith，concatWith），我们可以用类似的方式实现。只不过我们不是在 onError 中做事，而是在 onCompleted 中做事：
// ...
@Override
public void onCompleted() {
if (!once) { // (3)
once = true;
otherFactory.call(e).subscribe(this); // (4)
} else {
cs.onCompleted(); // (5)
}
}
@Override
public void onError(Throwable e) {
cs.onError(e);
}
最后，我们可能需要在第一个 Completable 超时后，切换到另一个上：
public static Completable timeout(
Completable first,
long timeout, TimeUnit unit, Scheduler scheduler
Completable other) {
return first.lift(cs -> new CompletableSubscriber() {
final CompositeSubscription csub =
new CompositeSubscription(); // (1)
final AtomicBoolean once = new AtomicBoolean(); // (2)
@Override
public void onSubscribe(Subscription s) {
csub.add(s); // (3)
Scheduler.Worker w = scheduler.createWorker();
csub.add(w);
cs.onSubscribe(csub);
w.schedule(this::onTimeout, timeout, unit); // (4)
}
@Override
public void onCompleted() {
if (once.compareAndSet(false, true)) {
csub.unsubscribe(); // (5)
cs.onCompleted();
}
}
@Override
public void onError(Throwable e) {
if (once.compareAndSet(false, true)) {
csub.unsubscribe();
cs.onError(e);
}
}
void onTimeout() {
if (once.compareAndSet(false, true)) { // (6)
csub.clear();
other.subscribe(new CompletableSubscriber() {
@Override
public void onSubscribe(Subscription s) {
csub.add(s);
}
@Override
public void onCompleted() {
cs.onCompleted();
}
@Override
public void onError(Throwable e) {
cs.onError(e);
}
});
}
}
});
}
这个操作符就更复杂一些了：
- 我们需要追踪前后两个 Completable 的 Subscription，以及用于触发超时的 worker；
- 由于第一个 Completable 的终止事件和超时事件存在竞争，我们需要用一个原子变量来控制只有其中之一胜出（就像
amb()那样）； - 当第一个上游的 Subscription 到来时，我们创建一个容器，并把 Subscription 加入到容器中，然后创建一个 worker，再把容器传递给下游；
- 最后，我们调度一次 onTimeout 的执行。这一设定让我们避免了 onTimeout 和 onSubscribe 之间的竞争，例如，如果 schedule 发生在 timeout() 被调用时，那超时就有可能在 Subscription 到来之前发生，那我们就需要做一些额外的事情以确保不出问题（这里不详细展开）。大多数时候，在 Reactive-Stream 兼容的 RxJava 2.x 中使用这种风格的 onSubscribe 实现，能够减少很多让我们头疼的事情；
- 如果第一个 Completable 的终止事件先来，我们就原子地、有条件地设置标记变量（这能阻止 onTimeout 执行其内部逻辑）。由于容器也管理着 worker，所以我们在取消了容器之后，就把终止事件发往下游；
- 如果 onTimeout 先发生，并且成功设置了标记变量，那我们就把容器清空掉，然后用一个全新的 CompletableSubscriber 订阅到第二个 Completable 中。这里我们之所以先清空容器，是为了容纳第二个 Completable 的 Subscription，而如果不清空，一个已经被取消的容器就不能使用了。清空容器的第二个好处是，它也会取消第一个 Completable，以及进行调度的 worker；
Hot Completable?
Hot Completable 仅仅是这里的最后一点思考，我并不确定它（或者一个 published Completable）在此之外是否有实际使用场景，但我们在这里还是看看如何实现它。由于 Completable 并不会发出数据，所以只需要实现一种 CompletableSubject 即可，它不会重放任何数据，只重放终止事件。
首先看看 CompletableSubject 的结构：
public final class CompletableSubject
extends Completable implements CompletableSubscriber {
public static CompletableSubject create() {
State state = new State();
return new CompletableSubject(state);
}
static final class State
implements CompletableOnSubscribe, CompletableSubscriber {
// TODO state fields
boolean add(CompletableSubscriber t) {
// TODO implement
}
void remove(CompletableSubscriber t) {
// TODO implement
}
@Override
public void call(CompletableSubscriber t) {
// TODO implement
}
@Override
public void onSubscribe(Subscription d) {
// TODO implement
}
@Override
public void onCompleted() {
// TODO implement
}
@Override
public void onError(Throwable e) {
// TODO implement
}
}
static final class CompletableSubscription
extends AtomicBoolean implements Subscription {
/** */
private static final long serialVersionUID =
-3940816402954220866L;
final CompletableSubscriber actual;
final State state;
public CompletableSubscription(
CompletableSubscriber actual, State state) {
this.actual = actual;
this.state = state;
}
@Override
public boolean isUnsubscribed() {
return get();
}
@Override
public void unsubscribe() {
if (compareAndSet(false, true)) {
state.remove(this);
}
}
}
final State state;
private CompletableSubject(State state) {
super(state);
this.state = state;
}
@Override
public void onSubscribe(Subscription d) {
state.onSubscribe(d);
}
@Override
public void onCompleted() {
state.onCompleted();
}
@Override
public void onError(Throwable e) {
state.onError(e);
}
}
它的结构和以前的 Subject 别无二致。我们必须通过工厂方法创建 CompletableSubject，并且需要 CompletableSubscriber 的方法把事件委托到共享的 State 对象上。CompletableSubscription 用于追踪每个 CompletableSubscriber，并根据它们的标识管理取消订阅。
State 类需要记录一个终止标记，以及一个可能发生的 Throwable，加上当前的下游 CompletableSubscriber 数组：
Throwable error;
volatile CompletableSubscription[] subscribers = EMPTY;
static final CompletableSubscription[] EMPTY =
new CompletableSubscription[0];
static final CompletableSubscription[] TERMINATED =
new CompletableSubscription[0];
boolean add(CompletableSubscription t) {
if (subscribers == TERMINATED) {
return false;
}
synchronized (this) {
CompletableSubscription[] a = subscribers;
if (a == TERMINATED) {
return false;
}
CompletableSubscription[] b =
new CompletableSubscription[a.length + 1];
System.arraycopy(a, 0, b, 0, a.length);
b[a.length] = t;
subscribers = b;
return true;
}
}
void remove(CompletableSubscription t) {
CompletableSubscription[] a = subscribers;
if (a == EMPTY || a == TERMINATED) {
return;
}
synchronized (this) {
a = subscribers;
if (a == EMPTY || a == TERMINATED) {
return;
}
int j = -1;
for (int i = 0; i < a.length; i++) {
if (a[i] == t) {
j = i;
break;
}
}
if (j < 0) {
return;
}
if (a.length == 1) {
subscribers = EMPTY;
return;
}
CompletableSubscription[] b =
new CompletableSubscription[a.length - 1];
System.arraycopy(a, 0, b, 0, j);
System.arraycopy(a, j + 1, b, j, a.length - j - 1);
subscribers = b;
}
}
add 和 remove 的实现与其他 Subject 的实现一模一样，主要用于维护当前的下游 CompletableSubscriber 数组。
接下来让我们处理新来的 Subscriber：
@Override
public void call(CompletableSubscriber t) {
CompletableSubscription cs =
new CompletableSubscription(t, this);
t.onSubscribe(cs);
if (add(cs)) {
if (cs.isUnsubscribed()) {
remove(cs);
}
} else {
Throwable e = error;
if (e != null) {
t.onError(e);
} else {
t.onCompleted();
}
}
}
我们为每个 CompletableSubscriber 创建一个 CompletableSubscription，用来保存 subscriber 以及当前的状态：如果 child 调用了 unsubscribe，那它就可以把自己从 state 的 subscriber 数组中移除了。注意，这里存在 add 和取消订阅的竞争，这会导致 CompletableSubscriber 被意外保留。所以我们在 add 之后检查是否取消订阅，如果已经取消，我们就再调用 remove。如果 add 返回了 false，这说明 CompletableSubject 已经处于终止状态，读取 error 成员我们可以判断是否发生了异常，并确定向下游发送何种事件。
处理 onXXX 也不复杂：
@Override
public void onSubscribe(Subscription d) {
if (subscribers == TERMINATED) {
d.unsubscribe();
}
}
@Override
public void onCompleted() {
CompletableSubscription[] a;
synchronized (this) {
a = subscribers;
subscribers = TERMINATED;
}
for (CompletableSubscription cs : a) {
cs.actual.onCompleted();
}
}
@Override
public void onError(Throwable e) {
CompletableSubscription[] a;
synchronized (this) {
a = subscribers;
error = e;
subscribers = TERMINATED;
}
for (CompletableSubscription cs : a) {
cs.actual.onError(e);
}
}
onSubscribe 的行为倒是值得商榷，这里我的做法是，如果 CompletableSubject 已经处于终止状态，就将其取消订阅。即便不这样，我们也不能做太多事，因为 CompletableSubject 可能订阅到很多 Completable，任何一个都可能让其进入到终止状态。我们也可以忽略传入的 Subscription，或者把它们都保存到一个容器中。
onError 和 onCompleted 的处理基本一致，我们原子地把 subscriber 数组设置为终止标记，然后向此前的每个 subscriber 发出相应的终止事件。注意，在 onError 中，我们先设置 error 成员，再设置 subscribers 成员，这让我们可以在前面的 call() 函数中得到恰当的可见性。
总结
在本文中，我详细展示了如何实现 Completable 操作符，既包括数据源操作符，也包括转换型操作符，甚至还实现了一个 Subject。
Completable 更像是一个类型工具，它不会发出任何事件，只是表示有“副作用”的代码执行是否成功。简化了的 API 用起来可能比 Observable 更加方便。
实现 Completable 操作符比实现支持 backpressure 的 Observable 操作符更简单，但我们还是需要考虑取消订阅链条、避免 onXXX 的竞争，以及利用好 AtomicXXX 类型以更高效地实现状态管理。
既然我们已经有了更多编写 Subject 的经验，下一个系列让我们看看 ConnectableObservable。
