无法取消订阅 Rx

Question

背景

我正在编写一些执行以下操作的软件：

用户点击"start"。
任务已启动，执行一些工作并启动事件以更新 GUI。
可观察对象使用来自任务的事件，并将数据打印到 GUI 中的富文本框。

第一次点击 "START" 时一切正常，但之后就不行了。第一次单击开始时，我得到如下所示的输出：

这看起来不错，没什么问题。但是，当我第二次单击“开始”时，我得到以下输出。

现在，我相信我知道为什么会发生这种情况。据我所知，我的观察者从我第一次单击“开始”开始就从未取消订阅，因此所有内容都会打印两次。单击开始按钮时，会发生以下情况：

    /// <summary>
    /// Starts the test.
    /// </summary>
    /// <param name="sender">The "start" button.</param>
    /// <param name="e">Clicking on the "start" button.</param>
    private void button_go_Click(object sender, RoutedEventArgs e)
    {
        var uiContext = SynchronizationContext.Current;
        var results = Observable.FromEventPattern<TestResultHandler, TestResultArgs>(
            handler => (s, a) => handler(s, a),
            handler => this.myTest.Results += handler,
            handler => this.myTest.Results -= handler)
            .ObserveOn(uiContext)
            .Subscribe(x => this.richTextBox_testResults.Document.Blocks.Add(new Paragraph(new Run(x.EventArgs.Results))));

        // start running the test
        this.runningTest = new Task(() => { this.myTest.Run(); });
        this.runningTest.Start();

        // block on purpose, wait for the task to finish
        this.runningTest.Wait();
    }

我不熟悉 .NET 中的响应式扩展（使用它还不到一小时）。我基本上使用了 Stephen Cleary 在 C# 中的并发指南中的示例，以便从这里开始。尽管如此，我相信我有点知道问题出在哪里...

拟议计划

如果我可以取消订阅 observable，那么我认为这个问题就会消失。不过，我认为这有点复杂，基于 ReactiveX.io 中使用的反应模式，听起来我实际上应该只在任务实际存在期间巧妙地观察任务，然后自然地取消订阅。这是我在调试这个问题时所得到的，我不是 100% 确定这个取消订阅的东西真的能解决问题。

问题

有什么方法可以取消订阅 Observable 吗？或者，有没有一种方法可以让我仅在存在期间观察我的任务？

Answer 1

Rx 使用 IDisposable 接口使您能够取消订阅尚未自然结束的可观察订阅。如果您的 Observable 发送 OnCompleted 或 OnError 通知，那么订阅会自动为您处理。

这种方法的一个明显优势是您可以创建一个 CompositeDisposable 来聚合您的所有订阅以启用单个退订。这比删除事件处理程序所需的大杂烩要好得多。

在您的代码中，您不会结束订阅，因此每次点击 button_go 您都会创建一个新的订阅。

您可以使用四种解决方案，每种都有所不同。

(1)

请记住，需要调用 .Dispose() 的关键是您的可观察订阅是否自然结束并且您希望它结束。因此，您可以简单地在查询中添加一个 .Take(1)，使其在产生一个值后自然结束。

private void button_go_Click(object sender, RoutedEventArgs e)
{
    var uiContext = SynchronizationContext.Current;
    var results = Observable.FromEventPattern<TestResultHandler, TestResultArgs>(
        handler => (s, a) => handler(s, a),
        handler => this.myTest.Results += handler,
        handler => this.myTest.Results -= handler)
        .Take(1)
        .ObserveOn(uiContext)
        .Subscribe(x => this.richTextBox_testResults.Document.Blocks.Add(new Paragraph(new Run(x.EventArgs.Results))));

    // start running the test
    this.runningTest = new Task(() => { this.myTest.Run(); });
    this.runningTest.Start();

    // block on purpose, wait for the task to finish
    this.runningTest.Wait();
}

然后系统会自动为您处理订阅。

(2)

您可以使用 SerialDisposable 来管理每个订阅。 MSDN 文档将其描述为：

Represents a disposable whose underlying disposable can be swapped for another disposable which causes the previous underlying disposable to be disposed.

您的代码将如下所示：

private SerialDisposable _results = new SerialDisposable();

private void button_go_Click(object sender, RoutedEventArgs e)
{
    var uiContext = SynchronizationContext.Current;
    _results.Disposable = Observable.FromEventPattern<TestResultHandler, TestResultArgs>(
        handler => (s, a) => handler(s, a),
        handler => this.myTest.Results += handler,
        handler => this.myTest.Results -= handler)
        .ObserveOn(uiContext)
        .Subscribe(x => this.richTextBox_testResults.Document.Blocks.Add(new Paragraph(new Run(x.EventArgs.Results))));

    // start running the test
    this.runningTest = new Task(() => { this.myTest.Run(); });
    this.runningTest.Start();

    // block on purpose, wait for the task to finish
    this.runningTest.Wait();
}

(3)

您始终可以确保只创建一次订阅。

private IDisposable _results = null;

private void button_go_Click(object sender, RoutedEventArgs e)
{
    if (_results == null)
    {
        var uiContext = SynchronizationContext.Current;
        _results = Observable.FromEventPattern<TestResultHandler, TestResultArgs>(
            handler => (s, a) => handler(s, a),
            handler => this.myTest.Results += handler,
            handler => this.myTest.Results -= handler)
            .ObserveOn(uiContext)
            .Subscribe(x => this.richTextBox_testResults.Document.Blocks.Add(new Paragraph(new Run(x.EventArgs.Results))));
    }

    // start running the test
    this.runningTest = new Task(() => { this.myTest.Run(); });
    this.runningTest.Start();

    // block on purpose, wait for the task to finish
    this.runningTest.Wait();
}

(4)

最后一种方法是将整个操作包装在一个可观察对象中，并针对每个新订阅发起。这是使用 Rx 的正确方法。 Observables 应该保持自己的状态，这样订阅就可以 100% 相互独立。

现在，您的代码使用 this.myTest 和 this.runningTest，因此它显然具有状态。您应该尝试删除这些。但如果不这样做，您的代码将如下所示：

private void button_go_Click(object sender, RoutedEventArgs e)
{
    var uiContext = SynchronizationContext.Current;
    Observable
        .Create<System.Reactive.EventPattern<TestResultArgs>>(o =>
        {
            var subscription =
                Observable
                    .FromEventPattern<TestResultHandler, TestResultArgs>(
                        h => this.myTest.Results += h,
                        h => this.myTest.Results -= h)
                    .Take(1)
                    .Subscribe(o);
            this.runningTest = new Task(() => { this.myTest.Run(); });
            this.runningTest.Start();
            return subscription;
        })
        .ObserveOn(uiContext)
        .Subscribe(x => this.richTextBox_testResults.Document.Blocks.Add(new Paragraph(new Run(x.EventArgs.Results))));

    // block on purpose, wait for the task to finish
    this.runningTest.Wait();
}

理想情况下，您应该将 myTest 的创建和销毁合并到可观察对象中。

所以，我倾向于做这样的事情：

private void button_go_Click(object sender, RoutedEventArgs e)
{
    var uiContext = SynchronizationContext.Current;
    Observable
        .Create<System.Reactive.EventPattern<TestResultArgs>>(o =>
        {
            var myTest = new MyTest();
            var subscription =
                Observable
                    .FromEventPattern<TestResultHandler, TestResultArgs>(
                        h => myTest.Results += h,
                        h => myTest.Results -= h)
                    .Take(1)
                    .Subscribe(o);
            myTest.Run();
            return subscription;
        })
        .ObserveOn(uiContext)
        .Subscribe(x => this.richTextBox_testResults.Document.Blocks.Add(new Paragraph(new Run(x.EventArgs.Results))));
}

最后一个确实是您问题的答案："Or, is there a way that I could observe my task only for the duration that it exists?"

无法取消订阅 Rx

Can't unsubscribe from Rx

c#

task

reactive-programming

winforms

system.reactive