订阅者缺少消息；这是 Rx 的错误还是我做错了？

Question

我有一个带有以下构造函数的 Window WPF 应用程序

        numbers = Observable.Generate(DateTime.Now,
                                         time => true,
                                         time => DateTime.Now,
                                         time => { return new     Random(DateTime.Now.Millisecond).NextDouble() * 99 + 2; },
                                         time => TimeSpan.FromSeconds(1.0));


        numbers.ObserveOnDispatcher()
            .Subscribe(s => list1.Items.Add(s.ToString("##.00")));

        numbers.Where(n => n < 10).ObserveOnDispatcher().
            Subscribe(s => list2.Items.Add(s.ToString("##.00")));

现在这里是列表的屏幕截图 - 注意左侧列表中缺少 3.76...这种行为是间歇性的。

Answer 1

简短的回答是你做错了。 Rx 运行良好。

当您创建一个 observable 时，您是在创建一个随时间变化的值序列的定义，而不是随时间变化的实际值序列。这意味着，每当您拥有 observable 的订阅者时，您都会为每个订阅者创建 observable 的新实例。

因此，在您的情况下，您有两个运行此序列的实例：

var numbers =
    Observable
        .Generate(
            DateTime.Now,
            time => true,
            time => DateTime.Now,
            time => new Random(DateTime.Now.Millisecond)
                .NextDouble() * 99 + 2,
            time => TimeSpan.FromSeconds(1.0));

现在，由于您立即连续两次订阅此 observable，因此此 observable 的两个实例将尝试几乎同时生成值。所以DateTime.Now.Millisecond 的值在大多数情况下都是一样的，但现在总是一样。从new Random(x).NextDouble() 返回的值对于相同的x 是相同的，因此为什么大多数时候您从 observable 的两个实例中获得相同的值。只是当DateTime.Now.Millisecond 不同时，您会得到两个不同的值，并且订阅者似乎缺少值。

---

这是一个替代版本，应该可以按您最初的预期工作：

var rnd = new Random((int)DateTime.Now.Ticks);

var numbers =
    Observable
        .Generate(0, n => true, n => 0,
            n => rnd.NextDouble() * 99 + 2,
            n => TimeSpan.FromSeconds(1.0));

var publishedNumbers = numbers.Publish();

publishedNumbers
    .ObserveOnDispatcher()
    .Subscribe(s => list1.Items.Add(s.ToString("##.00")));

publishedNumbers
    .Where(n => n < 10)
    .ObserveOnDispatcher()
    .Subscribe(s => list2.Items.Add(s.ToString("##.00")));

publishedNumbers.Connect();