序
本文主要研究下java里头的reactive streams与观察者模式。
reactive streams
reactive编程范式是一个异步编程范式，主要涉及数据流及变化的传播，可以看做是观察者设计模式的扩展。
java里头的iterator是以pull模型，即订阅者使用next去拉取下一个数据；而reactive streams则是以push模型为主，订阅者调用subscribe方法订阅，发布者调用订阅者的onNext通知订阅者新消息。
reactive streams java api
reactive streams定义了4个java api，如下
Processor<T,R>
processor既是Subscriber也是Publisher，代表二者的处理阶段
Publisher<T>
publisher是数据的提供者, 将数据发布给订阅者
Subscriber<T>
在调用Publisher.subscribe(Subscriber)之后，Subscriber.onSubscribe(Subscription)将会被调用
Subscription
Subscription代表订阅者与发布者的一次订阅周期，一旦调用cancel去掉订阅，则发布者不会再推送消息。
观察者模式
观察者模式的实现有推模型和拉模型
- 拉模型
即发布者通知订阅有新消息，订阅者再去找发布者拉取
- 推模型
即发布者通知订阅者有消息，通知的时候已经带上了一个新消息
reactor实例
maven
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-core</artifactId>
<version>3.1.2.RELEASE</version>
</dependency>
reactor 3 是java里头reactive streams的一个实现，基于reactive streams的java api，是spring 5反应式编程的基础。
Flux实例
@Test
public void testBackpressure(){
Flux.just(1, 2, 3, 4)
.log()
.subscribe(new Subscriber<Integer>() {
private Subscription s;
int onNextAmount;
@Override
public void onSubscribe(Subscription s) {
this.s = s;
s.request(2);
}
@Override
public void onNext(Integer integer) {
System.out.println(integer);
onNextAmount++;
if (onNextAmount % 2 == 0) {
s.request(2);
}
}
@Override
public void onError(Throwable t) {}
@Override
public void onComplete() {}
});
try {
Thread.sleep(10*1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
小结
从上面的代码看，reactive streams实际上是推拉结合的模式的结合。为什么还要拉呢？
rabbitmq vs kafka
rabbitmq是以推为主的，如果消费者消费能力跟不上，则消息会堆积在内存队列中(
必要时可能写磁盘)
kafka则是以拉为主的，生产者推送消息到broker，消费者自己根据自己的能力从broker拉取消息，由于消息是持久化的，因此无需关心生产消费速率的不平衡
backpressure
backpressure这个是为处理生产速率与消费速率不平衡这个问题而衍生出来的，订阅者可以在next方法里头根据自己的情况，使用request方法告诉发布者要取N个数据，发布者则向订阅者推送N个数据。通过request达到订阅者对发布者的反馈。而对于发布者而言，为了实现backpressure，则需要有一个缓存队列来缓冲订阅者没来得及消费的数据。涉及到缓冲，就涉及容量是有界还是无界，如果是有界则在缓冲慢的时候，处理策略是怎样等等。
