layout: post title: Reactive-Streams API（二）：SingleSubscription，SingleDelayedSubscription 和 RangeSubscription tags:

  1. - Reactive Stream

原文 The Reactive-Streams API (part 2){:target=”_blank”}

介绍

在本文中，我将把我们以前的 SingleProducer 和 SingleDelayedProducer 移植到基于 reactive-streams 的 Subscription。

首先，很多人可能认为这个转换过程很麻烦，但幸运的是，如果我们已经想清楚了在 rx.Producer 中如何实现 request()，那我们基本上就已经完成了 75% 了。剩下的 25% 来自于我们要把 rx.Subscriber.isUnsubscribed() 中的逻辑转移到 request() 中，因为 org.reactivestreams.Subscriber 中没有 isUnsubscribed()（其他资源管理类的接口都没有这个方法了）。

SingleSubscription

由于 SingleSubscription 本身并不复杂，这里我就直接一步到位：

  1. import org.reactivestreams.*;
  3. public final class SingleSubscription<T> 
  4. extends AtomicBoolean implements Subscription {
  5.     private static final long serialVersionUID = 1L;
  7.     final T value;                                       // (1)
  8.     final Subscriber<? super T> child;
  9.     volatile boolean cancelled;                          // (2)
  11.     public SingleSubscription(T value, 
  12.             Subscriber<? super T> child) {               // (3)
  13.         this.value = Objects.requireNonNull(value);
  14.         this.child = Objects.requireNonNull(child);
  15.     }
  16.     @Override
  17.     public void request(long n) {
  18.         if (n <= 0) {
  19.             throw new IllegalArgumentException(
  20.                 "n > 0 required");                       // (4)
  21.         }
  22.         if (compareAndSet(false, true)) {
  23.             if (!cancelled) {                            // (5)
  24.                 child.onNext(value);
  25.                 if (!cancelled) {
  26.                     child.onComplete();
  27.                 }
  28.             }
  29.         }
  30.     }
  31.     @Override
  32.     public void cancel() {
  33.         cancelled = true;                                // (6)
  34.     }
  35. }

就是这样！这里我向大家展示了把 Producer 的实现迁移到 reactive-streams Subscription 并不需要太多的工作。但这里还是有几点值得一提：

  1. 我们需要在成员变量中保存将要发出的值，以及下游 subscriber。
  2. 然而由于 RS 中没有了 isUnsubscribed()，而且取消订阅也变成了 cancel()，所以我们需要在一个 volatile 变量中记录是否已经取消订阅。如果你还记得的话，我说过我们无法预知 request() 和 cancel() 的调用情况，所以我们需要保证它们的线程安全性。
  3. 由于 RS 不允许 null，我们在构造函数中就检查错误。
  4. 我的“Let them throw!”哲学告诉我们非正请求数量是编程错误，我们应该抛出 IllegalArgumentException。
  5. 由于没有 child.isUnsubscribed() 函数了，我们只能检查我们的 cancelled 变量。
  6. 为了保证取消的幂等性，我们只是安全的更改 cancelled 变量。

SingleDelayedSubscription

SingleSubscription 都这么简单了，SingleDelayedSubscription 又能复杂到哪里去呢？

  1. public final class SingleDelayedSubscription<T> 
  2. extends AtomicInteger implements Subscription {
  3.     private static final long serialVersionUID = -1L;
  5.     T value;
  6.     final Subscriber<? super T> child;
  8.     static final int CANCELLED = -1;                           // (1)
  9.     static final int NO_VALUE_NO_REQUEST = 0;
  10.     static final int NO_VALUE_HAS_REQUEST = 1;
  11.     static final int HAS_VALUE_NO_REQUEST = 2;
  12.     static final int HAS_VALUE_HAS_REQUEST = 3;
  14.     public SingleDelayedSubscription(Subscriber<? super T> child) {
  15.         this.child = Objects.requireNonNull(child);
  16.     }
  17.     @Override
  18.     public void request(long n) {
  19.         if (n <= 0) {
  20.             throw new IllegalArgumentException("n > 0 required");
  21.         }
  22.         for (;;) {
  23.             int s = get();
  24.             if (s == NO_VALUE_HAS_REQUEST
  25.                     || s == HAS_VALUE_HAS_REQUEST
  26.                     || s == CANCELLED) {                       // (2)
  27.                 return;
  28.             } else if (s == NO_VALUE_NO_REQUEST) {
  29.                 if (!compareAndSet(s, NO_VALUE_HAS_REQUEST)) {
  30.                     continue;
  31.                 }
  32.             } else if (s == HAS_VALUE_NO_REQUEST) {
  33.                 if (compareAndSet(s, HAS_VALUE_HAS_REQUEST)) {
  34.                     T v = value;
  35.                     value = null;
  36.                     child.onNext(v);
  37.                     if (get() != CANCELLED) {                  // (3)
  38.                         child.onComplete();
  39.                     }
  40.                 }
  41.             }
  42.             return;
  43.         }
  44.     }
  46.     public void setValue(T value) {
  47.        Objects.requireNonNull(value);
  48.        for (;;) {
  49.            int s = get();
  50.            if (s == HAS_VALUE_NO_REQUEST
  51.                    || s == HAS_VALUE_HAS_REQUEST
  52.                    || s == CANCELLED) {                        // (4)
  53.                return;
  54.            } else if (s == NO_VALUE_NO_REQUEST) {
  55.                this.value = value;
  56.                if (!compareAndSet(s, HAS_VALUE_NO_REQUEST)) {
  57.                    continue;
  58.                }
  59.            } else if (s == NO_VALUE_HAS_REQUEST) {
  60.                if (compareAndSet(s, HAS_VALUE_HAS_REQUEST)) {
  61.                    child.onNext(value);
  62.                    if (get() != CANCELLED) {                   // (5)
  63.                        child.onComplete();
  64.                    }
  65.                }
  66.            }
  67.            return;
  68.        }
  69.     }
  71.     @Override
  72.     public void cancel() {
  73.         int state = get();
  74.         if (state != CANCELLED) {                              // (6)
  75.             state = getAndSet(CANCELLED);
  76.             if (state != CANCELLED) {
  77.                 value = null;
  78.             }
  79.         }
  80.     }
  81. }

看起来和原来的状态机非常类似，但是多了一个 CANCELLED 状态，我们无需在 onNext 之前检查状态不为 CANCELLED，因为 CAS 操作隐含了这一条件，但我们应该在 onComplete() 之前进行检查。

为什么我们不使用一个 volatile boolean 来记录是否取消呢？其实完全可以。这种选择仅仅是出于个人偏好：增加一个成员变量，或者是扩展一个新状态。我主要是想要展示一下后者怎么实现。

RangeSubscription

我并不打算在这里把以前所有的 Producer 都改写为 Subscription，但我这里还想展示一个包括取消状态的状态机例子：

  1. public final class RangeSubscription 
  2. extends AtomicLong implements Subscription {
  3.     private static final long serialVersionUID = 1L;
  5.     final Subscriber<? super Integer> child;
  6.     int index;
  7.     final int max;
  9.     static final long CANCELLED = Long.MIN_VALUE;          // (1)
  11.     public RangeSubscription(
  12.             Subscriber<? super Integer> child, 
  13.             int start, int count) {
  14.         this.child = Objects.requireNonNull(child);
  15.         this.index = start;
  16.         this.max = start + count;
  17.     }
  18.     @Override
  19.     public void request(long n) {
  20.         if (n <= 0) {
  21.             throw new IllegalArgumentException(
  22.                 "n > required");
  23.         }
  24.         long r;
  25.         for (;;) {
  26.             r = get();
  27.             if (r == CANCELLED) {                          // (2)
  28.                 return;
  29.             }
  30.             long u = r + n;
  31.             if (u < 0) {
  32.                 u = Long.MAX_VALUE;
  33.             }
  34.             if (compareAndSet(r, u)) {
  35.                 break;
  36.             }
  37.         }
  38.         if (r != 0L) {                                     // (p1)
  39.             return;
  40.         }
  41.         for (;;) {
  42.             r = get();
  43.             if (r == CANCELLED) {                          // (3)
  44.                  return;
  45.             }
  46.             int i = index;
  47.             int m = max;
  48.             long e = 0;
  49.             while (r > 0L && i < m) {                      // (p2)
  50.                 child.onNext(i);
  51.                 if (get() == CANCELLED) {                  // (4)
  52.                     return;
  53.                 }
  54.                 i++;
  55.                 if (i == m) {
  56.                     child.onComplete();
  57.                     return;
  58.                 }
  59.                 r--;
  60.                 e++;
  61.             }
  62.             index = i;
  63.             if (e != 0) {
  64.                 for (;;) {
  65.                     r = get();
  66.                     if (r == CANCELLED) {                  // (5)
  67.                         return;
  68.                     }
  69.                     long u = r - e;
  70.                     if (u < 0) {
  71.                         throw new IllegalStateException(
  72.                                 "more produced than requested!");
  73.                     }
  74.                     if (compareAndSet(r, u)) {
  75.                         break;
  76.                     }
  77.                 }
  78.             }
  79.             if (r <= 0L) {                                 // (p3)
  80.                 break;
  81.             }
  82.         }
  83.     }
  84.     @Override
  85.     public void cancel() {
  86.         if (get() != CANCELLED) {                          // (6)
  87.             getAndSet(CANCELLED);
  88.         }
  89.     }
  90. }

为了简洁起见，我省略了快速路径的逻辑。剩下的部分和原来的 RangeProducer 类似，但是取消状态被合并进了计数状态中，我们几乎需要在所有的地方（1~5）重新读出计数并和 CANCELLED 对比。注意发射计数再也不能用 getAndAdd() 了，直接递增可能会覆盖掉 CANCELLED 状态。最后在取消时使用 getAndSet() 可以保证幂等性。

译者注：这一段代码还是很复杂的，即便我之前翻译过 RangeProducer 的实现，看起来依然需要一些思考，所以这里进行一些分析（对应于上面的 p 标号）：

  1. 当我们成功更新请求计数之后，只有从 0 开始增加请求计数的线程可以进入后面的发射循环。发射过程会递减请求计数，当请求处理完毕之后（请求计数重新变为 0），下次的请求调用才有可能进入发射循环。
  2. 这里发射循环有两个条件，一是有未处理的请求（r > 0），而是发射数据没有超出范围（i < m），这两者很可能是不同的。此外，如果发射数据已经超出了范围，而且请求计数也递减为 0 了，那后续的请求仍然能通过（p1）的检查，但不会进入发射循环，因为通不过 i < m 的检查。
  3. 这里是请求处理完毕，但发射数据没有超出范围的退出路径。

总结

在本文中，我展示了两种把 rx.Producer 改写为 RS Subscription 的方式（boolean 成员或者新状态），它们都能保证取消行为的正确性。对它们进行取舍需要进行权衡：boolean 成员会增加对象大小，新状态会增加算法复杂性。

下一篇文章中，我将介绍如何处理 RS 中缺失的另一个 rx.Subscriber 特性：用 add(rx.Subscriber) 把资源和下游 subscriber 结合起来。