diff --git a/spring/webflux/Reactor.md b/spring/webflux/Reactor.md index 1782e4c..202c18b 100644 --- a/spring/webflux/Reactor.md +++ b/spring/webflux/Reactor.md @@ -48,6 +48,17 @@ reactive stream定义的机制接近于上述描述,其提供两种模式: 对于响应式序列,其可以分为两种: - cold sequence:对于cold sequence,会为每个订阅者重新开始流程,包括source publisher。例如source中若封装了http调用,会为每个subscriber都执行一个新的http请求 - hot sequence:subscriber只有在其订阅后才收到信号,即使没有subscriber在监听,hot sequence仍然能够发送signal + +## Subscriber和Publisher +### Publisher +对于publisher,其提供了`subscribe`方法供subscriber进行注册,在执行subscribe方法并向其传入`Subscriber`对象后,上游publisher会调用下游的`onSubscribe`方法,并向`onSubscribe`方法传入`Subscription`对象。下游可以通过`Subscription`对象调用`request(n)`请求。 + +> 若中间存在operator(例如map)在担任publisher角色的同时,还对上游进行了订阅,那么对上游的实际订阅操作只会在operator被下游subscriber订阅时触发。 +> +> `任何变更状态的操作都只在实际subscriber执行订阅操作后被触发`。 + +### Subscriber +当下游调用`request(n)`方法之后,会向上游请求`n`个数据。上游会向下游发送`onNext`信号来传输生成的数据。 ## Reactor Core reactor引入了两个实现`Publisher`的类:`Mono`和`Flux`。 @@ -69,7 +80,75 @@ Flux是一个通用的reactive类型,并且,所有的event type都是可选 - 无限序列并不一定为空,例如`Flux.interval(Duration)`会产生一个`Flux`,其是无限的并且发送tick ### Mono `0...1` -`Mono`是一个标准的`Publisher`,其通过`onNext`信号发送至多一个item,然后通过`onComplete`信号结束(成功场景);或直接通过`onError`信号结束(失败场景)。 +`Mono`是一个标准的`Publisher`,其通过`onNext`信号发送至多一个item,然后再结束时发送`onComplete`信号结束(成功场景);或直接发送`onError`信号结束(失败场景)。 大多数Mono实现在调用完subscriber的`onNext`方法之后,预计会立马调用subscriver的`onComplete`方法。但是,`Mono.never`是一个例外,其并不会发送任何信号,并且其`onNext`和`onError`的组合是被明确禁止的。 +### 创建Mono/Flux并进行订阅的方式 +#### String sequence +如果想要创建String序列,可以通过如下方式: +```java +Flux seq1 = Flux.just("foo", "bar", "foobar"); + +List iterable = Arrays.asList("foo", "bar", "foobar"); +Flux seq2 = Flux.fromIterable(iterable); +``` +#### Flux.empty +```java +Mono noData = Mono.empty(); +``` +#### Flux.range +在下面示例中,`Flux.range`第一个参数是range的起始值,第二个参数是要产生的元素个数 +```java +Flux numbersFromFiveToSeven = Flux.range(5, 3); +``` +故而,其产生的内容为`5,6,7`。 + +#### Lambda Subscribe +在进行订阅时,`Flux`和`Mono`使用了lambda,在调用subscribe时,有如下几种重载的选择: +```java +// 订阅并触发sequence的产生 +subscribe(); + +// 对每个产生的值通过consumer执行处理操作 +subscribe(Consumer consumer); + +// 在reactive stream异常终止时,对error进行处理 +subscribe(Consumer consumer, + Consumer errorConsumer); + +// 在sequence处理完时,执行额外的complete操作 +subscribe(Consumer consumer, + Consumer errorConsumer, + Runnable completeConsumer); + +// 该方法会针对`subscribe`方法产生的`Subscription`对象执行操作 +// 该重载已废弃 +subscribe(Consumer consumer, + Consumer errorConsumer, + Runnable completeConsumer, + Consumer subscriptionConsumer); +``` +对于subscribe的使用,示例如下 +```java +Flux.range(5,3) + .map(x->{ + if(x<7) { + return x; + } + throw new RuntimeException(String.format("fucking value {%s} equals or greater than 7", x)); + }) + .subscribe(v->System.out.printf("[%s]-%d\n", Thread.currentThread().getName(), v), + (e) -> { + System.out.printf("[%s]-Error Caught: %s\n", Thread.currentThread().getName(), e.getMessage()); + }, + ()->{ + System.out.printf("[%s]-Complete: %s\n", Thread.currentThread().getName(), "fucking ended"); + }); +``` +其执行结果如下: +``` +[main]-5 +[main]-6 +[main]-Error Caught: fucking value {7} equals or greater than 7 +``` \ No newline at end of file