在c#中使用RX获取可观察序列中的最新项

Question

以以下内容为例:

var ob = Observable.Interval(TimeSpan.FromSeconds(1)).StartWith(500).Replay(1).RefCount();

我在这里想要实现的是在任何给定时间"同步"获取序列中最新项的值。这意味着像FirstAsync这样的扩展不能为我弥补。

StartWith和Replay位确保始终存在一个值，并且RefCount位在我的实际代码中是必要的，以检测何时可以执行一些处置操作。

因此，为了模拟这个"任意给定时间"部分，让我们尝试获取5秒后的最新值:

Observable.Timer(TimeSpan.FromSeconds(5)).Subscribe(x =>
{
    // Try to get latest value from "ob" here.
});

所以有5秒的延迟，我需要从序列中获取值5，这些是我到目前为止没有成功的尝试:

1. ob.First() -返回500
2. ob.Latest().Take(1) -同上
3. ob.MostRecent(-1).First() -同上
4. ob.MostRecent(-1) -给我一个充满"500"的IEnumerable<long>
5. ob.Last() -永远不会返回，因为它正在等待序列完成，而它永远不会返回
6. ob.Latest().Last() -同上
7. ob.ToTask().Result -同上
8. ob.ToEnumerable() -同上
9. ob.MostRecent().Last()同上

似乎没有太多的资源可以让人们真正做到这一点。我能找到的最接近的是:"Rx:从Observable流中获取第一个和最近的值的操作符"，但它毕竟不是一个同步调用(仍然使用订阅)，所以它不适合我。

有人知道这是否真的可行吗?

Answer 1

指出为什么你的代码可能没有像你期望的那样工作

var ob = Observable.Interval(TimeSpan.FromSeconds(1)).StartWith(500).Replay(1).RefCount();
//Note at this point `ob` has never been subscribed to,
// so the Reference-count is 0 i.e. has not be connected.
Observable.Timer(TimeSpan.FromSeconds(5)).Subscribe(x =>
{
    // Try to get latest value from "ob" here.
    //Here we make our first subscription to the `ob` sequence.
    //  This will connect the sequence (invoke subscribe)
    //   which will
    //      1) invoke StartWith
    //      2) invoke onNext(500)
    //      3) invoke First()
    //      4) First() will then unsubscribe() as it has the single value it needs
    //      5) The refCount will now return to 0
    //      6) The sequence will be unsubscribed to.
    ob.First().Dump();  
    //Any future calls like `ob.First()` will thus always get the value 500.
});

可能你想要的是

var ob = Observable.Interval(TimeSpan.FromSeconds(1))
    .Publish(500);
var connection = ob.Connect();
//Note at this point `ob` has never been subscribed to, so the ReferenceCount is 0 i.e. has not be connected.
var subscription = Observable.Timer(TimeSpan.FromSeconds(5)).Subscribe(x =>
{
    // Try to get latest value from "ob" here.
    ob.First().Dump();
});
//Sometime later
subscription.Dispose();
connection.Dispose()

然而，您确实不希望将同步调用与Rx混合在一起。您通常也不希望在订阅中订阅(因为.First()是一个订阅)。您可能想要做的是获取最新的值，并将其存储在某个地方。使用.First()只是一个滑坡。你最好写一些像

这样的东西

var subscription = Observable.Timer(TimeSpan.FromSeconds(5))
    .SelectMany(_=>ob.Take(1))
    .Subscribe(x =>
    {
        //Do something with X here.
        x.Dump();
    }); 

Answer 2

您需要这样做:

var ob = Observable.Interval(TimeSpan.FromSeconds(1)).StartWith(500);
var latestAndThenTheRest =
    Observable
        .Create<long>(o =>
        {
            var bs = new BehaviorSubject<long>(1);
            var s1 = ob.Subscribe(bs);
            var s2 = bs.Subscribe(o);
            return new CompositeDisposable(s1, s2);
        });

你需要考虑的唯一一件事是，ob必须是一个热门的可观察对象，这样才有意义。如果它是冷的，那么每个订阅者将获得一个全新的订阅到ob序列的开始。

Answer 3

澄清一下，感谢@LeeCampbell的回答。

什么不工作:

var ob = Observable.Interval(TimeSpan.FromSeconds(1)).StartWith(500).Replay(1).RefCount();
Observable.Timer(TimeSpan.FromSeconds(5)).Subscribe(x =>
{
    ob.First().Dump();
    // This gives you 500.
    // Because this is the first time any one subscribes to the observable,
    // so it starts right here and gives you the initial value.
});

实际操作:

var ob = Observable.Interval(TimeSpan.FromSeconds(1)).StartWith(500).Replay(1).RefCount();
ob.Subscribe(); // Subscribe to start the above hot observable immediately.
Observable.Timer(TimeSpan.FromSeconds(5)).Subscribe(x =>
{
    ob.First().Dump(); 
    // This would give you either 3 or 4, depending on the speed and timing of your computer.
});

Answer 4

我不确定这个答案是否对你有帮助，但是你看过行为学科吗?它是一个IObservable，会记住它的最新值。它有点像一个常规变量和一个可观察对象的组合。

否则，为什么不订阅'ob'并将最新的值存储在一个变量中呢?