为什么这个 CancellationDisposable 永远不会在 Observable.Dispose() 上被取消？

Question

我在 WinForms 应用程序中使用 RxFramework。我尝试运行 Observable 异步并使用 CancellationDisposable 在用户单击按钮时取消操作。但它不起作用！

假设我有一个带有 2 个按钮和一个 ProgressBar 的表单。 Button1_click 订阅新线程上的观察者。然后立即按下 Button2_click 以取消操作。 为什么 cancel.Token.IsCancellationRequested 永远不会为真？

private IDisposable obs = null;
private void button1_Click(object sender, EventArgs e) {
    var countObserver = Observable.Create<int>(observer => {
        var cancel = new CancellationDisposable();

        if (!cancel.Token.IsCancellationRequested) {
            //Step 1 of a long running process using lot of resources...
            observer.OnNext(1);
        }
        if (!cancel.Token.IsCancellationRequested) {
            //Step 2 of a long running process using lot of resources...
            observer.OnNext(1);
        }
        if (!cancel.Token.IsCancellationRequested) {
            //Step 3 of a long running process using lot of resources...
            observer.OnNext(1);
        }
        observer.OnCompleted();

        return cancel;
    });

    obs = countObserver
        .ObserveOn(new ControlScheduler(this))
        .SubscribeOn(Scheduler.ThreadPool)
        .Subscribe(i => {
            //Update a progress bar here...
        });

}

private void button2_Click(object sender, EventArgs e) {
    if (obs != null)
        obs.Dispose();
}

Answer 1

之所以会发生这种情况，是因为您传递给 Observable.Create 的 lambda 不会返回 CancellationDisposable 直到它完成所有步骤。因此，操作的顺序如下：

1. 您调用 Subscribe
2. UI 线程块
3. 在 ThreadPool 线程上，执行进入
4. 创建 lambda CancellationDisposable
5. 您多次检查取消并执行整个过程。在此期间，您的进度条会更新。
6. cancel 从 lambda
7. UI 线程被释放返回，obs 获取其值
8. 您单击 Button2 并调用 obs.Dispose
9. cancel 获取 cancel.Token.IsCancellationRequested=true。但什么也没有发生，因为一切都已经发生了。

这是固定代码：

private void button1_Click(object sender, EventArgs e) {
  var countObserver = Observable.Create<int>(observer => {
    var cancel = new CancellationDisposable();

    // Here's the magic: schedule the job in background and return quickly
    var scheduledItem = Scheduler.ThreadPool.Schedule(() =>
    {
      if (!cancel.Token.IsCancellationRequested) {
        //Step 1 of a long running process using lot of resources...
        observer.OnNext(1);
      }
      if (!cancel.Token.IsCancellationRequested) {
        //Step 2 of a long running process using lot of resources...
        observer.OnNext(1);
      }
      if (!cancel.Token.IsCancellationRequested) {
        //Step 3 of a long running process using lot of resources...
        observer.OnNext(1);
      }
      observer.OnCompleted();
    });

    return new CompositeDisposable(cancel, scheduledItem);
});

obs = countObserver
    .ObserveOn(new ControlScheduler(this))
    .Subscribe(i => {
        //Update a progress bar here...
    });

}

Answer 2

相反，上面的代码存在许多错误，但实际上有一种更好的方法来完成此操作（警告：前面在 TextArea 中编码）：

countObservable = Observable.Timer(new ControlScheduler(this));

var buttonObservable = Observable.FromEventPattern<EventArgs>(
    x => button1.Click += x, x => button1.Click -= x);

countObservable
    .TakeUntil(buttonObservable)
    .Subscribe(x => /* Do stuff */);