C# Reactive Extensions - 我应该退回一次性用品吗？答案

【问题标题】：C# Reactive Extensions - Should I return a disposable?C# Reactive Extensions - 我应该退回一次性用品吗？
【发布时间】：2019-12-06 20:22:26
【问题描述】：

我试图了解何时是在传递给Observable.Create 的函数中返回一次性的好时机，而不是通过using 语句在范围内处置任何资源。

对于Observable 是无限流的情况，是否会返回更多一次性用品？即使是这样，我也不明白即使流过早关闭，using 块仍不会释放资源

【问题讨论】：

您是否担心不处置一次性订阅可能产生的后果？
@TheodorZoulias 也许这更像是一个懒惰vs-eager清理的问题？我只是没有真正在网上看到任何建议/最佳实践来解释什么时候该做什么以及这样做的实际原因。
我认为在 RX 中使用 Disposables 仅用于取消订阅目的，而不是用于释放非托管资源。
@TheodorZoulias - 可以有托管资源 - 特别是对于基于时间的可观察对象。
您永远不应该在一次性管道中使用using。始终使用Observable.Using。如果我理解你的问题，你正在写一个Observable.Create，你在Create 内的using 语句中调用o.OnNext 和o.OnCompleted？如果是这样，请停止它，因为它会导致您的 observable 在订阅完成之前运行完成。

标签： c# reactive-programming system.reactive

【解决方案1】：

我认为 Observable 范式上的 Disposable 接口仅用于摆脱订阅的目的（即停止对观察到的事件的回调），正如 Theodor Zoulias 指出的那样。它不管理流上的任何资源。您可能会混淆 Disposable 接口在其他场景中的使用。

关于处置订阅：

我可以看到返回 Disposable 的用例之一是当您有多个调用函数时：假设您有一个 Observable 列表，您可以对其进行迭代并调用函数 .Dispose()一次取消多个订阅。

您还可以将该流作为一次性的传递给另一个 Observable，以便在某些事件发生时被释放。由于整个 RX 范式是关于不知道什么时候执行的，这很有趣。我在一个应用程序中工作，如果发生某个事件，我必须取消订阅，并且我将 Observable Subscription (IDisposable) 传递给此类事件/流的观察者。

以下几行：

 IDisposable subscription1 = observableOne.Subscribe(_ => # code omitted); 
 observableTwo.Subscribe(_ => { 
    subscription1?.Dispose(); 
    subscription1 = null; 
});

正如 Enigmaticy 所指出的，虽然这体现了我的观点，但实现这一点的更好方法是：

observableOne.TakeUntil(observableTwo).Subscribe(_ => #code ommited);

我有一段时间没有使用 C#，但这些是我在使用 vs Disposable 作为对象时可以看到的用例。当您想取消订阅时，它为您提供了更大的灵活性：~

【讨论】：

嗯，我不确定这是否完全回答了我的问题。但听起来这个想法是将资源的处置交给可观察对象本身。这可能只是性能开销的原因吗？我们是否应该避免在subscribe 实现中调用dispose，而是返回disposable 或action 来处理任何资源？
我认为 Observable 范式上的 Disposable 接口仅用于摆脱订阅的目的（即停止对观察到的事件的回调），正如 Theodor Zoulias 指出的那样。它不管理流上的任何资源。 .Dispose() 方法将简单地调用 Observers 上的 OnComplete。
“.Dispose() 方法将简单地在观察者上调用 OnComplete” - 不，这不会发生。可观察的只是不复存在。当您在订阅中调用 .Dispose() 时，不会调用 OnCompleted。
“将流作为一次性的传递给另一个 Observable”是什么意思？我不明白你会怎么做。
@cesartalves - GC 永远不会在任何对象上调用 .Dispose()。它可能调用一个调用.Dispose()的终结器，但你必须检查每个对象的实现以了解是否是这种情况。

【解决方案2】：

感谢在这篇文章中回复的每个人，这帮助我更好地理解了一点。我对 RX 的理解有很多磕磕绊绊，我认为其中很多都归结为有限的文档，而且似乎网上很多人也不太完全理解，所以有很多错误信息需要整理。

这个其他答案对我有用 https://stackoverflow.com/a/7707768/7183974.

真正归结为当我们的Observable.Create 方法中有非阻塞代码时。因此，当我们的 observable 被订阅时，我们会立即返回一个一次性的，它可以在我们需要提前取消订阅的情况下清理任何异步/并发进程。

这对于您的 observable 使用其他异步（基于推送）代码的情况是必要的。

对于您只想基于推送的迭代（基于拉取）代码，您可以使用Observable.Create 但我认为仅使用迭代器会更好，如果您需要它是基于推送的 API，那么只需使用ToObservable。

我试图实现一个基于推送的迭代器，所以一次性对我来说似乎是多余的，这让我感到困惑。我已经将我的代码重构为基于拉的，如果我需要它再次基于推送，我会使用ToObservable。

【讨论】：