Reactive Streams 规范

• 全局错误钩子（Global Hooks）的注册机制与适用场景
• onErrorDropped 与 onOperatorError 的触发时机差异
• 生产环境监控与错误兜底的最佳实践

Reactor 提供了两类全局错误钩子用于兜底处理无法被正常传递的错误。onOperatorError 在操作符执行过程中出现异常、且该异常无法沿正常 onError 路径传递时被调用，例如在 map 等操作符内部抛出的异常已经通过 onError 传递，但某些操作符（如 onNext 内部调用再次抛错）的场景会回调它，通常用于记录日志、附加上下文或做统一采样。onErrorDropped 则发生在某个错误无法被任何下游订阅者接收时，最典型的场景是下游因 cancel 已经取消，而上游仍然推送了一个 onError 信号，此时该错误会被"丢弃"并交给该钩子处理；此外在 onNext 发生时下游已取消也会触发 onErrorDropped。两者都属于"最后防线"性质的兜底机制，不应被用来替代正常的操作符级错误处理，更多是用于可观测性，例如在日志或 APM 中标记"被丢弃的错误"。

理解两者的关键在于"错误能否被正常传递"。onErrorDropped 处理的是"想传但没人接收"的错误（通常是下游已取消），onOperatorError 处理的是"操作符执行过程中产生的、需要被包装或记录"的错误。生产环境应始终注册这两个钩子，避免异常被静默吞掉导致排查困难。

• BaseSubscriber 的 toString/hookOnSubscribe/hookOnNext 等钩子方法
• 手动 request(n) 与背压的来源控制
• 与默认订阅者的差异（不自动请求）

BaseSubscriber 是 Reactor 提供给自定义订阅者的抽象基类，它实现了串行化（serialized）的 Subscriber 接口，并提供了多个可覆写的钩子方法，如 hookOnSubscribe、hookOnNext、hookOnError、hookOnComplete、hookOnCancel 等。与默认订阅者自动请求所有元素不同，BaseSubscriber 默认不会请求任何元素，开发者必须在 hookOnSubscribe 中调用 subscription.request(n) 主动拉取数据，从而完全掌控背压节奏。典型用法是"批量拉取"模式：每次 hookOnNext 处理完一个元素后再请求下一个，或按固定窗口请求，从而把上游的推送转化为受控的拉取，避免内存被无界填满。

BaseSubscriber 的核心价值是让背压控制从"隐式"变为"显式"。它提供受保护字段 subscription 和 request() 方法，配合 hook 方法实现可读性强的自定义订阅逻辑。相比匿名实现 Subscriber，它能避免重复代码并保证串行化安全。

• 冷流（每次订阅独立执行、每个订阅者获得完整序列）与热流（订阅者共享同一数据源）的语义
• ConnectableFlux 与 connect()/autoConnect() 的实现
• share()/refCount() 的自动连接与引用计数

冷流（Cold Publisher）在每次订阅时都会重新生成数据，每个订阅者都获得一份独立的完整序列，例如 Flux.just 或 Flux.fromIterable 每次订阅都会重新产生数据。热流（Hot Publisher）则是一个共享的数据源，无论多少订阅者订阅，它们都接收到同一份实时数据，例如 Sinks.Many 或 ConnectableFlux。在 Reactor 中，通过 ConnectableFlux 可以把冷流变成热流：cold.publish() 返回 ConnectableFlux，只有调用 connect() 后数据才开始流动，订阅者只能接收到订阅之后的数据（可配合 replay 缓存）。share() 是 publish().refCount(1) 的便捷封装，它在第一个订阅者到来时自动 connect，在最后一个订阅者取消时自动断开，从而简化了手动连接管理。

热流与冷流的本质区别在于"数据是否与订阅者绑定"。冷流每订阅一次就执行一次数据生成，热流数据独立于订阅者产生。选择时需要权衡：冷流适合每次订阅都需要独立计算结果的场景，热流适合广播、事件推送等一对多共享场景。connect 控制数据开始流动的时机，refCount 控制生命周期。

• 四种背压策略的语义与数据丢弃行为
• 有界/无界缓冲策略的内存风险
• 生产场景下的选择依据

背压策略决定了下游处理速度慢于上游时，上游产生的多余元素如何处理。onBackpressureBuffer 将多余元素缓冲起来，默认无界，可能造成内存溢出，可配置有界缓冲及溢出策略；onBackpressureLatest 只保留最新元素，丢弃除最新外的历史元素，适合"只关心最新状态"的场景；onBackpressureDrop 把无法及时处理的元素直接丢弃，适合可容忍丢失、只接受部分数据的场景；onBackpressureError 在缓冲溢出时发送 onError 信号终止流，适合数据丢失不可接受、宁可报错也不丢数据的场景。生产取舍上，若下游可以变慢且必须不丢数据，用有界 Buffer 并配合监控；若数据是"最新值优先"（如状态同步），用 Latest；若允许降采样，用 Drop；若必须严格完整，则用 Error 并及时告警。

选择背压策略的本质是"在内存与数据完整性之间做权衡"。缓冲策略牺牲内存换取完整性，丢弃策略牺牲完整性换取内存和延迟。实际生产应避免无界缓冲，"完全消化"与"完全不丢"往往不可兼得，需要结合业务容忍度决定。

• Mono.defer 的延迟求值语义（每次订阅重新执行 Supplier）
• Mono.cache 的缓存语义与并发重复订阅
• Global 缓存与请求作用域隔离的冲突

Mono.defer 允许在订阅时再执行 Supplier 来创建 Mono，从而确保每次订阅重新计算，避免把"创建时"的上下文（如请求 ID、线程局部变量）固化到订阅结果中。Mono.cache 则会把第一次订阅的结果缓存起来，后续订阅直接复用缓存结果。在请求作用域中，如果使用了全局的静态缓存（如静态的 Mono.cache 实例），不同请求会共享同一个缓存结果，导致请求 A 的数据被请求 B 复用，造成数据错乱——这就是"缓存泄露"。正确做法是：缓存对象必须按请求作用域创建，或使用只缓存"无状态、全局可复用"的数据；对于依赖请求上下文的 Mono，应使用 defer 保证每次订阅重新求值，或者使用 cache 但确保缓存实例生命周期与请求一致（如存放在请求作用域对象中）。

核心矛盾在于"缓存实例的生命周期"与"请求作用域"不一致。cache 是跨订阅共享的，若缓存实例被全局持有，则跨越了请求边界。defer 则把求值推迟到订阅发生，天然与订阅上下文绑定。因此生命周期敏感时优先 defer，或让 cache 实例随请求创建销毁。

• boundedElastic 的线程池与队列上限（默认并发数与队列容量）
• 根据任务动态创建/回收线程的机制
• 与阻塞式 IO 适配的定位

Schedulers.boundedElastic() 提供一个有界的弹性线程池，用于执行阻塞式工作（如 JDBC 调用、阻塞 IO、同步外部服务调用）。它默认最多创建 10 倍 CPU 核数的线程（默认上限 10 * CPU 核数，且有一个最小线程数），并有一个有界任务队列（默认 100000 的队列容量）。当线程数达到上限后，新任务会进入队列排队，队列满则拒绝。与固定线程池不同，boundedElastic 会根据任务负载动态创建线程，空闲线程会超时回收，从而在"足够弹性"与"防止资源耗尽"之间取得平衡。它是 Reactor 官方推荐用于包裹阻塞操作的调度器，因为阻塞任务不应占用 Netty EventLoop 等响应式线程。

boundedElastic 的名称体现了"有界 + 弹性"两个特性：弹性保证能适应短期阻塞任务激增，有界保证不会无限创建线程导致资源耗尽。它替代了旧版 elastic()（无界、易耗尽），是生产环境处理阻塞调用的首选。

• Mono 0..1 元素（最多一个）与 Flux 0..N 元素（任意多个）的语义
• 两者的转换（flatMapMany、next、single 等）
• 返回值类型对 API 设计的影响

Mono 表示一个最多产生 0 或 1 个元素的响应式流，对应"单值结果"或"空结果"的语义，通常用于单个查询结果、单个 HTTP 响应、单个写操作等场景。Flux 表示 0 到 N 个元素的响应式流，对应"集合、流、批次"的语义，用于列表查询、流式数据处理等。两者都继承自 Publisher，且是异步的。Mono 与 Flux 之间可以互相转换：Mono.flatMapMany 将 Mono 展开为 Flux，Mono.next() 将 Flux 的第一个元素变为 Mono，Flux.collectList() 将 Flux 聚合为 Mono。在设计 API 时，返回 Mono 表示"预期最多一个结果"，返回 Flux 表示"可能多个结果"，类型本身即是一种契约。

Mono 与 Flux 的区别不仅是"数量"，更是"语义契约"。Mono 常用于链式组合（then、flatMap、zipWith），因为单值易于组合；Flux 则适合流式处理。合理选择类型能提升代码可读性和类型安全。

• Processor 同时作为 Publisher 与 Subscriber 的双重身份
• 背压传递与泄漏（subscription 未正确传递、cancel 未处理）的风险
• 官方建议用 Sinks 替代自定义 Processor 的原因

Processor 同时实现 Publisher 与 Subscriber，既能从上游接收数据，又能向下游发布数据，相当于一个"可插拔"的中间节点。自定义 Processor 时最容易出现的问题：一是背压丢失，即下游的 request(n) 没有正确向上游传递，导致上游无限生产而下游无法消化，造成内存溢出；二是订阅/取消泄漏，即没有在上游与下游之间正确桥接 subscription 与 cancel 信号，导致流无法终止或资源未释放。正因为手动实现正确的背压桥接和串行化极其繁琐且易错，官方明确建议不要自定义 Processor，而是使用 Sinks（Sinks.Many、Sinks.One）与 Flux.create 等高层 API 来构建自定义发布源，它们内部已处理了背压、串行化和取消语义。

Processor 的问题在于它把"背压桥接"的全部复杂细节暴露给开发者，而背压契约要求严格遵循 request/cancel 的传递规则，稍有偏差就导致泄漏或溢出。Sinks 将这些细节封装起来，是更安全、更工程化的替代。

• 四个接口的定义与职责
• 背压与取消的契约（request(n)、cancel）
• 规范对信号顺序与串行化的要求

Reactive Streams 定义了四个核心接口：Publisher 负责发布数据流，通过 subscribe 方法建立订阅；Subscriber 负责消费数据，通过 onSubscribe、onNext、onError、onComplete 四个回调接收信号；Subscription 是订阅双方之间的连接，拥有 request(n)（请求 n 个元素）与 cancel()（取消订阅）两个方法，是背压的关键；Processor 同时实现 Publisher 与 Subscriber，用于在流中插入中间处理节点。规范契约要求：信号必须串行化（subscriber 的每个方法只能被串行调用，不能并发调用）；onNext 的数量必须不超过订阅者请求的总数（除非采用无界请求）；onError 与 onComplete 是终止信号，一旦发出后不能再调用 onNext；cancel 后订阅者不能再接收信号。这些契约保证了背压的正确性、内存安全与可组合性。

四要素的契约是响应式编程的根基。request(n) 相当于"我还能接收 n 个"，生产者在收到请求前不得发送元素，从而形成从上到下的背压信号传播。Processor 的存在使流可以组合成更复杂的流水线，但因其契约复杂，官方更推荐用操作符组合。

• request(n) 的信用（credit）机制
• 生产者按 request 数量控制 onNext 的发送节奏
• 无界 request 与溢出风险

背压的实现基于"信用（credit）"机制。订阅者通过 Subscription.request(n) 向生产者声明"我还能接收 n 个元素"，生产者必须确保向该订阅者发送的 onNext 总数不超过 n，除非请求的是 Long.MAX_VALUE（无界）。生产者维护一个"已请求但未发送"的计数，每发送一个 onNext 就递减，计数降为 0 时停止发送并等待下一次 request。这样，消费者通过控制 request 的节奏反向控制生产者的生产速度，从而避免数据在消费者端堆积溢出。Reactor 内部通过操作符间的 request 传递和异步缓冲（如 FluxPrefetch）实现这种节奏控制，即使生产者本身是同步的，也能通过缓冲与调度实现异步背压。

request(n) 的本质是"信用额度"，它把"消费速度"作为信号沿链向上传递。生产者并非无限生产，而是按需生产。理解这一点才能真正理解为什么响应式系统内存可控。

• onSubscribe 中必须调用 request(n) 或 cancel 的时机
• 通过 Subscription 的原子性保证 onNext 串行化
• 终止信号后不得再调用 onNext

自定义 Subscriber 时，onSubscribe 是背压的起点：订阅者必须在这里调用 subscription.request(n)（或 Long.MAX_VALUE 以请求无界），否则收不到任何元素；如果想要取消，则调用 subscription.cancel()。onNext 必须被串行化调用，规范要求 Subscriber 各方法不能被并发调用，因此实现时通常用锁或原子状态保证只有线程能同时进入 onNext；同时，收到 onError/onComplete 后必须停止调用 onNext，通常通过一个状态标志位控制。此外，onNext 中处理完元素后，若采用"逐个拉取"模式，应再次调用 request(1) 以继续。规范还要求 cancel 之后不能再调用 onNext 等。多数情况下直接用 BaseSubscriber 或操作符即可，自定义 Subscriber 主要用于深度定制或嵌入式场景。

自定义 Subscriber 的要点实际是"严格遵守 Reactive Streams 契约"：正确请求、及时取消、串行化、终止后不再发信号。任何一点违反都会导致数据错乱、内存泄漏或死锁。

• TCK 的组成与作用（验证 Publisher/Subscriber 实现符合规范）
• 测试用例覆盖的契约（背压、取消、终止信号）
• 对自定义 Publisher/Subscriber 的验证方式

Reactive Streams TCK 是官方提供的兼容性测试套件，用于验证 Publisher/Subscriber/Processor 的实现是否严格遵守 Reactive Streams 规范。它提供了大量测试用例（如 PublisherVerification、SubscriberWhiteboxVerification 等），覆盖背压、request 计数、终止信号、取消、串行化等各类契约场景。开发者只需让自定义实现继承对应的测试基类并填充工厂方法，TCK 就会自动运行全套规范测试。通过 TCK 验证可以确信自定义实现与其他响应式库（如 Reactor、RxJava）可以互操作，而不会因契约违反而导致难以排查的背压或内存问题。

TCK 的价值在于"互操作性保证"。正因为响应式规范要求实现兼容，TCK 作为统一验证标准，确保不同库的实现能自由组合。对绝大多数开发者而言，使用 Reactor 等成熟库无需自行实现，因此 TCK 主要用于库作者或深度定制场景。

• n 必须为正数（大于 0）
• n 为 Long.MAX_VALUE 表示无界请求
• 非法负数的行为

根据 Reactive Streams 规范，request(n) 中的 n 必须大于 0，即 n 必须是正整数。n 为 Long.MAX_VALUE 表示"无界请求"，即订阅者可以接收任意数量的元素，生产者无需等待更多 request。若 n 为负数或 0，规范要求实现抛出 IllegalArgumentException，这是对契约的强制校验。此外，多次 request(n) 的请求量会累加（累积信用），但累加时需要注意防止溢出（规范允许把 request 视为"至少这多次"）。合法边界是保证背压正确性的基础。

request 的合法边界本质是规范的一致性和安全性约束。n 为正数保证 credit 语义正确，Long.MAX_VALUE 提供无界模式，负数值则触发异常防止错误传播。理解边界有助于定位背压相关的异常。

• Schedulers 各策略（parallel、single、boundedElastic、immediate）的用途
• publishOn 切换下游执行线程，subscribeOn 切换上游（订阅/执行）线程
• 切换位置对线程模型的影响

Reactor 的线程模型由 Schedulers 管理，提供多个线程策略：Schedulers.parallel() 提供适合 CPU 密集型的固定大小并行线程池；Schedulers.single() 提供单线程，适合时间敏感的低开销任务；Schedulers.boundedElastic() 提供有界弹性线程池，适合阻塞式 IO；Schedulers.immediate() 在当前线程立即执行。publishOn 与 subscribeOn 是控制线程切换的操作符：publishOn 影响其下游操作符的执行线程（切换点之后的链在指定调度器运行）；subscribeOn 影响整个订阅链的订阅与执行线程（通常影响上游的产生位置，若已指定则不改变后续的 publishOn 切换）。publishOn 可多次使用，每次切换下游；subscribeOn 通常影响最上游的订阅者进入点。

线程模型的核心是"在哪个线程执行哪段链"。publishOn 是"切下游"，subscribeOn 是"切订阅者产生的源头"。理解两者差异有助于避免阻塞 Netty 线程、合理分配线程资源。

• onError 的终止语义与单次订阅不可复用
• retry/retryWhen 从外层重新订阅实现重试
• 熔断（CircuitBreaker）在响应式场景的封装

在响应式流中，onError 是终止信号，一旦发出订阅即结束，无法在同一订阅内继续。因此重试必须在外层通过"重新订阅"实现：retry() 在收到 onError 后重新订阅上游，retryWhen() 允许自定义重试策略（如退避、次数、条件）。熔断则是更高级的防护：通过 CircuitBreaker 操作符（如 Resilience4j 提供的）在连续失败达到阈值时打开熔断器，直接返回错误而不调用下游，从而保护下游服务。重试与熔断往往组合使用：熔断器在宏观层面防止雪崩，重试在微观层面增强单次成功率，且重试需配合退避与总截止时间防止无限重试。

关键认知是"onError 终止订阅，但可以重新订阅"。重试本质是"用新订阅替代失效订阅"，熔断本质是"用短路保护替代盲目调用"。两者配合能兼顾可靠性与资源保护。

• 虚拟线程的模型（阻塞式写法 + 轻量线程）与响应式（异步回调用法）的对比
• 虚拟线程的优势：可读性、无需改造代码
• 响应式仍占优的场景：高吞吐流式、背压、大量并发连接、可观测组合

虚拟线程旨在让开发者用"阻塞式"的同步代码获得高并发能力，因为虚拟线程极其轻量，可以创建大量虚拟线程处理阻塞 IO，从而极大简化编程模型。响应式则通过异步流与背压实现高并发，但需要回调式编程，学习曲线陡、调试困难。然而，在以下场景响应式仍优于阻塞 + 虚拟线程：一是需要内置背压的流式数据处理（如大流量事件流、LLM 流式输出），因为背压需要显式控制内存与速度；二是需要响应式操作符组合（如重试、熔断、合并、窗口）的复杂数据流；三是单连接内多路复用或长连接保持大量并发（如 WebSocket、RSocket、gRPC 流）；四是需要细粒度线程调度与可观测性的场景。虚拟线程虽简化了阻塞 IO，但无法自动提供背压和流式组合能力。

虚拟线程解决的是"阻塞 IO 的线程成本"问题，响应式解决的是"异步流、背压与组合"问题。两者解决不同维度的痛点。虚拟线程让"写法简单"，响应式让"数据流控制精细"。实际项目中常混合使用，阻塞数据库/外部调用用虚拟线程，流式/背压敏感场景用响应式。

• request(n) 的信用累积语义（request 可多次叠加）
• 多订阅者在共享源（热流）下的独立背压
• 并发 request 的原子累加与防溢出

request(n) 的语义是"订阅者声明还能接收 n 个元素"，多次调用会累加信用额度。对于单订阅者，生产者只需把累计请求量减去已发送量即可掌握剩余额度。对于多个订阅者，每个订阅者拥有独立的 subscription 与独立的信用额度，生产者必须分别为每个订阅者发送各自的元素；若它们是热流（ConnectableFlux 等），每个订阅者的 request 独立计数，生产者按各自额度分别控制发送节奏。对于并发 request 调用，Reactor 内部使用原子操作（如 AtomicLong）累加信用，并处理溢出（防止加和超过 Long.MAX_VALUE），从而保证并发安全。正确理解 request 累积语义是避免"背压丢失"或"过度发送"的关键。

request 的累积性是背压的基础数学结构。每个订阅者独立维护 credit，组合时（如分流、合并）需要按各分支的 request 分别传播。并发安全累加防止了线程竞争导致额度错误。

• 无界 request（Long.MAX_VALUE）的含义
• 背压丢失（请求量未正确传递）导致的生产者无限生产
• 与内存溢出的关系及防范

背压丢失指背压信号未能在链路上正确传递，导致生产者无法感知消费者的处理能力，从而无限生产数据。最常见的原因是使用了无界 request（Long.MAX_VALUE），即订阅者声明"我可以接收任意数量"，此时生产者不再受 request 限制而持续推送，若下游消费者处理慢且没有缓冲限制，数据会在消费者端堆积，最终导致内存溢出（OOM）。此外，自定义 Publisher 未正确传递 request、或操作符未传播下游请求，也会造成背压丢失。防范策略包括：避免不必要地使用无界 request；使用有界缓冲（onBackpressureBuffer 配置上限）；对无法处理的场景显式使用 drop/latest；监控背压与内存指标。

背压丢失的本质是"信用额度失控"。无界 request 是合法的，但会关闭背压保护，因此应谨慎使用。内存溢出是背压丢失最直接的后果，因此生产环境要监控响应式缓冲与内存使用。