进入响应流时代

Reactor 2.0 的开发始于 2014 年底，与 Reactive Streams 大致同期。我们热衷于加入这项工作并早期采用背压协议来缓解我们主要的消​​息传递限制：有界容量。我们在 Reactor 2.0 中首次尝试实现基于环形缓冲区的调度器的 Reactive Streams 实现，并衍生出一种日益流行的响应式模式：Reactive Extensions。

与此同时，Reactive Streams 开始受到关注，整个库生态系统都在讨论这一转变。常见的担忧？实现 Reactive Streams 语义绝非易事。我们观察到对响应式基础的需求日益增加，以解决消息传递并实现常见的流式运算符。因此，我们为 Reactor Core 创建了一个专门的项目空间，并与 Spring Framework 团队开始了一项重点工作。

从 2.5 开始，Reactor 现在被组织成多个项目，2.0.x 等维护分支保持不变。这反映在发布管理中，例如Reactor Core 2.5 M1是唯一可用的里程碑，其他项目将随其独有的版本发布。

为了支持这种新的项目模型，我们在http://projectreactor.io 上部署了一个新的、希望更受欢迎的网站。

响应流的协作式新方法

这种新的组织结构大大降低了参与项目活动的成本。该项目受益于 Spring API 设计协作以及来自 Sébastien Deleuze 和 Brian Clozel 的直接贡献。Reactor 还欢迎新的外部贡献者和评审员的帮助。

• Alex Petrov 正在重新发明流行的 Reactor 事件路由功能。
• Anatoly Kadyshev 致力于为 Reactor IO 搭建令人惊叹的高效 Aeron Reactive Streams 桥接。
• Ben Hale 和他的团队正在开发新的 Reactive Cloud Foundry 客户端 API。Ben 不仅早期采用了 Reactor 2.5 Core 和 Stream，还不断与我们一起迭代他的实际用例。
• Damien Vitrac 贡献了新的项目网站样式，并正在准备 Reactor Console 用户体验。
• Dávid Karnok，研究员，RxJava 的主要活跃提交者，以及出色的 Advanced RxJava 博客的作者，极大地影响了我们的内部操作模型。我们对效率和响应式模式的共同热情促使我们创建了一个研究空间，Reactive Streams Commons。

Reactor 2.5 中的依赖和协作

Reactive Streams Commons

Reactive Streams Commons 存储库是一项开放研究工作，专注于 Reactive Extensions 及更多领域的效率，以实现 Reactive Streams 规范。它完全由 Reactor Core 和 Stream 内联，它们作为这项工作所关注的许多变革的契约门户。

"RSC" 因此是一个类似于 JCTools 处理并发队列的自由形式项目。它最大的进展之一是“融合”协议，以减少响应式处理链中大多数同步和某些异步阶段的开销。最后，这项工作帮助修复了上百个流式错误，我们的测试过程现在包括 RSC 单元/集成测试和 JMH 基准测试，并结合 Reactor 自己的集成测试和 基准测试。

Reactor Core 2.5.0.M1

今天的 Reactor 博客系列以一个令人愉快的事件开始：Reactor Core 2.5.0.M1 发布！在其新的范围和与 Reactive Streams Commons 的紧密联系下，Reactor Core 提供了足够的 Rx 覆盖范围来构建响应式应用程序或库，例如 Spring Reactive Web 支持。对于不耐烦的读者，请查看 GitHub 上已有的快速入门。

快速浏览一下分散-聚集场景

Mono.from(userRequestPublisher)
    .then(userRepository::findUserProfile, 
          userRepository::findUserPaymentMethod)
    .log("user.requests")
    .or(Mono.delay(5)
            .then(n -> Mono.error(new TimeoutException()))
    .mergeWith(userRepository::findSimilarUserDetails)
    .map(userDetailsTuple -> userDetailsTuple.t1.username)
    .publishOn(SchedulerGroup.io())
    .subscribe(responseSubscriber);

详情

• Flux，一个具有精简 Rx 范围的 Publisher，用于发布 0 到 N 个数据信号。操作符包括 create()、interval()、merge()、zip()、concat()、switchOnError() 和 switchOnEmpty()。

• Mono，一个具有精简 Rx 派生范围的 Publisher，适用于严格类型化这种特定数量性质的 0 或 1 个数据信号。操作符包括 delay()、then()、any()、and()、or()、otherwise()、otherwiseIfEmpty()、where() 和阻塞的 get()。

• 基于纯 Java 接口（Runnable、Callable）的新简单调度契约。

-- 包含 SchedulerGroup、TopicProcessor 和 WorkQueueProcessor。 -- 取代了之前的 Enviroment/Dispatcher 组合，同时解决了相同的需求，并且很快将记录简单的迁移路径。不再有静态状态持有调度器的引用。 -- 链接操作符：publishOn() 和 dispatchOn()

• 使用 TestSubscriber 支持 Publisher 源的测试。
• 将 Callable、Runnable、Iterable、Java 8 CompletableFuture、Java 9 Flow.Publisher、RxJava 1 Observable 和 Single 转换为支持 Reactive Streams 的 Flux 和 Mono，无需额外的桥接依赖。
• 全面修订并集成的 Javadoc，包括略微调整的 marble 图。
• 一个微型工具包，包含实用工具和基础 Subscriber，可随意重用以实现您自己的响应式组件。

-- 一种经济高效的定时器 API 和实现（哈希轮定时器）。 -- 新的融合 API，用于虚拟地合并响应式链中的两个或更多阶段。 -- 一个经过调整的 QueueSupplier，将为正确的容量提供正确的队列。

• 基于状态和流表示的新内省 API。

-- Publisher 日志记录，如果可用，则回退到 java.util.logging 或 SLF4J。可直接在 Flux 和 Mono 上使用 log() 操作符。 -- 与包括 Reactive Streams 在内的任何其他契约正交，一切都可以是 Backpressurable、Completable 或作为 Receiver 生成泛型 Object（可能是 Subscriber），这反过来又允许我们追踪流的完整图并使用状态指示器对其进行增强：

下一步是什么？

我们希望收集您的反馈意见，您可以访问相应的 issues 存储库或加入我们最近创建的 Gitter 频道。敬请关注下一篇关于 Reactor Stream 2.5.0.M1 的文章，它是 Reactive Streams 的完整 Rx 实现。