Reactor是响应式编程规范的一个实现，维基百科对响应式编程的总结如下：
响应式编程是一种异步编程范例，主要关注数据流和数据变化通知。这意味着可以使用编程语言轻松表达静态（数组）或动态（事件发射器）数据流。更多有关响应式编程的描述可以参考Reactive programming
响应式编程迈出的第一步是微软在.NET系统中创建了响应式扩展库（Rx）。在微软创建Rx之后，RxJava在JVM上实现了响应式编程。随着时间推移，经过Reactive Streams的不断努力制定了Java实现响应式编程的规范，规范为JVM上的响应式库定义了一组接口和交互规则。Java 9的Flow类已经实现了规范定义的接口（从Java 9 开始，java开始默认支持响应式编程，有条件的小伙伴该考虑升级Java版本了）。
在面向对象的编程语言中，响应式编程通常作为观察者模式的一种扩展。如果对比迭代器设计模式和主流的响应式流模式对比，会发现在几乎所有的库中Iterable-Iterator 都有双重性（可以互相转换）。两者主要的区别是：迭代器设计模式基于拉，响应式流基于推。
迭代器是命令式编程模式，尽管访问数据的方法仅由Iterable负责。实际上在使用迭代器时由开发者决定何时选择序列中的next()元素。在响应式流中，和上面Iterable-Iterator对应的是Publisher-Subscriber，新值出现时Publisher 会通知Subscriber ，推送是响应的关键。同样，对推送值的操作是声明式而不是命令式，代码表达计算的逻辑，而不是描述其精确的控制流。
响应式流除了推送值之外，同样以良好的方式定义了错误处理和操作完成。一个Publisher可以向其Subscriber推送新的值，也可以发送错误信号或者完成信号。错误信号和完成信号都会终止序列，下面的表达式准确简练的描述了这个逻辑：
onNext x 0..N [onError | onComplete]
这种模式非常的灵活，可以支持没有值，一个值或n个值（包括无限序列，比如时间）。但是为什么我们首先需要这样一个异步响应式库呢?
阻塞是一种浪费
现在应用有大量的并发用户，尽管现代化硬件的能力在不断提升，但是软件性能依然是一个关键问题。有两种方法可以提高软件的性能：
- 并行使用更多的线程和更多的硬件资源，
- 提高现有资源的使用率。
通常，Java 开发者使用阻塞代码开发程序，这种方法在没有性能瓶颈之前非常的完美，因为阻塞代码更容易理解也更容易编写。当程序出现性能瓶颈时，引入另外的线程来运行相同的阻塞代码（活多了需要加人）。但是，资源的这种扩展会迅速引入竞争和并发问题。更糟糕的是，阻塞会浪费资源。如果程序遇到延迟（特别是I/O操作，比如数据库请求或网络调用），因为线程需要等待数据而处于空闲状态进而导致资源的浪费。
因此，并行不是灵丹妙药。充分使用硬件的能力十分必要，但是推理过程十分复杂而且更加容易造成浪费。
异步是一副良药吗？
通过编写异步、非阻塞代码，可以切换到另一个活动的使用相同基础资源的任务并在异步处理完成后返回到当前的任务。通过异步代码可以提高资源的使用率，减少资源浪费。
Java提供了下面两种异步编程模型：
-
Callbacks： 异步方法没有返回值，但是需要一个额外的
callback参数（可以是lambda表达式或匿名类），当结果可用时会被调用。 -
Futures:：方法立即返回一个
Future<T>。异步计算T的值，但是Future封装了对T值的访问。T值可能不是立即可用，而且Future对象支持轮询直到值T可用。Java的ExecutorService执行Callable<T>时返回一个Future对象。
Java提供的这两种编写异步代码的技术足够好了吗？这些技术很好，但并不适用于每一种场景，而且都有各自的局限性。Callbacks的缺点是很难组合在一起使用，而且多个回调组合在一起使用时，代码很快就会变的难以阅读和维护（通常称为回调地狱）。
下面以在用户界面显示用户前五个收藏夹为样例说明Callbacks的局限性。业务场景为：显时用户前五个收藏夹，如果用户没有收藏夹则显示建议。
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的实现有很多的代码，这些代码难以理解，要想一步一步弄懂逻辑也比较困难，而且代码还有部分重复。下面是基于Reactor的等价实现：
userService.getFavorites(userId)
.flatMap(favoriteService::getDetails)
.switchIfEmpty(suggestionService.getSuggestions())
.take(5)
.publishOn(UiUtils.uiThreadScheduler())
.subscribe(uiList::show, UiUtils::errorPopup);
基于callback的代码实现如果要增加超时逻辑会十分的困难，但是基于Reactor的实现只要使用timeout方法即可轻松完成:
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对象比回调好一点，但它们在组合方面仍然做得不好，尽管CompletableFuture在Java 8中带来了改进。编排多个Future 对象可行但是并不容易。除此之外，Future 还有其他问题：
- 调用
get()方法很容易导致Future对象阻塞， - 不支持懒加载/懒计算，
- 它们缺乏对多值和高级错误处理的支持。
考虑另一个例子：我们获得一个id列表，我们想从中获取一个名称和一个统计信息，并将它们成对组合，所有这些都是异步的.。下面的代码使用一个 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",
"Name NameABSLAJNFOAJNFOANFANSF has stats 121");
由于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"
);
使用回调和Future对象的危险是相似的，这也是响应式编程通过Publisher-Subscriber对解决的问题。
3.3. 从命令式编程到响应式编程
诸如Reactor的响应式编程库旨在解决JVM上“经典”异步方法的缺点，同时也着重对以下方面进行改进：
- 可组合性 和 可读性，
- 把数据当做流处理，同时提供丰富的操作方法，
- 在订阅之前不会发生任何事情，
- 背压 或 消费者向生产者发送信号通知数据生产速率过高或过低的能力，
- 对并发不可知更高价值和更高级的抽象。
3.3.1. 可组合性和可读性
“可组合性”指的是编排多个异步任务的能力，可以将前面任务的结果作为后续任务的输入。当然也可以以fork-join的方式运行多个任务。此外，我们可以在更高级的系统中把异步任务作为离散组件重用。
编排任务的能力与代码的可读性和可维护性紧密相关。随着异步处理层的数量和复杂性的增加，编写和阅读代码变得越来越困难。正如我们所见，callback模型十分简单，但是callback一个缺点就是处理变的复杂，一个callback需要在另外一个callback中执行，这样一层一层的嵌套。这就是“回调地狱”，这种代码难以阅读和分析逻辑。
Reactor提供了丰富的组合操作，代码可以反应对处理过程抽象的组织，一切尽量保持在同一层（尽量减少嵌套，这也是和callback模式相比最大的改进之一）。
3.3.2. 类比工厂的生产线
数据在响应式程序中的处理过程，可以被看作是数据在组装流水线中移动。Reactor既是传送带又是工作站。原材料从来源（第一个Publisher）倾泻而出（中间经过多道工序加工），最终成为可以推送给消费者（Subscriber）的成品。
原材料可以经过各种转换和其他中间步骤，也可以成为将中间零件组装在一起的更大装配线的一部分。如果某一点出现故障或堵塞（某到工序耗时长），那么出问题的工作站可以向上游发出信号，以限制原材料的流动（有问题及时向上游反馈，上游做出响应，避免进一步恶化）。
3.3.3. Operators
在Reactor中，operator就是流水线中的工作站。每个operator都会将行为添加到Publisher 中，并将上一步的Publisher 包装到新实例中。这样构建了一个完整的链接，数据从第一个Publisher 向下游移动并由每一个链接进行转换，最后，由一个Subscriber 结束数据的数据处理过程。在Subscriber 订阅之前数据不会被处理也不会向下游移动。
尽管响应式流规范根本没有定义operator，但是像Reactor的响应式库提供的最佳附加值之一就是提供了丰富的operator，从简单的转换和过滤到复杂的编排和错误处理，这些内容涉及很多的领域。
3.3.4. 不 subscribe()不会发生任何事情
在Reactor中，当你编写了一个Publisher 链，默认数据不会注入。实际上只是创建了一个异步处理过程的抽象描述（这有助于重用和组合）。通过subscribing 动作，将Publisher 绑定到Subscriber ，这会触发数据在整个链路中移动。内部实现是通过Subscriber 发送Request 信号，信号被传播到上游一直到Publisher 。request 也是实现背压的关键方法。
3.3.5. 背压
向上游传播信号通常用来实现背压，在和流水线的类比中描述为当工作站处理比上游工作站的处理速度慢时，沿着流水线向上游反馈。背压其实就是下游向上游发送信号，并影响上游数据处理的一种机制。
响应式流规范定义的实际机制可以简单的概括为：一个subscriber可以以“无界” 模式工作，并让数据源以最快的速度推送所有数据，或者使用request 机制向数据源发送信息，向数据源反馈已经准备好处理n个元素。
中间operator可以在中途改变request。想象一下一个buffer 以十个元素为一组将元素进行分组。如果subscriber请求一个buffer，数据源发送十个元素是可以被接受。一些operator也实现了预拉取策略 ，这避免了request(1) 不断往返。如果在请求之前生成元素的成本很低，这种操作就非常的有帮助，可以显著的提高处理效率。
这会将推模式转换为推拉混合模式，如果上游已经准备了数据，下游则可以从上游获取n个元素。但是如果数据还没有准备好，那么当有数据时上游就会将数据推送到下游。
3.3.6. Hot vs Cold
Rx响应式库家族将响应序列分为两大类：“热”和“冷”。这种区别主要与响应式流对subscriber的响应有关：
- 对于每一个
Subscriber，包括在数据源位置，冷序列都会重新开始。例如，如果源包装了HTTP调用，则将为每个subscription发出一个新的HTTP请求。 - 对于每一个
Subscriber，热序列并非都会从头开始。相反，后面的subscriber只能收到订阅完成之后产生的数据。但是一些热响应式流可以缓存或者对历史数据全部或部分重放，也就是说迟来的subscriber可以收到在完成订阅动作之前的数据。从一般的角度来看，即使没有订阅者在订阅数据，热序列甚至会发出数据（“订阅之前什么也没有发生”规则的例外）。
