订阅者缺少消息;这是Rx还是它错误的错误？

Question

我有下面的构造

 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"))); 

一个窗口WPF应用程序现在这里是列表的截图 - 注意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)); 

现在，因为你是直接相继签署这项观察到两次这种观察到的两个实例将试图生成值几乎在同一时间。所以价值DateTime.Now.Millisecond将大部分时间相同，但现在总是。从new Random(x).NextDouble()返回的值对于相同的x是相同的，因此为什么大多数时候您从可观察值的两个实例获得相同的值。只是当DateTime.Now.Millisecond是不同的，你会得到两个不同的值，看起来用户缺少值。

---

这里的替代版本，你最初预期应该工作：“这是其中Rx错误或它的我做错了”

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();