由于错误隔离是 Reactive 协定中一个很重要的部分，Stream API 已经实现了容错管道和相关的服务调用。 错误隔离可以防止 onNext、onSubscribe 和 onComplete 回调函数弹出任何异常。作为替代，这些异常被传递给onError 回调函数，并传播给下游。一些 Action 可以积极或消极的对待这些信号，类如 when()仅仅观察错误，而onErrorResumeNext()则可以切换至备用发布者。
最后一步，我们将探访 UserService.allFriends 查询，它将从数据中获取整个数据集。 表 17，进化成响应型微服务，第三部分，UserService.allFriends 的背压结果是的，这很啰嗦。…但现在，我们将查询的结果依次流式处理（可能已经通过 SQL 限定分页）。Streams.createWith 是一个 PublisherFactory，它将截断请求，执行启动或停止操作。
第二步，我们将思维扩大到消费方面。在过渡阶段，请牢记我们的 Stream 可以使用运算符来阻塞 。 我们有两个问题亟需处理：一个是鲁棒性（网络分裂问题等），一个是如何避免在新服务执行前等待前一服务：表 16，进化成响应型微服务，第二部分，RickAndMortyService 中的并行请求和永续性结果Streams.merge() 将两个查询合并，是一个非阻塞的协调操作。
首先要做的就是隔离微服务的访问。现在，我们将不再返回 T 或 Future，取而代之的将是 Publisher 以及特定的 Stream 或 Promise。立竿见影的好处就是我们可以不必费心考虑错误处理或是线程（仅是现在)了：错误将在 onError 调用（非冒泡）中传递，而线程可以在后续过程中使用类似 dispatchOn 的方法进行调整。另一个好处就是它能让我们的代码更加实用。
微服务 这个概念近年来愈发流行了起来。简单来说，我们编写软件组件的一个首要目标就是鼓励独立性、缩放适应性以及重用性。事实上，我们使用微服务已经超过 30 年了。Unix 下一个微服务的使用示例history | grep password甚至在应用程序内部，我们也可以找到类似概念的功能粒度。命令式 Java 代码中微服务的示例User rick = userService.get("Rick");
说明简介Reactor 是一个基于 JVM 之上的异步应用基础库。为 Java 、Groovy 和其他 JVM 语言提供了构建基于事件和数据驱动应用的抽象库。Reactor 性能相当高，在最新的硬件平台上，使用无堵塞分发器每秒钟可处理 1500 万事件。
为了协调数据的并行序列，我们可以组合发布者。由于生成序列是合并的结果，它们也可以用于数据的[异步转换]异步转换。 通过非阻塞的协调方式可以避免开发者使用 Future.get() 或 Promise.await()，这两个方法在多信号存在是容易引发问题。非阻塞意味着管道除了订阅者的需求，不会做任何等待。订阅者的请求将被切分至最小，然后分配给已经组合的发布者。
多数情况下，依照 Reactor Stream 的协定，背压可以被自动处理。如果订阅者（Subscriber）请求的数据并没有超过其处理能力（例如类似 Long.MAX_VALUE 的东西），数据源上游可以避免发送过多数据。
将数据 T 按照序列分发给 Stream<T> 的主要目的有三： 将数据 T 的一个序列暴露给一系列有限且分组的观察和统计：取和计算、平均值计算或灵活的聚合（Map、Tuple……）。将分组数列同 dispatchOn 结合，并为生成的每个 Stream 进行并行化处理。对每个独立的分组序列重复 onComplete() 调用，例如，在异步 IO 模块中界定一次数据冲洗。如果是同聚合所有的 Stream.
将数据 T 按照序列分组为列表 List 的主要目的有二： 将匹配分界条件的序列暴露给一个 JVM API 常用的Iterable 结构体。减少 onNext(T) 的信号量，类如 buffer(5) 会将一个有10元素的序列转换成2个列表（每个列表有5个元素）。收集数据将会产生内存甚或 CPU 的开销，应当适当的调整大小。建议使用小巧且定时的分界，以避免任何长时间的聚合。
搾取系统所有的的CPU 和 内存，减少使用过度导致的延迟！ 克林贡谚语 在读过一俩遍 101 种数据流崩溃说明 之后，你的黑客之心蠢蠢欲动，已经想尽快实验一下了。事实上，有效分派并非 处理中每秒百万级数据待办列表 中的唯一事项，这还远远不够。在分布式系统中，独立部分及缓冲 IO 写入的延迟成本依然是一个常见问题。
有赖于信号回调，Reactive Streams 和 Reactive Extensions 通常目标就是不干预线程行为。 Stream 整个就只关心它将会在某个从现在到某个时间 T 之间执行。 非并行的信号可能也会禁止 Subscriber 的无共享(share-nothing)并发访问, 不过信号和请求可以在两个非对称的线程上运行。Stream 默认被分配了一个SynchronousDispatcher 并将通过 Stream.
Reactor 提供了基于 Reactive Streams 标准的 Stream 或者 Promise 来组成静态类型的数据管道。它是一个非常实用且灵活的组件。它在被用于像 RxJava 的 Observable 那样仅仅只对异步操作进行组合时是足够灵活的。而它也足够强大，可以像一个异步工作队列那样运行，取出或者加入任意的组件，或者来自于其它标准实现之一的 Reactive Streams 组件。[3]。
图 9. 程序员狗如何使用 Reactor-StreamReactor Streams 拥有下面这些功能部件:Stream 及其直接实现。包含 reactive extensions 和其它一些 API 组件。带有一套特殊 A+ 风格 API 的 Promise。可以使用 Promise.stream() 再转换回到 Stream。静态工厂，可以一站式的创建出相关的组件。
注意，你应该再也别去使用 Future.get() 了。— Stephane Maldini与一个银行业的客户 首先来看看一个 Java 8 示例中流 (Stream) 的运作方式import static reactor.Environment.*;import reactor.rx.Streams;import reactor.rx.BiStreams;//...Environment.initialize()//找到一个 String 列表中开头的 10 个词Streams.from(aListOfString) .
字节操作是许多数据管线配置中用到的核心概念。从 reactor-net 到通过 IO 接收发送编组和解组字节中都得到广泛运用。reactor.io.buffer.Buffer 是 Java ByteBuffer 操作的修饰符，提供了一系列操作，目的是通过调整字节缓冲区大小以及读取或覆盖预分配字节来最小化字节拷贝。在字节缓冲区中追踪定位可以让开发者快速进入脑痛期，至少对我们是这样。
基于环形缓冲区的响应式数据流处理者有一些很棒的特性：高吞吐量重现最新未消费数据如果没有订阅者在监听，数据不会丢失（不同于Reactor-数据流的广播者）。
核心处理者的工作比调度者更加专一：计算支持背压的异步任务。同时它直接实现了 org.reactivestreams.Processor 接口，可以良好地与其它响应式数据流提供方合作。比如同时作为订阅者和发布者的处理者。你可以将它插入到响应式数据流链中你想要的地方(源，处理过程，槽)。! 规范并不明确推荐直接使用 Processor.onNext(d)。
关注时代Java
