为什么通过 Latest 在 RX.NET 中订阅 observable 只接受 1 个订阅者？

Question

我的目标是从一个 observable 中获得两个订阅者，但我只对事件流中的最新项目感兴趣。我希望其他人被丢弃。将其视为每 1 秒更新一次且忽略任何中间值的股票价格屏幕。 这是我的代码：

    var ob = Observable.Interval(TimeSpan.FromMilliseconds(100)) // fast event source
        .Latest().ToObservable().ToEvent();

    ob.OnNext += (l =>
                      {
                          Console.WriteLine(Thread.CurrentThread.ManagedThreadId);
                          Thread.Sleep(1000); // slow processing of events
                          Console.WriteLine("Latest: " + l);
                                                    });

    ob.OnNext += (l =>
        {
            Console.WriteLine(Thread.CurrentThread.ManagedThreadId);
            Thread.Sleep(1000); // slow processing of events
            Console.WriteLine("Latest1: " + l);
            //  subject.OnNext(l);
        });

然而，由于上面的代码，尽管我附加了两个事件（即使您使用订阅表示法也没关系）只有第一个订阅被定期调用。第二个根本不运行。为什么会这样？

Answer 1

首先我认为您的要求是以下之一：

1. 您只想获取未来值
2. 或者您想获取最新值（如果有）和任何未来值
3. 或者您只需要最近的值（如果有）
4. 或者您只想对刻度进行采样并获取每秒的值
5. 或者您的消费者速度较慢，您需要执行减载（例如在 GUI 中）

1) 的代码

var ob = Observable.Interval(TimeSpan.FromMilliseconds(1000)) // fast event source
    .Publish();
ob.Connect();

2) 的代码

var ob = Observable.Interval(TimeSpan.FromMilliseconds(1000)) // fast event source
    .Replay(1);
ob.Connect();    

3) 的代码

var ob = Observable.Interval(TimeSpan.FromMilliseconds(1000)) // fast event source
    .Replay(1);
ob.Connect();  
var latest = ob.Take(1); 

4) 的代码可以是这样，但在您认为的窗口周围有一些微妙的行为。

var ob = Observable.Interval(TimeSpan.FromMilliseconds(200)) // fast event source
    .Replay(1);
//Connect the hot observable
ob.Connect();

var bufferedSource = ob.Buffer(TimeSpan.FromSeconds(1))
    .Where(buffer => buffer.Any())
    .Select(buffer => buffer.Last());  

5) 的代码可以在 James World 的博客 http://www.zerobugbuild.com/?p=192 上找到，并且在伦敦的许多银行应用程序中很常见。

Answer 2

我认为你不明白 .Latest() 的作用。

public static IEnumerable<TSource> Latest<TSource>(
    this IObservable<TSource> source
)

在每次迭代时返回最后一个采样元素并随后阻塞直到可观察源序列中的下一个元素可用的可枚举序列。

请注意，它会阻塞等待来自 observable 的下一个元素。

因此，当您使用.Latest() 将IObservable<> 转换为IEnumerable<> 时，您必须使用.ToObservable() 将其转换回IObservable<> 才能调用.ToEvent()。那就是它倒下的地方。

此代码的问题在于您创建的代码会阻塞。

如果你这样做，它会起作用：

var ob = Observable.Interval(TimeSpan.FromMilliseconds(100)).ToEvent();

没有必要打电话给.Latest()，因为您总是从可观察对象中获取最新值。您永远无法获得更早的值。它是可观察的，而不是时间机器。

我也不明白你为什么打电话给.ToEvent()。有什么需求？

这样做：

var ob = Observable.Interval(TimeSpan.FromMilliseconds(100));

ob.Subscribe(l =>
{
    Console.WriteLine(Thread.CurrentThread.ManagedThreadId);
    Thread.Sleep(1000); // slow processing of events
    Console.WriteLine("Latest: " + l);
});

ob.Subscribe(l =>
{
    Console.WriteLine(Thread.CurrentThread.ManagedThreadId);
    Thread.Sleep(1000); // slow processing of events
    Console.WriteLine("Latest1: " + l);
});