响应式编程简介
目录
Reactor 是响应式编程范式的一种实现，该范式可以概括如下：
响应式编程是一种异步编程范式，关注数据流与变更的传播。这意味着，借助所使用的编程语言，可以轻松表达静态（如数组）或动态（如事件发射器）的数据流。
作为迈向响应式编程的第一步，Microsoft 在 .NET 生态中创建了 Reactive Extensions（Rx）库。随后 RxJava 在 JVM 上实现了响应式编程。随着时间推移，针对 Java 的标准化通过 Reactive Streams 工作而出现，这是一份规范，它为 JVM 上的响应式库定义了一组接口与交互规则。其接口已被整合进 Java 9 的 Flow 类中。
在面向对象语言中，响应式编程范式常常被呈现为观察者设计模式的扩展。您也可以将主要的响应式流模式与大家熟悉的迭代器设计模式做对比，因为在所有这些库中都存在与 Iterable-Iterator 配对的二元对应物。一个主要区别在于：迭代器是拉取（pull）模式，而响应式流是推送（push）模式。
使用迭代器是一种命令式编程模式，尽管取值的方式完全由 Iterable 负责。确实，何时访问序列中的下一个 next() 元素，完全由开发者决定。在响应式流中，与上述配对等价的是 Publisher-Subscriber。但向订阅者通知新可用值的是 Publisher，并且是在值到来时即时通知，这种推送方面正是响应式的关键。此外，施加于被推送值之上的操作是以声明式而非命令式表达的：程序员描述计算的逻辑，而不是描述其确切的控制流。
除了推送值之外，错误处理与完成方面也以明确的方式得到涵盖。Publisher 可以向其 Subscriber 推送新值（通过调用 onNext），也可以发出错误信号（通过调用 onError）或完成信号（通过调用 onComplete）。错误与完成都会终止序列。这可以概括如下：
onNext x 0..N [onError | onComplete]
这种方式非常灵活。该模式支持没有值、一个值或 n 个值的用例（包括无限的值序列，例如时钟持续不断的滴答）。
但是，我们究竟为什么首先需要这样一个异步响应式库呢？
1. 阻塞可能造成浪费
现代应用可能面临海量的并发用户，尽管现代硬件的能力在不断进步，现代软件的性能仍然是关键关切。
大致来说，改善程序性能有两种途径：
-
并行化：使用更多线程与更多硬件资源。
-
追求更高效率：改进现有资源的使用方式。
通常，Java 开发者使用阻塞代码编写程序。在出现性能瓶颈之前，这种做法没问题。这时就该引入额外的线程，运行类似的阻塞代码。但这种资源利用上的扩展很快就会带来争用与并发问题。
更糟的是，阻塞会浪费资源。仔细观察就会发现：一旦程序涉及某种延迟（尤其是 I/O，例如数据库请求或网络调用），资源就被浪费了，因为线程（可能是许多线程）现在闲置等待数据。
因此，并行化应用并非银弹。它是充分发挥硬件威力所必需的，但也复杂难懂，且容易造成资源浪费。
2. 异步来救场？
前面提到的第二种途径——追求更高效率——可以成为解决资源浪费问题的方案。通过编写异步、非阻塞代码，您让执行切换到另一个使用相同底层资源的活动任务上，待异步处理完成后再切回当前进程。
但是，如何在 JVM 上编写异步代码呢？Java 提供了两种异步编程模型：
-
回调（Callbacks）：异步方法没有返回值，而是额外接受一个
callback参数（一个 lambda 或匿名类），当结果可用时被调用。一个广为人知的例子是 Swing 的EventListener层次结构。 -
Future：异步方法_立即_返回一个
Future<T>。异步过程计算一个T值，但Future对象封装了对它的访问。该值并非立即可用，可以轮询该对象直到值可用。例如，运行Callable<T>任务的ExecutorService就使用Future对象。
这些技术够好吗？并非适用于所有用例，而且两种途径都有局限。
回调难以相互组合，很快就会导致代码难以阅读和维护（即众所周知的「回调地狱」）。
来看一个例子：在 UI 上向用户展示他最喜爱的前五个条目，或者在他没有收藏时给出推荐。这要经过三个服务（一个给出收藏 ID，第二个获取收藏详情，第三个给出带详情的推荐），如下所示：
userService.getFavorites(userId, new Callback<List<String>>() { 1
public void onSuccess(List<String> list) { 2
if (list.isEmpty()) { 3
suggestionService.getSuggestions(new Callback<List<Favorite>>() {
public void onSuccess(List<Favorite> list) { 4
UiUtils.submitOnUiThread(() -> { 5
list.stream()
.limit(5)
.forEach(uiList::show); 6
});
}
public void onError(Throwable error) { 7
UiUtils.errorPopup(error);
}
});
} else {
list.stream() 8
.limit(5)
.forEach(favId -> favoriteService.getDetails(favId, 9
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);
}
});
我们有基于回调的服务：一个
Callback接口，其一个方法在异步过程成功时被调用，另一个在发生错误时被调用。第一个服务带着收藏 ID 列表调用其回调。
如果列表为空，我们必须转到
suggestionService。suggestionService向第二个回调给出一个List<Favorite>。由于我们处理的是 UI，需要确保消费代码运行在 UI 线程中。
我们使用 Java 8 的
Stream将处理的推荐数量限制为五个，并在 UI 的图形列表中展示它们。在每一层，我们以相同方式处理错误：在弹窗中显示。
回到收藏 ID 这一层。如果服务返回了完整列表，我们需要转到
favoriteService获取带详情的Favorite对象。由于我们只想要五个，我们先对 ID 列表做流式处理以将其限制为五个。再来一个回调。这一次我们得到一个填充完整的
Favorite对象，并在 UI 线程内将其推送到 UI。
代码量不小，而且有点难以跟随，还有重复的部分。看看它在 Reactor 中的等价写法：
userService.getFavorites(userId) 1
.flatMap(favoriteService::getDetails) 2
.switchIfEmpty(suggestionService.getSuggestions()) 3
.take(5) 4
.publishOn(UiUtils.uiThreadScheduler()) 5
.subscribe(uiList::show, UiUtils::errorPopup); 6
我们从一个收藏 ID 的流开始。
我们_异步地_将它们转换为带详情的
Favorite对象（flatMap）。现在我们有了一个Favorite的流。如果
Favorite流为空，我们通过suggestionService切换到回退方案。我们最多只关心结果流中的五个元素。
最后，我们希望在 UI 线程中处理每一份数据。
我们通过描述如何处理数据的最终形态（在 UI 列表中展示它）以及在出错时如何处理（弹出弹窗）来触发这个流。
如果您想确保收藏 ID 在 800ms 内取回，或者在耗时更长时从缓存获取，该怎么办？在基于回调的代码中，这是一项复杂的任务。在 Reactor 中，它变得像在链中添加一个 timeout 操作符一样简单，如下所示：
userService.getFavorites(userId)
.timeout(Duration.ofMillis(800)) 1
.onErrorResume(cacheService.cachedFavoritesFor(userId)) 2
.flatMap(favoriteService::getDetails) 3
.switchIfEmpty(suggestionService.getSuggestions())
.take(5)
.publishOn(UiUtils.uiThreadScheduler())
.subscribe(uiList::show, UiUtils::errorPopup);
如果上面这一部分在 800ms 内没有发出任何元素，就传播一个错误。
出错时，回退到
cacheService。链的其余部分与前一个示例类似。
Future 对象比回调稍好一些，但它们在组合方面仍然表现不佳，尽管 Java 8 通过 CompletableFuture 带来了改进。编排多个 Future 对象是可行的，但并不容易。此外，Future 还有其他问题：
-
通过调用
get()方法，很容易再次陷入阻塞局面。 -
它们不支持惰性计算。
-
它们缺乏对多值与高级错误处理的支持。
再看一个例子：我们得到一个 ID 列表，想从中获取名称和一个统计值，并将它们成对组合，全部异步完成。以下示例用一个 CompletableFuture 类型的列表来完成这件事：
CompletableFuture<List<String>> ids = ifhIds(); 1
CompletableFuture<List<String>> result = ids.thenComposeAsync(l -> { 2
Stream<CompletableFuture<String>> zip =
l.stream().map(i -> { 3
CompletableFuture<String> nameTask = ifhName(i); 4
CompletableFuture<Integer> statTask = ifhStat(i); 5
return nameTask.thenCombineAsync(statTask, (name, stat) -> "Name " + name + " has stats " + stat); 6
});
List<CompletableFuture<String>> combinationList = zip.collect(Collectors.toList()); 7
CompletableFuture<String>[] combinationArray = combinationList.toArray(new CompletableFuture[combinationList.size()]);
CompletableFuture<Void> allDone = CompletableFuture.allOf(combinationArray); 8
return allDone.thenApply(v -> combinationList.stream()
.map(CompletableFuture::join) 9
.collect(Collectors.toList()));
});
List<String> results = result.join(); 10
assertThat(results).contains(
"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");
我们从一个 future 开始，它给出一个待处理的
id值列表。一旦拿到列表，我们想启动更深入的异步处理。
对列表中的每个元素：
异步获取关联的名称。
异步获取关联的统计值。
组合两个结果。
现在我们有一个 future 列表，代表所有组合任务。要执行这些任务，我们需要将列表转换为数组。
将数组传给
CompletableFuture.allOf，它输出一个Future，在所有任务完成时完成。棘手之处在于
allOf返回的是CompletableFuture<Void>，因此我们遍历 future 列表，通过使用join()收集它们的结果（在这里它不会阻塞，因为allOf确保所有 future 都已完成）。一旦整个异步流水线被触发，我们等待它处理完毕，并返回可以断言的结果列表。
由于 Reactor 开箱即提供更多组合操作符，这个过程可以被简化，如下所示：
Flux<String> ids = ifhrIds(); 1
Flux<String> combinations =
ids.flatMap(id -> { 2
Mono<String> nameTask = ifhrName(id); 3
Mono<Integer> statTask = ifhrStat(id); 4
return nameTask.zipWith(statTask, 5
(name, stat) -> "Name " + name + " has stats " + stat);
});
Mono<List<String>> result = combinations.collectList(); 6
List<String> results = result.block(); 7
assertThat(results).containsExactly( 8
"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"
);
这一次，我们从一个异步提供的
ids序列（一个Flux<String>）开始。对序列中的每个元素，我们（在
flatMap调用体的函数中）异步处理两次。获取关联的名称。
获取关联的统计值。
异步组合这两个值。
在值可用时将其聚合到一个
List中。在生产环境中，我们会异步地继续处理这个
Flux，进一步组合它或订阅它。大多数情况下，我们会返回result这个Mono。由于我们是在测试中，改为阻塞，等待处理完成，然后直接返回聚合后的值列表。断言结果。
使用回调与 Future 对象的困境是相似的，而这正是响应式编程用 Publisher-Subscriber 配对所要解决的问题。
3. 从命令式编程到响应式编程
像 Reactor 这样的响应式库，旨在解决 JVM 上这些「经典」异步途径的缺点，同时还关注以下几个附加方面：
-
可组合性与可读性
-
数据作为一个流，用丰富的操作符词表进行处理
-
不订阅就什么都不会发生
-
背压，即_消费者能够向生产者发出信号，告知其发出速率过高_
-
高层级但高价值的抽象，并且与_并发无关_
3.1. 可组合性与可读性
所谓「可组合性」，指的是编排多个异步任务的能力：我们使用前一个任务的结果作为后续任务的输入。或者，我们可以以 fork-join（分叉-汇合）的方式运行若干任务。此外，我们还可以将异步任务作为离散组件复用于更高层的系统中。
编排任务的能力与代码的可读性、可维护性紧密耦合。随着异步处理的层数在数量与复杂度上不断增长，组合与阅读代码变得越来越困难。如我们所见，回调模型很简单，但它的主要缺点之一是：对于复杂的过程，你需要让一个回调在另一个回调中执行，而那个回调本身又嵌套在又一个回调里，如此层层叠加。那团乱麻就是众所周知的「回调地狱」。正如您能猜到的（或从经验中得知的），这样的代码非常难以回头去理解和推理。
Reactor 提供了丰富的组合选项，代码在其中映照出抽象过程的组织方式，并且一切通常都保持在同一层级（嵌套被降到最低）。
3.2. 装配线类比
您可以把响应式应用所处理的数据想象成在一条装配线上移动。Reactor 既是传送带，也是工作站。原材料从一个来源（原始 Publisher）涌出，最终成为成品，准备好被推送给消费者（即 Subscriber）。
原材料可以经过各种转换和其他中间步骤，也可以成为更大装配线的一部分，将中间件聚合在一起。如果在某一点出现故障或堵塞（也许装箱产品耗时长得不成比例），受影响的工作站可以向上游发出信号，限制原材料的流量。
3.3. 操作符
在 Reactor 中，操作符就是我们装配线类比中的工作站。每个操作符都为一个 Publisher 添加行为，并把上一步的 Publisher 包装成一个新实例。整个链因此被链接起来：数据源自第一个 Publisher，沿链向下移动，并被每个环节转换。最终，一个 Subscriber 完成整个处理过程。请记住：在 Subscriber 订阅 Publisher 之前，什么都不会发生，我们稍后就会看到这一点。
理解操作符会创建新实例，可以帮助您避免一个常见错误：误以为链中使用的某个操作符没有生效。请参阅 FAQ 中的这一条。
虽然 Reactive Streams 规范完全没有规定操作符，但响应式库（如 Reactor）带来的最大附加价值之一，就是它们提供的丰富操作符词表。这些操作符覆盖的范围很广，从简单的转换与过滤，到复杂的编排与错误处理。
3.4. 不订阅就什么都不会发生
在 Reactor 中，当您编写一个 Publisher 链时，数据默认并不会开始流入其中。相反，您创建的是对异步过程的抽象描述（这有助于复用与组合）。
通过订阅这一动作，您把 Publisher 与一个 Subscriber 关联起来，从而触发整条链中的数据流。这在内部是通过 Subscriber 发出的单个 request 信号实现的，该信号向上游传播，一直回到源头 Publisher。
3.5. 背压
向上游传播信号也被用于实现背压。我们在装配线类比中把背压描述为：当某个工作站处理得比上游工作站慢时，沿流水线向上发出的反馈信号。
Reactive Streams 规范定义的真实机制与这个类比相当接近：订阅者可以以_无界_模式工作，让源头以其所能达到的最快速率推送所有数据；也可以使用 request 机制向源头发出信号，告知它自己最多准备好处理 n 个元素。
中间操作符也可以在传输途中改变请求量。想象一个 buffer 操作符，它把元素分组为十个一批。如果订阅者请求一个缓冲区，那么源头生成十个元素是可以接受的。一些操作符还实现了预取策略，它避免了 request(1) 的往返开销，并且如果在被请求之前就生成元素的代价不高，这种策略是有益的。
这把推送模型转变成了一种推-拉混合模式：当下游能够轻松获得 n 个元素时，可以从上游拉取；而当元素尚未准备好时，它们在被生成时由上游推送下来。
3.6. 热与冷
Rx 家族的响应式库区分了响应式序列的两大类别：热与冷。这种区分主要与响应式流如何应对订阅者有关：
-
冷序列会为每个
Subscriber重新开始，包括在数据源头处。例如，如果源头包装的是一次 HTTP 调用，那么每次订阅都会发起一次新的 HTTP 请求。 -
热序列不会为每个
Subscriber从头开始。相反，较晚的订阅者会收到在其订阅_之后_发出的信号。不过请注意，某些热响应式流可以全部或部分地缓存或重放发出历史。从一般角度看，热序列甚至可以在没有任何订阅者监听时就发出信号（这是对「不订阅就什么都不会发生」这一规则的例外）。
要进一步了解 Reactor 语境中的热与冷，请参阅这个 Reactor 专属小节。