system.reactive - 为什么我需要在完成后处理订阅？

Question

Intro To RX 一书将 OnSubscribe 的返回值描述为 IDisposible 并指出在调用 OnError 和 OnCompleted 时应该处理订阅。

An interesting thing to consider is that when a sequence completes or errors, you should still dispose of your subscription.

From Intro to RX: Lifetime Management, OnError and OnCompleted

这是为什么？

作为引用，这是我目前正在研究的类(class)。我可能会在某个时候将其提交给代码审查。

using System;
using System.Reactive;
using System.Reactive.Linq;
using System.Reactive.Subjects;

/// <summary>
/// Provides a timeout mechanism that will not timeout if it is signalled often enough
/// </summary>
internal class TrafficTimeout
{
    private readonly Action onTimeout;
    private object signalLock = new object();
    private IObserver<Unit> signals;

    /// <summary>
    /// Initialises a new instance of the <see cref="TrafficTimeout"/> class.
    /// </summary>
    /// <param name="timeout">The duration to wait after receiving signals before timing out.</param>
    /// <param name="onTimeout">The <see cref="Action"/> to perform when the the timeout duration expires.</param>
    public TrafficTimeout(TimeSpan timeout, Action onTimeout)
    {
        // Subscribe to a throttled observable to trigger the expirey
        var messageQueue = new BehaviorSubject<Unit>(Unit.Default);
        IDisposable subscription = null;
        subscription = messageQueue.Throttle(timeout).Subscribe(
        p =>
        {
            messageQueue.OnCompleted();
            messageQueue.Dispose();
        });

        this.signals = messageQueue.AsObserver();
        this.onTimeout = onTimeout;
    }

    /// <summary>
    /// Signals that traffic has been received.
    /// </summary>
    public void Signal()
    {
        lock (this.signalLock)
        {
            this.signals.OnNext(Unit.Default);
        }
    }
}

Answer 1

Subscribe 扩展方法返回的一次性对象仅返回以允许您在 可观察对象自然结束之前手动取消订阅可观察对象 。

如果 observable 完成 - 使用 OnCompleted 或 OnError - 那么订阅已经为你处理了。

试试这个代码:

var xs = Observable.Create<int>(o =>
{
 var d = Observable.Return(1).Subscribe(o);
 return Disposable.Create(() =>
 {
 Console.WriteLine("Disposed!");
 d.Dispose();
 });
});

var subscription = xs.Subscribe(x => Console.WriteLine(x));

如果你运行上面的你会看到“Disposed!”当 observable 完成时写入控制台，而无需在订阅上调用 .Dispose()。

需要注意的一件重要事情:垃圾收集器永远不会在可观察订阅上调用 .Dispose()，因此如果订阅在您的订阅超出范围之前没有(或可能没有)自然结束，您 必须 处理您的订阅。

以这个为例:

var wc = new WebClient();

var ds = Observable
    .FromEventPattern<
        DownloadStringCompletedEventHandler,
        DownloadStringCompletedEventArgs>(
            h => wc.DownloadStringCompleted += h,
            h => wc.DownloadStringCompleted -= h);

var subscription =
    ds.Subscribe(d =>
        Console.WriteLine(d.EventArgs.Result));

ds observable 只会在它有订阅时附加到事件处理程序，并且只会在 observable 完成或订阅被处理时分离。由于它是一个事件处理程序，observable 永远不会完成，因为它正在等待更多事件，因此处理是与事件分离的唯一方法(对于上面的示例)。

当您有一个 FromEventPattern observable 并且您知道它只会返回一个值时，明智的做法是在订阅之前添加 .Take(1) 扩展方法以允许事件处理程序自动分离，然后您无需手动处理订阅。

像这样:

var ds = Observable
    .FromEventPattern<
        DownloadStringCompletedEventHandler,
        DownloadStringCompletedEventArgs>(
            h => wc.DownloadStringCompleted += h,
            h => wc.DownloadStringCompleted -= h)
    .Take(1);

我希望这有帮助。