doc: 阅读reactor-core文档

spring/webflux/Reactor.md

@@ -49,6 +49,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`。
 - Flux：代表包含`0...N`个items的reactive sequence
@@ -69,7 +80,75 @@ Flux是一个通用的reactive类型，并且，所有的event type都是可选
 - 无限序列并不一定为空，例如`Flux.interval(Duration)`会产生一个`Flux<Long>`，其是无限的并且发送tick
 
 ### Mono `0...1`
-`Mono<T>`是一个标准的`Publisher<T>`，其通过`onNext`信号发送至多一个item，然后通过`onComplete`信号结束（成功场景）；或直接通过`onError`信号结束（失败场景）。
+`Mono<T>`是一个标准的`Publisher<T>`，其通过`onNext`信号发送至多一个item，然后再结束时发送`onComplete`信号结束（成功场景）；或直接发送`onError`信号结束（失败场景）。
 
 大多数Mono实现在调用完subscriber的`onNext`方法之后，预计会立马调用subscriver的`onComplete`方法。但是，`Mono.never`是一个例外，其并不会发送任何信号，并且其`onNext`和`onError`的组合是被明确禁止的。
 
+### 创建Mono/Flux并进行订阅的方式
+#### String sequence
+如果想要创建String序列，可以通过如下方式：
+```java
+Flux<String> seq1 = Flux.just("foo", "bar", "foobar");
+
+List<String> iterable = Arrays.asList("foo", "bar", "foobar");
+Flux<String> seq2 = Flux.fromIterable(iterable);
+```
+#### Flux.empty
+```java
+Mono<String> noData = Mono.empty();
+```
+#### Flux.range
+在下面示例中，`Flux.range`第一个参数是range的起始值，第二个参数是要产生的元素个数
+```java
+Flux<Integer> numbersFromFiveToSeven = Flux.range(5, 3);
+```
+故而，其产生的内容为`5,6,7`。
+
+#### Lambda Subscribe
+在进行订阅时，`Flux`和`Mono`使用了lambda，在调用subscribe时，有如下几种重载的选择：
+```java
+// 订阅并触发sequence的产生
+subscribe(); 
+
+// 对每个产生的值通过consumer执行处理操作
+subscribe(Consumer<? super T> consumer); 
+
+// 在reactive stream异常终止时，对error进行处理
+subscribe(Consumer<? super T> consumer,
+          Consumer<? super Throwable> errorConsumer); 
+
+// 在sequence处理完时，执行额外的complete操作
+subscribe(Consumer<? super T> consumer,
+          Consumer<? super Throwable> errorConsumer,
+          Runnable completeConsumer);
+
+// 该方法会针对`subscribe`方法产生的`Subscription`对象执行操作
+// 该重载已废弃
+subscribe(Consumer<? super T> consumer,
+          Consumer<? super Throwable> errorConsumer,
+          Runnable completeConsumer,
+          Consumer<? super Subscription> 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
+```