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

Posted 2023-03-31

【中文标题】订阅者缺少消息；这是 Rx 的错误还是我做错了？

【英文标题】：Subscribers are missing messages; Is this a bug with Rx or its me doing it wrong?

【发布时间】：2013-09-14 21:40:39

【问题描述】：

我有一个带有以下构造函数的 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...这种行为是间歇性的。

【问题讨论】：

“这是 Rx 的错误还是我做错了？” - 你给赔率吗？

我只是在问一个问题。赔率肯定对我不利。

图片似乎是一个损坏的链接。

我可以正常查看，您的浏览器/计算机可能会阻止来自您正在浏览的站点以外的站点的图像。这是网址s22.postimg.org/9paxw9rq9/Rx_bug.png

似乎是公司网络阻止了它。

【参考方案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();

【讨论】：

很棒的观察。感谢分享。

你成就了我的一天...我将代码更改为使用 Second 而不是 Millisecond 现在我看不到这个问题了。

@fahadash - 将其更改为秒并不能解决问题，它只会降低 1000 倍的可能性。然后你就不能正确地产生随机数。你需要重构 observable 并且可能使用Publish 来共享 observable。

@fahadash - 我已经为您添加了一个替代解决方案，它的行为应该与您最初预期的一样。

..我认为数字序列可以进一步简化为Observable.Interval(TimeSpan.FromSeconds(1.0)).Select(_=>rnd.NextDouble() * 99 + 2)。我认为这更具表现力，即每秒从这个序列中投射一个随机数。