c# - 为什么我用 Repeat() 得到不确定的结果

Question

我正在努力扩展我在 Rx 方面的知识。所以我只是在玩弄流，并尝试让它们按照我期望的方式运行。

虽然我已经读到 Repeat() 运算符在实践中有困难，因为您可能会在 OnCompleted 和重新订阅之间丢失通知，但我无法自行弄清楚为什么会发生以下情况。

        var subject = new Subject<string>();

        var my = subject
            .Take(1)
            .Merge(Observable.Empty<string>().Delay(TimeSpan.FromMilliseconds(2000)))
            .Repeat();
        my.Subscribe(Console.WriteLine);

        var stopwatch = new Stopwatch();
        stopwatch.Start();
        Scheduler.ThreadPool.Schedule(TimeSpan.FromSeconds(1), () => subject.OnNext("1 at " + stopwatch.ElapsedMilliseconds));
        Scheduler.ThreadPool.Schedule(TimeSpan.FromSeconds(2), () => subject.OnNext("2 at " + stopwatch.ElapsedMilliseconds));
        Scheduler.ThreadPool.Schedule(TimeSpan.FromSeconds(3), () => subject.OnNext("3 at " + stopwatch.ElapsedMilliseconds));
        Scheduler.ThreadPool.Schedule(TimeSpan.FromSeconds(4), () => subject.OnNext("4 at " + stopwatch.ElapsedMilliseconds));
        Scheduler.ThreadPool.Schedule(TimeSpan.FromSeconds(5), () => subject.OnNext("5 at " + stopwatch.ElapsedMilliseconds));
        Scheduler.ThreadPool.Schedule(TimeSpan.FromSeconds(6), () => subject.OnNext("6 at " + stopwatch.ElapsedMilliseconds));

        Console.ReadLine();

当我运行这个例子时，结果是完全不确定的:

结果 1:

1 at 1006
3 at 3007
5 at 4995

遗漏了 2 和 4 很好，但即使在这个结果中也有一些奇怪之处，因为实际上 3 和 5 之间并没有真正的 2 秒差距。

然而，结果可能更糟。看到这个:

1 at 1003
2 at 2003
4 at 4005
6 at 6004

1 和 2 之间没有 2 秒的差距。正好是一秒。他为什么不把它留下来？

如果有人能为我澄清事情，我会非常高兴!

编辑

我刚刚注意到这里可能是 Merge 出错了。如果我将查询重构为 Concat，事情似乎会按预期发生:

        var my = subject
            .Take(1)
            .Concat(Observable.Empty<string>().Delay(TimeSpan.FromMilliseconds(2000)))
            .Repeat();

Answer 1

Windows(和其他桌面操作系统)不是运行时操作系统，因此您不能指望它的计时器精确到毫秒。特别是如果您有更多计时器，这可能会导致不确定的行为，这正是您的情况。

这是您的原始序列的工作方式:

• 时间 ~ 0
  • Take(1) 订阅subject
  • 延迟的空可观察对象的计时器开始计时
• 时间 ~ 1
  • 1被添加到subject，1被写出；在此之后，没有人再订阅 subject
• 时间 ~ 2
  • 延迟的空可观察对象的计时器用完。因此，Take(1) 再次订阅了 subject 并且另一个用于延迟空可观察对象的计时器开始
  • 大约同时，2被添加到subject

由于时间上的细微差别，两个 Action 的时间差不多。 2 可以按任何顺序发生。并且顺序很重要，在 Take() 重新订阅之前或之前添加 2。因此，2 可以写出，也可以不写出。

如果你想要的是这样的序列:

1. 等待第一个项目并返回
2. 等待大约两秒钟
3. 等待第二个项目并将其返回(忽略两秒等待期间添加的任何项目)
4. …

那么我认为您编辑的代码是正确的。

但这绝不能保证确定性的结果。在我的计算机上，如果我将延迟的空可观察对象的等待时间更改为 1960 毫秒，我在使用 Concat() 时会得到不确定的结果。