从观察者模式到Reactive Stream

Reactive 的系统就是通过实现回弹性、弹性能够对客户请求进行及时响应的系统。

正如 Reactive Manifesto 说的：

Systems built as Reactive Systems are more flexible, loosely-coupled and scalable.

说到松耦合，最基本的模式之一就算是“观察者模式”了。从观察者模式更进一步，可以衍生出“发布订阅模式”。较之前者，发布订阅模式多了一个‘事件通道’的角色。通过这个通道，使得订阅者不必知道发布者——两者解耦。Spring 在应用层面对“发布-订阅”模式进行了不错的支持(同步的、异步的)，诸如@EventListener 等的日常用法参见： https://www.baeldung.com/spring-events https://reflectoring.io/spring-boot-application-events-explained/

响应式流 = 观察者 + 迭代器

从 Iterator 定义看，Iterator是一个“拉”模型，当我们需要数据时通过调用 next()拉取，hasNext()是查询是否还有数据可用。如果上游数据有什么异常情况，只能是通过 next()抛出的异常感知甚至感知不到。public interface Iterator { E next(); boolean hasNext(); }

以 RxJava 的 RxObserver 为例，感受下其接口语义上的改进：

public interface RxObserver<T> {
  void onNext(T next);
  void onComplete();
  void onError();
}

以 on 形式的前缀开头的方法名让我很明显的感受到一种“通知”的味道。上游组件告诉我们有新数据了、结束了、出错了！一种“推”的味道。

演进到这一步是不是就完美了呢？不。 如果 RxObserver 与上游组件是跨网络通信，那么我们可以想象每次的 onNext 通过网络一次处理一个数据的这种模式并不高效。而且 RxObserver 不能想上游表达自己的需求(比如需要几个数据？不再需要数据等)，这就很容易延伸到 RxObserver 作为消费者的处理速度与上游不匹配时如何与上游协调工作这样的问题。 Reactive Stream 规范对以上需求进行了标准化，如下。其中 Subscription 就可用于向上游 Publisher 表达是否还需要、需要多少数据这样的请求。如果每次 request(1)，整个模式就相当于“拉”模型；request(Integer.MAX_VALUE)就相当于“推”模型了。

public interface Publisher<T> {
    public void subscribe(Subscriber<? super T> s);
}

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

public interface Subscription {
    public void request(long n);
    public void cancel();
}

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

Reference：

https://github.com/reactive-streams/reactive-streams-jvm/blob/master/README.md

http://reactivex.io/

A Playful Introduction to Rx by Erik Meijer

https://www.baeldung.com/spring-events https://reflectoring.io/spring-boot-application-events-explained/