跳到主要内容

处理需要清理的对象

Doubao 中英对照 Dealing with Objects that Need Cleanup

目录

在非常特定的场景下，您的应用可能会遇到一些类型，它们在不再被使用之后需要某种形式的清理。这是一个高级场景——例如当您使用引用计数对象，或处理堆外对象时。Netty 的 ByteBuf 同时是这两种情况的典型例子。

为了确保这类对象被正确清理，您需要针对每一个 Flux 都加以考虑，同时还要在若干全局钩子中加以处理（请参阅使用全局钩子）：

  • doOnDiscard 这个 Flux/Mono 操作符

  • onOperatorError 钩子

  • onNextDropped 钩子

  • 操作符专属处理器

之所以需要这么多处，是因为每个钩子都是针对特定的一部分清理场景而设计的，而用户可能希望（例如）在 onOperatorError 中，除了清理逻辑之外再实现特定的错误处理逻辑。

请注意，有些操作符并不太适合处理需要清理的对象。例如，bufferWhen 可能会引入相互重叠的缓冲，这意味着我们前面用过的那个“局部钩子”可能会认为第一个缓冲被丢弃了，从而清理其中一个元素，而该元素此时仍存在于第二个缓冲中、仍然是有效的。

important

出于清理的目的，所有这些钩子都必须是幂等的。在某些情况下，同一个对象可能会被它们重复处理多次。与执行类级 instanceOf 检查的 doOnDiscard 操作符不同，这些全局钩子面对的实例可以是任意 Object。需要由用户的实现来区分哪些实例需要清理、哪些不需要。

1. doOnDiscard 操作符或局部钩子​

这个钩子是专门为清理那些原本永远不会暴露给用户代码的对象而设置的。它的定位是：为在正常情况下运行的数据流提供清理钩子（而不是为那些推送了过多元素的畸形源准备的，那种情况由 onNextDropped 覆盖）。

它是局部的，意思是：它通过一个操作符来激活，并且只应用于某个特定的 Flux 或 Mono。

显而易见的场景包括那些从上游过滤元素的操作符。这些元素永远不会到达下一个操作符（或最终订阅者），但这属于正常的执行路径。因此，它们会被传递给 doOnDiscard 钩子。您可能会用到 doOnDiscard 钩子的情形包括：

  • filter：与过滤器不匹配的元素被视为“被丢弃”。

  • skip：被跳过的元素会被丢弃。

  • buffer(maxSize, skip) 且 maxSize < skip：一种“丢弃式缓冲”——两个缓冲之间的元素会被丢弃。

但 doOnDiscard 并不局限于过滤类操作符，那些出于背压目的而在内部排队数据的操作符也会使用它。更具体地说，大多数时候这在取消订阅时尤为重要。一个从数据源预取数据、之后再按需排空给订阅者的操作符，在被取消时可能还有尚未发出的数据。这类操作符会在取消期间使用 doOnDiscard 钩子来清理其内部用于背压的 Queue。

注意

对 doOnDiscard(Class, Consumer) 的每次调用都是叠加生效的，但它只对位于其上游的操作符可见并起作用。

2. onOperatorError 钩子​

onOperatorError 钩子用于以一种横切的方式修改错误（类似于 AOP 中的捕获并重新抛出）。

当错误发生在处理某个 onNext 信号的过程中时，当时正在发出的那个元素会被传递给 onOperatorError。

如果这类元素需要清理，您就需要在 onOperatorError 钩子中实现清理逻辑，而且这通常要叠加在错误改写代码之上。

3. onNextDropped 钩子​

对于畸形的 Publisher，可能会出现这样的情况：某个操作符在本不应该收到元素时却收到了元素（典型情形是在已经收到 onError 或 onComplete 信号之后）。在这种情况下，这个意外元素会被“丢弃”——也就是传递给 onNextDropped 钩子。如果您有需要清理的类型，就必须在 onNextDropped 钩子中检测到它们，并同样在那里实现清理代码。

4. 操作符专属处理器​

有些在操作过程中处理缓冲或收集值的操作符，针对“收集到的数据没有向下游传播”的情况，提供了专属的处理器。如果您把这类操作符用于需要清理的类型，就需要在这些处理器中执行清理。

例如，distinct 就有这样一个回调：当该操作符终止（或被取消）时会调用它，以清理它用来判断元素是否重复的那个集合。默认情况下，该集合是一个 HashSet，清理回调是 HashSet::clear。然而，如果您处理的是引用计数对象，您可能希望把它换成一个更复杂的处理器，在对集合调用 clear() 之前先对其中的每个元素执行 release。