我谨代表 Reactor 团队，很高兴宣布最新的 Reactor 里程碑 Californium-M1 ? ?

团队一直忙于 Californium，这是 Reactor 3 的第三个主要版本。我们现在已准备好听取您对一些特定问题的反馈，并且我们还准备了大量增强功能和错误修复供您使用。

Californium-M1 BOM

对于其第三个发布列车，我们继续沿用元素周期表中按字母顺序递增的名称主题。Californium 是第一个在加利福尼亚合成的元素。

该里程碑的 BOM 包含

• reactor-core 3.2.0.M3
• reactor-extra 3.2.0.M1（带有一些 API 对齐更改）
• reactor-netty 0.8.0.M1

今年早些时候 (M1)，reactor-core 发布了一个早期预览版，该版本只专注于“错误模式继续”功能，并且核心库在 6 月份还发布了一个非定期里程碑版本 (M2)。本博客文章除了涵盖全新的 M3 中的变更，还包括了 M2 中的变更。

Reactor Netty 0.8.0.M1

这里的重要更新是 reactor-netty。敬请期待一篇更详细的博客文章，它将详细介绍 API 变更和新功能背后的原因，其中包括：

API 重构

团队对 API 进行了大规模重构，使其在构建客户端和服务器时更具指导性，避免了 0.7.x 版本中容易出现的难以调试的配置错误。

新 API 还更清晰地勾勒出了生命周期。

HTTP2 支持

是的，HTTP2 支持！目前主要是透明地升级到 HTTP2，但我们正在努力在不久的将来将 HTTP2 独立流作为一等公民添加进来。

Reactor Core 3.2.0

M2 和 M3 总共带来了 70 多项变更，相较于之前的 Bismuth 版本。

reactor-core 的 API 变更少于 reactor-netty，更新考虑主要在于里程碑版本之间的差异。有关更多详细信息，请参阅 M2 和 M3 的“更新考虑”部分。

我们最需要您对以下功能的反馈：

指标

已添加 Micrometer 和 .metrics() 支持 (#1183, #1123)。新的 .metrics() 操作符仅在 Micrometer 存在于类路径上时才执行操作。

它会记录 onNext 计时、订阅到完成计时、信号计数等指标——所有这些都从上游操作符产生的信号的角度来看。

它在 M2 中引入，但在 M3 中得到了改进，并且一些标签名称也发生了（破坏性）更改 (#1245)。

请注意，一个重要的目标是避免在公共 Reactor API 中暴露 Micrometer 内容。我们不希望强制依赖 Micrometer，我们努力将其使用限制在仅当我们检测到它在类路径上时才加载的内部类中。

接下来：在 GA 之前，还应该有对 Schedulers（或者说支持某些 Schedulers 的 ExecutorServices）的基本检测支持。我们还在研究一种为 Reactor 全局选择特定 MeterRegistry 的方法，同样不暴露引用 MeterRegistry 接口的公共 API。

高级重试

我们添加了一个预配置的 retryWhen 替代方案，带有指数退避和抖动 (retryBackoff())。请参阅 #1122。

此版本的重试反映了我们认为在重试方面的行业最佳实践。它是过于简单的 retry(n)、复杂的 retryWhen(Function) 和 reactor-addons 中更可配置的 RetryFunction 之间的一个良好折衷。

基于资源的响应式闭包

为了帮助您构建响应式事务块，我们添加了 usingWhen。与 using 类似，它包装一个资源，从中生成一个 Flux，并确保在 Flux 终止时正确清理资源。

主要区别在于：

• 资源通过 Publisher 异步提供。
• 清理也是异步的 (Function<Resource, Publisher>)，并且只延迟终端信号的传播，而不延迟 onNext 信号的传播。
• 操作符可以为完成、错误和取消终止拥有单独的异步“清理”。

这在 M2 中引入，但在 M3 中进行了微小更改，以修复 Context 传播并支持 Publisher<Resource> 的取消。通过在此操作符返回的主 Flux<T> 在 Resource 发出之前取消，您的取消指令将传播到 Publisher<Resource>。

onDiscard 钩子

这个全局钩子，以 Consumer<Object> 的形式出现，旨在成为处理需要特殊清理的堆外对象的高级用户所需的最后一块拼图。

通常，Netty 的 ByteBuf 或 Spring 5 的 DataBuffer 属于此类别：它们是池化的、堆外的，并且一旦不再使用就需要调用 release()，否则会导致内存泄漏。

此类元素在响应式序列中可能被遗漏，并且在三种主要情况下永远不会到达用户代码：

1. 操作符的源格式不正确，不符合 RS 规范（例如，它在发出 onComplete 之后才发出 ByteBuf）。
2. 操作符根据其语义过滤某些元素（例如 filter）。
3. 操作符出于背压优化目的进行预取，但被取消，从而丢弃其预取队列。

情况 (1) 已由 onNextDropped 钩子涵盖，但情况 (3) 绝对没有。情况 (2)（过滤语义）则介于两者之间，例如，可以在过滤 Predicate 内部进行清理。但这很麻烦，而且容易被遗忘。

因此，我们在 Hooks 的功能集中添加了 onDiscard 来涵盖 (2) 和 (3)。请注意，与“出错时继续”功能不同，目前没有公共 API 可以为特定的 Flux 实例设置钩子。有一个使用 Context 的不受支持的变通方法，官方 API 很可能会在 GA 或后续维护版本中出现。

onDiscard 钩子具有以下特性和要求：

• 它是附加的，这意味着调用 Hooks.onDiscard(Consumer) 两次将使用 Consumer#andThen 组合这两个消费者。
• 它不是键控的，这意味着虽然多次调用是附加的，但它只能完全重置（不能按每个 Consumer 进行重置）。
• Consumer 在类型转换之前必须执行 instanceof 检查，因为它将用于不同类型的对象。
• Consumer 决不能抛出异常，并且应根据需要包含 try catch 块。
• Consumer 必须是幂等的，因为它可能会在同一实例上多次调用（例如，在缓冲区重叠的情况下）。

附带说明一下，errorStrategyContinue() 在 M3 中已重命名为 onErrorContinue()。

Reactor Extra 3.2.0.M1

最后，reactor-extra 在 retry/repeat 工具方面有较小的 API 更改。它与 core 操作符对齐，使用相同的默认值和 Long 而不是 Integer 索引。

下一步

reactor-core 的下一步是重构 Processor 对象的暴露方式。当前的 FluxProcessor<IN, OUT> 有点臃肿，因为它扩展并暴露了整个 Flux API。

此外，FluxProcessor#sink() 和相关的 FluxSink 太容易被误用，特别是当人们既想将 Processor 订阅到 Publisher 源，又想通过 sink() 手动推送数据时，这目前并不真正受支持。sink() 应该只调用一次，并且返回的 FluxSink<T> 实例应该重用，这一点也不够清晰。

因此，我们正在研究一个 Processor<T, T> 的门面，它直接实现 FluxSink（而不是 Flux），在同时用作订阅者和接收器时工作，并且有一个 asFlux() 视图方法来可选地在其之上构建 Flux 操作符链。

MonoProcessor 很可能会遵循这些步骤，成为一个（更简单的）接口，具体实现将重命名为 MonoNextProcessor。我们还在考虑提供一个独立的 MonoSink 实现，用户可以直接操作，而无需使用 Mono.create()。

结论

酷的人不会等待 GA 发布！快去试驾那个闪亮的里程碑版本吧，一如既往，欢迎反馈。:)

祝您反应式编程愉快！