Observable.Defer - 需要澄清它究竟是做什么的

Question

假设我想生成一个随机数的异步流,每100毫秒抽出一个新值.在尝试提出解决方案时,我的第一次尝试看起来像这样:

        var random = new Random();
        Observable.Start(() => random.Next())
                  .Delay(TimeSpan.FromMilliseconds(100))
                  .Repeat()
                  .Subscribe(Console.WriteLine);

如果你尝试运行它,你会发现它只是一遍又一遍地重复相同的值.好吧,我想我误解了重复是如何工作的.玩了一会儿之后,我想出了这个并且它有效:

        var random = new Random();
        Observable.Defer(()=> Observable.Start(() => random.Next()))
                  .Delay(TimeSpan.FromMilliseconds(100))
                  .Repeat()
                  .Subscribe(Console.WriteLine);

所以我去了MSDN文档来了解Defer实际上在做什么,这就是它所说的:

返回一个可观察的序列,只要新的观察者订阅,就会调用可观察的工厂.

我想我的困惑是这样的:在我的代码示例中,我只是订阅了一次Observable,那为什么它似乎Observable.Start(...)一遍又一遍地调用？还是我误会了Repeat()？任何澄清都会很棒.

Answer 1

你是误解重复.基本上,重复将在每次完成时继续重新订阅观察.

Observable.Start显然在第一次调用Subscribe时缓存该值,并在每次订阅时返回它.这是在第一个例子中反复导致相同数字的原因.

Defer的工作方式如下:

IObservable<T> Defer(Func<IObservable<T>> factory)
{
    return Observable.Create<T>(obs => factory().Subscribe(obs));
}

因此,每次订阅Deferobservable时,都必须调用该factory函数.在这种情况下,工厂创建一个新的 observable,从而避免了缓存行为Start.

获得所描述序列的另一种方法是使用Interval获取时间并Select获取随机数.

Observable.Interval(TimeSpan.FromMilliseconds(100))
          .Select(i => random.Next())
          .Subscribe(Console.WriteLine);

Answer 2

"生成"随机数的简单方法是使用.Generate.试试这个:

var rnd = new Random();

Observable
    .Generate(
        0,
        x => true,
        x => x,
        x => rnd.Next(),
        x => TimeSpan.FromMilliseconds(100.0))
    .Subscribe(n => { });