初见-响应式编程-002

🤗 ApiHug × {Postman|Swagger|Api...} = 快↑ 准√ 省↓

GitHub - apihug/apihug.com: All abou the Apihug
apihug.com: 有爱，有温度，有质量，有信任
ApiHug - API design Copilot - IntelliJ IDEs Plugin | Marketplace

#Reactive

The Reactive Manifestoopen in new window:

Systems built as Reactive Systems are more flexible, loosely-coupled and scalable. This makes them easier to develop and amenable to change. They are significantly more tolerant of failure and when failure does occur they meet it with elegance rather than disaster. Reactive Systems are highly responsive, giving users effective interactive feedback.

flexible,
loosely-coupled
scalable

Responsive，响应时间，服务质量
Resilient，容错，恢复能力
Elastic，弹性，动态扩容
Message Driven，事件驱动，低耦合

#Reactor

Reactor is an implementation of the Reactive Programming paradigm, which can be summed up as follows:

Reactive programming is an asynchronous programming paradigm concerned with data streams and the propagation of change. This means that it becomes possible to express static (e.g. arrays) or dynamic (e.g. event emitters) data streams with ease via the employed programming language(s). — https://en.wikipedia.org/wiki/Reactive_programmingopen in new window

最初微软创建了 .NET 里的 Reactive Extensions (Rx)； RxJava 实现了 JVM 上的 Reactive编程模式，最终 JAVA 9 融入了 Flowopen in new window -- java.util.concurrent.Flowopen in new window

在 OO 编程里面 reactive 常被当做 Observer 观察者设计模式，当然你可把 Reactive Stream 和你熟悉的 Iterator 设计模式做对比；两种实现里面都有Iterable-Iterator 两个概念，主要的不一样在， Iterator 是一种拉 pull 模式，而 reactive 是一种push 推模式。

在两个流程中都有 next()，在 reactive stream 更类似于 Publisher-Subscriber 模式，由Publishr 来控制新到的value 给 Subscriber； push 是 reactive 里面非常重要的一面；

程序实现着注重收到 value 后的计算逻辑 Operation，而不是整个控制流程。

整个流程里面喂入数据 push 自然是整个响应流程里面最核心的流程， onNext 用来完成此动作，还包含错误 onErro() 和最终的结束处理 onComplete(), 整个流程可以被抽象为:

onNext x 0..N [onError | onComplete]

整个流程的处理可以说非常灵活，可以有 0 个， 1个， N 个，或者无限多的数据，比如一个定时器。

回到问题的本质， WHY 我们为什么需要异步的 reactive 模式呢？

#Asynchronicity 能解决

并行，多核已经是常态来增加吞吐量和响应时间。

多线程用来最大化利用资源；但是多线程异步可以解决问题通知，带来了很大的挑战。 JVM 解决此引入两个概念：

callback, 异步方法用来通知结果，一般是一个内部类，或者一个 lamdba 表达式
future，异步调用立即返回一个 Future<T>，但是结果 T 尚不能立即获得，结果获得后才能通过 poll 获得。 ExecutorService 跑 Callable<T> 时返回 Future。

#Callback地狱

callback 贯穿整个链路的调用过程：

userService.getFavorites(userId, new Callback<List<String>>() { public void onSuccess(List<String> list) { if (list.isEmpty()) { suggestionService.getSuggestions(new Callback<List<Favorite>>() {public void onSuccess(List<Favorite> list) { UiUtils.submitOnUiThread(() -> { list.stream().limit(5).forEach(uiList::show); });}public void onError(Throwable error) { UiUtils.errorPopup(error);}});} else {list.stream() .limit(5).forEach(favId -> favoriteService.getDetails(favId, new Callback<Favorite>() {public void onSuccess(Favorite details) {UiUtils.submitOnUiThread(() -> uiList.show(details));}public void onError(Throwable error) {UiUtils.errorPopup(error);}}));}}public void onError(Throwable error) {UiUtils.errorPopup(error);}
});

如果换成 reactor:

userService.getFavorites(userId) .flatMap(favoriteService::getDetails) .switchIfEmpty(suggestionService.getSuggestions()) .take(5) .publishOn(UiUtils.uiThreadScheduler()) .subscribe(uiList::show, UiUtils::errorPopup);

reactor 不仅仅让整个流程更精简，通知提供服务质量控制(类似熔断)，比如我们保证整个服务质量在 800ms内返回，超时后从 fallback cache 或者其他获取:

userService.getFavorites(userId).timeout(Duration.ofMillis(800)) .onErrorResume(cacheService.cachedFavoritesFor(userId)) .flatMap(favoriteService::getDetails) .switchIfEmpty(suggestionService.getSuggestions()).take(5).publishOn(UiUtils.uiThreadScheduler()).subscribe(uiList::show, UiUtils::errorPopup);

#Future

Future 避免了回调地狱，但是依然不太容易进行组装，虽然在 Java 8 里面引入了 CompletableFuture, 编制多个 Future 在一起虽然可以操作，但是Future 还是有其他问题：

Future 的 get() 依然是阻塞的
不支持延迟计算
缺乏对多结果的支持，更好的错误处理

这样一个业务场景; 从一个 ID 列表，去查询他们的name + 统计，所有的都是异步 CompletableFuture 例子:

CompletableFuture<List<String>> ids = ifhIds(); CompletableFuture<List<String>> result = ids.thenComposeAsync(l -> { Stream<CompletableFuture<String>> zip =l.stream().map(i -> { CompletableFuture<String> nameTask = ifhName(i); CompletableFuture<Integer> statTask = ifhStat(i); return nameTask.thenCombineAsync(statTask, (name, stat) -> "Name " + name + " has stats " + stat); });List<CompletableFuture<String>> combinationList = zip.collect(Collectors.toList()); CompletableFuture<String>[] combinationArray = combinationList.toArray(new CompletableFuture[combinationList.size()]);CompletableFuture<Void> allDone = CompletableFuture.allOf(combinationArray); return allDone.thenApply(v -> combinationList.stream().map(CompletableFuture::join) .collect(Collectors.toList()));});List<String> results = result.join(); 
assertThat(results).contains("Name NameJoe has stats 103","Name NameBart has stats 104","Name NameHenry has stats 105","Name NameNicole has stats 106",

reactor 更紧凑解决方案，更精炼容易理解：

Flux<String> ids = ifhrIds(); Flux<String> combinations =ids.flatMap(id -> { Mono<String> nameTask = ifhrName(id); Mono<Integer> statTask = ifhrStat(id); return nameTask.zipWith(statTask, (name, stat) -> "Name " + name + " has stats " + stat);});Mono<List<String>> result = combinations.collectList(); List<String> results = result.block(); 
assertThat(results).containsExactly( "Name NameJoe has stats 103","Name NameBart has stats 104","Name NameHenry has stats 105","Name NameNicole has stats 106","Name NameABSLAJNFOAJNFOANFANSF has stats 121"
);

#响应编程

除非上面我们看到，让编码更清晰直观， reactor 等还在这些方面上花了很多心思：

Composability and readability；组件化，可读性好。
Data as a flow manipulated with a rich vocabulary of operators，数据管道铺好，自由搭配运算逻辑。
Nothing happens until you subscribe，延迟计算 subscribe 触发计算。
Backpressure or the ability for the consumer to signal the producer that the rate of emission is too high，背压控制， consumer 和 producer配合，消费端和生产端。
High level but high value abstraction that is concurrency-agnostic；润物细无声，抽象的让你感知不到并发。

Composability and Readability 包括可以自由的组建编制任务，任务之间的依赖关系，上下关系，或者同步运行的 fork-join 风格，高度抽象层使用异步任务。

流水线一样的操作， Reactor 既是传送带也是工作站，原料从 Publisher 最终成品发送到消费端 Subscriber。

Operators, 相当于流水线上的工作站，整个流水线上就是上一个 Publisher 的产物，然后包装发送到下一个 Publisher，最终到一个 Subsccriber 里面。

延迟计算，在Reactor 里当你写一个 Publisher 链条，数据默认是不会启动起来，你只是创建了一个异步处理的抽象流程(Spark 里的RDD, 或者像一个流程的 DSL)。

当你触发 subscribing 时候，将 Publisher 和一个 Subscriber 绑定，同事触发数据流，流入到整个链中，内部通过 Subscriber 触发一个 request 信号传递到上游，最终到源的 Publisher。

背压，下游将信号传递给上游是用来实现 backpressure背压的一种方式，依然用流水线的比方，当下游的工作站赶不上上游的速度的时候需要反馈一个信号到上游去。

A subscriber can work in unbounded mode and let the source push all the data at its fastest achievable rate or it can use the request mechanism to signal the source that it is ready to process at most n elements.

可以 push-pull 方式混合，下游容量自由控制上游的推送速度，或者懒式的拉取。

Cold流和Hot流

Cold流不论订阅者在何时订阅该数据流，总是能收到数据流中产生的全部消息。
Hot流则是在持续不断地产生消息，订阅者只能获取到在其订阅之后产生的消息。

冷例子：

@Testpublic void cold_example() {Flux<String> source =Flux.fromIterable(Arrays.asList("blue", "green", "orange", "purple")).map(String::toUpperCase);source.subscribe(d -> System.out.println("Subscriber 1: " + d));source.subscribe(d -> System.out.println("Subscriber 2: " + d));
}

输出结果：

Subscriber 1: BLUE
Subscriber 1: GREEN
Subscriber 1: ORANGE
Subscriber 1: PURPLE
Subscriber 2: BLUE
Subscriber 2: GREEN
Subscriber 2: ORANGE
Subscriber 2: PURPLE

所有的 subscriber 都能得到结果。

热例子：

@Testpublic void hot_example() {Sinks.Many<String> hotSource = Sinks.unsafe().many().multicast().directBestEffort();Flux<String> hotFlux = hotSource.asFlux().map(String::toUpperCase);hotFlux.subscribe(d -> System.out.println("Subscriber 1 to Hot Source: " + d));hotSource.emitNext("blue", FAIL_FAST);hotSource.tryEmitNext("green").orThrow();hotFlux.subscribe(d -> System.out.println("Subscriber 2 to Hot Source: " + d));hotSource.emitNext("orange", FAIL_FAST);hotSource.emitNext("purple", FAIL_FAST);hotSource.emitComplete(FAIL_FAST);}

得到结果：

Subscriber 1 to Hot Source: BLUE
Subscriber 1 to Hot Source: GREEN
Subscriber 1 to Hot Source: ORANGE
Subscriber 2 to Hot Source: ORANGE
Subscriber 1 to Hot Source: PURPLE
Subscriber 2 to Hot Source: PURPLE