如何从长时间运行的操作中获得中间结果？

Question

参加以下课程，假设Calculate 是一个计算密集型函数。

class Algorithm
{
    FinalResultObject Calculate()
    {
        longPartialCalculation();
        //signal to caller that that part is ready of type MidResult1
        morePartialCalculation();
        //signal more is ready, different type of  MidResult2
        moreWork();
        return finalResult;             
    }
}

现在假设，只要用户准备好，中间结果就需要显示给用户。

我看到的选项是：

1. 使用单独的事件发出信号

2. 使用构造函数注入来注入正在调用其方法的处理程序类

3. 使用 RX observables

我是 RX 新手，但我喜欢我可以轻松地在 UI 线程上进行事件处理的想法。我想知道这是否是矫枉过正而不是预期的，因为它不是真正的整个数据流，而只是每个可观察的一个结果。另一方面，虽然就像事件订阅和取消订阅一样，看起来很麻烦。

有什么提示吗？

Answer 1

解决这个问题的 Rx 方法是定义一个 cold observable，如下所示：

IObservable<Result> Calculate(IScheduler scheduler)
{
  return Observable.Create<Result>(observer => 
    scheduler.Schedule(() =>
    {
       observer.OnNext(longPartialCalculation());
       observer.OnNext(morePartialCalculation());
       observer.OnNext(moreWork());
       observer.OnCompleted();
    }));
}

// Depending upon your needs, you could use inheritance as follows:
public abstract class Result { ... }
public class MidResult1 : Result { ... }
public class MidResult2 : Result { ... }
public class FinalResultObject : Result { ... }

您还可以定义一个指定默认调度程序的重载，例如 ThreadPoolScheduler 如果您想引入并发，或者 CurrentThreadScheduler 如果您不想。

要使用此方法返回的可观察对象，只需使用观察者调用Subscribe。您可以提供一个OnNext 处理程序来检查每个到达时的Result 对象，并提供一个OnCompleted 处理程序来处理完成。如果必须，您还可以提供OnError 处理程序来处理Exception。 （编辑：请注意，OnError 并没有被我的示例调用。）

如果您想确保所有这些处理程序都在 UI 线程上执行，并且您已将引入并发的调度程序（例如 ThreadPoolScheduler）传递给 Calculate 方法，那么您还可以应用 @987654334 @ 运算符（或基于 XAML 的平台上的 ObserveOnDispatcher）将所有通知编组到 UI 线程以供观察。

algo.Calculate(ThreadPoolScheduler.Instance)
    .ObserveOnDispatcher()
    .Subscribe(OnNextResult, OnCompleted);

请注意，Rx 的主要好处之一是能够查询；例如，一个简单的过滤器：

algo.Calculate(ThreadPoolScheduler.Instance)
    .Where(result => result.HasRequiredState)
    .ObserveOnDispatcher()
    .Subscribe(result => handle(result.RequiredState));

Answer 2

您可以使用 .Net 的 Progress<T>。您创建一个实例，传递一个处理程序或注册到它的事件，并在整个长时间运行的过程中通过它进行报告：

var progress = new Progress<string>(value => Console.WriteLine(value));
Calculate(progress);

FinalResultObject Calculate(IProgress<string> progress)
{
    longPartialCalculation();
    progress.Report("MidResult1");
    morePartialCalculation();
    progress.Report("MidResult2");
    moreWork();
    return finalResult;
}

在这种情况下，报告正在向控制台写入一个字符串，但您当然可以将其用于任何您想要的类型。

Progress<T> 还会在创建时捕获当前的SynchronizationContext，因此您可以在 UI 线程中创建它，并将其传递给非 UI 线程，而不会出现任何同步问题。