webflux之reactor-Subscriber

在进行第二个demo之前。先来讲解一下reactive-streams官网中有关Subscriber的一些规则。这里将基于第一篇reactor博客中的demo进行讲解。demo1。

public interface Subscriber<T> {public void onSubscribe(Subscription s);public void onNext(T t);public void onError(Throwable t);public void onComplete();
}

Subscriber规则

订阅者必须通过Subscription.request(long n)发信号通知需要接收onNext信号。
此规则的目的是确定订阅者有责任决定它能够和愿意接收的元素的时间和数量。为了避免由重入的订阅方法引起的信号重新排序，强烈建议同步订阅服务器实现在任何信号处理的最后调用订阅方法。建议订阅者请求他们能够处理的上限，因为一次只请求一个元素会导致本身效率低下的“停止等待”协议。
代码1 LambdaSubscriber类

@Override
public final void onSubscribe(Subscription s) {//这里是对重入做了验证，当前订阅不为空则取消新订阅并返回falseif (Operators.validate(subscription, s)) {this.subscription = s;if (subscriptionConsumer != null) {try {subscriptionConsumer.accept(s);}catch (Throwable t) {Exceptions.throwIfFatal(t);s.cancel();onError(t);}}else {//这里进行了request的调用，并且是直接调用可以处理的上限s.request(Long.MAX_VALUE);}}
}/*** Check Subscription current state and cancel new Subscription if current is set,* or return true if ready to subscribe.** @param current current Subscription, expected to be null* @param next new Subscription* @return true if Subscription can be used*/
public static boolean validate(@Nullable Subscription current, Subscription next) {Objects.requireNonNull(next, "Subscription cannot be null");if (current != null) {next.cancel();//reportSubscriptionSet();return false;}return true;
}

如果订阅者怀疑其信号处理将对其发布者的响应性产生负面影响，则建议异步调度其信号。
此规则的目的是订阅者不应该从执行的角度阻碍Publisher的进度。换句话说，订阅者不应该使发布者饥饿，没法使用CPU周期。

Subscriber.onComplete()和Subscriber.onError(Throwable t)绝不能调用Subscription或Publisher上的任何方法。
此规则的目的是在处理完成信号期间防止发布者，订阅和订阅者之间的周期和竞争条件。
代码2 LambdaSubscriber类

//这里2个方法都没有调用Subscription或Publisher上的方法@Override
public final void onComplete() {Subscription s = S.getAndSet(this, Operators.cancelledSubscription());if (s == Operators.cancelledSubscription()) {return;}if (completeConsumer != null) {try {completeConsumer.run();}catch (Throwable t) {Exceptions.throwIfFatal(t);onError(t);}}
}@Override
public final void onError(Throwable t) {Subscription s = S.getAndSet(this, Operators.cancelledSubscription());if (s == Operators.cancelledSubscription()) {Operators.onErrorDropped(t, Context.empty());return;}if (errorConsumer != null) {errorConsumer.accept(t);}else {throw Exceptions.errorCallbackNotImplemented(t);}
}

Subscriber.onComplete()和Subscriber.onError(Throwable t)必须考虑在收到信号后取消订阅。
这条规则可以参考Publisher的规则6
订阅者必须在onSubscribe信号之后调用给定订阅的Subscription.cancel()，如果它已经有一个活动的订阅。
此规则的目的是防止两个或更多个单独的发布者认为他们可以与同一个订阅服务器进行交互。执行此规则意味着防止资源泄漏，因为将取消额外的订阅。
这条规则的实现可以参考代码1中的逻辑判断
```
if (Operators.validate(subscription, s)) {
```
如果不再需要订阅，订阅者必须调用Subscription.cancel()。
此规则的目的是确定订阅者不能仅在不再需要订阅时抛弃订阅，他们必须调用取消，以便可以安全，及时地回收该订阅所持有的资源。
订阅者必须确保其订阅上的所有呼叫都来自同一线程或提供相应的外部同步。
此规则的目的是确定如果订阅者将由两个或多个线程同时使用订阅，则必须添加外部同步。
订阅者必须准备好在调用Subscription.cancel()之后接收一个或多个onNext信号，如果仍有请求的元素待解决。 Subscription.cancel()不保证立即执行底层清理操作。
此规则的目的是强调调用cancel()和发布者注意到取消之间可能存在延迟。
订阅者必须准备好接收带有或不带有预设Subscription.request(long n)调用的onComplete信号。
此规则的目的是确定完成与需求流无关，这允许提前完成的流，并且不需要轮询完成。
订阅者必须准备好接收带有或不带有预设Subscription.request(long n)调用的onError信号。
同第9条规则
订阅者必须确保在处理相应信号之前，所有对其信号方法的调用都已经准备好。即订阅者必须注意将信号正确发布到其处理逻辑。
此规则的目的是确定订阅服务器实现的责任是确保其信号的异步处理是线程安全的。请参阅第17.4.5节中的Happens-Before的JMM定义。

对于给定的订阅者，必须最多调用一次Subscriber.onSubscribe（基于对象相等）。

代码3 FluxRange类

//从这个方法可以看到onSubscribe方法的调用最终只会进入到一个判断逻辑，符合了规则12
@Override
@SuppressWarnings("unchecked")
public void subscribe(CoreSubscriber<? super Integer> actual) {long st = start;long en = end;if (st == en) {Operators.complete(actual);return;} elseif (st + 1 == en) {actual.onSubscribe(Operators.scalarSubscription(actual, (int)st));return;}if (actual instanceof ConditionalSubscriber) {actual.onSubscribe(new RangeSubscriptionConditional((ConditionalSubscriber<? super Integer>) actual, st, en));return;}actual.onSubscribe(new RangeSubscription(actual, st, en));
}

调用onSubscribe，onNext，onError或onComplete必须正常返回，除非任何提供的参数为null，在这种情况下它必须向调用者抛出java.lang.NullPointerException，对于所有其他情况，订阅者发出信号失败的唯一合法方式是取消其订阅。在违反此规则的情况下，任何与订阅服务器相关联的订阅必须被视为已取消，并且调用者必须以适合运行时环境的方式提高此错误条件。
此规则的目的是为订阅者的方法建立语义，以及在违反此规则的情况下允许发布者执行的操作。 «以适合运行时环境的方式提高此错误条件»可能意味着记录错误 - 或者让某人或某事意识到这种情况 - 因为错误无法通知错误的订阅者。
这条规则的实现可以参考LambdaSubscriber类中的onSubscribe，onNext，onError和onComplete方法的源码，前面的规则中都有展示过源码，这里就不再赘述了。

Processor规则

public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {}

处理器代表一个处理阶段 - 既是订阅者又是发布者，必须遵守两者的规则条约。
处理器可以选择恢复onError信号。如果它选择这样做，它必须考虑取消订阅，否则它必须立即将onError信号传播给它的订阅者。