限制 IObservable 导致未调用观察者

Question

我有这个简单的代码，其中：

• 创建 IObservable
• 采样半秒
• 使用 ThreadPool 调度程序订阅它
• 使用 SynchronizationContext 观察它

代码如下：

private void DisplayPoints()
{
    var x = 0;
    var ob = this.GeneratePoints();
    ob
      .Sample(TimeSpan.FromMilliseconds(500))
      .SubscribeOn(ThreadPoolScheduler.Instance)
      .ObserveOn(SynchronizationContext.Current)
      .Subscribe(d => Console.WriteLine(d));
}

private IObservable<double> GeneratePoints()
{
    return Observable.Create<double>(o => this.GeneratePoints(o));
}

private IDisposable GeneratePoints(IObserver<double> observer)
{
    var i = 0;
    while (true)
    {
        var value = random.Next(0, 100) * (1 / (double)random.Next(1, Math.Min(50, Math.Max(i, 1))));

        observer.OnNext(value);

        i++;
    }

    return Disposable.Empty;
}

但是，控制台上不会输出任何内容（即永远不会调用匿名观察者）。如果我删除 Sample 运算符，则会调用观察者，尽管这种行为显然不是预期的（UI 线程将被轰炸）。

我显然在这里遗漏了一些东西。我的目的是生成数据，通过 IObserver 推送数据，并通过 UI 显示其中的一些数据。

编辑：由于有些人误解了我的意图（尽管上面已经明确说明），我应该重申一下，我正在尝试做的事情：

• 使用算法生成一些数据（double 值似乎足以解决我的问题）
• 在 GUI 中显示数据

使用IObservable 和响应式扩展似乎是解决我的问题的好方法。

再重复一遍：我不会在实际代码中返回随机数 - 这只是一个占位符 让我的预期行为发挥作用。

Answer 1

您可能不想在紧密循环中生成随机数。最好使用时间间隔。下面每 200 毫秒生成一次随机数。

IObservable<double> observable =
     Observable.Interval(TimeSpan.FromMillSeconds(200))
          .Select((t,i) => random.Next(0, 100) 
                      * (1 / (double)random.Next(1, Math.Min(50, Math.Max(i, 1)))))

Enigmativity 为您编写的代码实际上也是紧密循环的。他关于您在订阅过程中推出价值的错误的观点也是正确的。您必须对代码进行的最小更改才能使其正常工作。

    private static Task GeneratePoints(IObserver<double> observer, CancellationToken token)
    {
        return Task.Run(() =>
        {
            var i = 0;
            var random = new Random();
            while ( true )
            {
                token.ThrowIfCancellationRequested();

                var value = random.Next(0, 100) * ( 1 / ( double ) random.Next(1, Math.Min(50, Math.Max(i, 1))) );

                observer.OnNext(value);

                i++;
            }
        });
    }

稍后

    Observable.Create<double>((observer, token) => GeneratePoints(observer, token));

注意正在传递的取消令牌。当序列的订阅者取消订阅时，将设置此令牌并终止循环。

然而，这是很多工作，Enigmativities 的答案更简单，并为您抽象出上述代码。对于更复杂的情况，了解如何手动执行此操作仍然很有用。

Answer 2

我怀疑您的问题与 Throttle 在内部通过 DefaultScheduler.Instance 引入并发性以及您对 IDisposable GeneratePoints(IObserver<double> observer) 的实现是非标准的事实有关。

尝试像这样重新实现IObservable<double> GeneratePoints()：

private IObservable<double> GeneratePoints()
{
    return Observable.Generate<int, double>(
        0,
        i => true,
        i => i + 1,
        i => random.Next(0, 100) * (1 / (double)random.Next(1, Math.Min(50, Math.Max(i, 1)))));
}

这可能会有所帮助。

问题来自您的 observable 在订阅过程中直接推出值。在创建 observable 时，您应该始终尝试使用标准的内置运算符。上面的代码使用了内置的Generate 操作符，所以它应该和你的代码配合得更好。

Answer 3

Throttle 只会在至少有 500 毫秒的间隙（在您的情况下）时才让值通过。由于GeneratePoints 推送值的速度比这快得多，因此不会发生任何事情。 Sample 可能是您想要的运算符，在这种情况下，它将每 500 毫秒产生一个值。

Source:       1111111111111111111----------111---111111
Throttle (5): -----------------------1-----------------
Sample (5):   ----1----1----1----1---1------------1----1