如何在不丢弃 RX 中的值的情况下减慢 Observable 的速度？答案

【问题标题】：How can i slow down an Observable without throwing away values in RX?如何在不丢弃 RX 中的值的情况下减慢 Observable 的速度？
【发布时间】：2015-03-09 09:22:12
【问题描述】：

我的场景：我有一个应该每秒运行一次的计算。运行后，应该等待大约 200 毫秒，以便其他东西赶上。如果计算在一秒钟后仍在运行，则应再次启动，但程序应等到完成并在完成后 200 毫秒开始下一次计算。

我现在的做法：

_refreshFinished = new Subject<bool>();
_autoRefresher = Observable.Interval(TimeSpan.FromMilliseconds(1000))
   .Zip(_refreshFinished, (x,y) => x)
   .Subscribe(x => AutoRefresh(stuff));

这段代码的问题是，我认为在计算完成后没有办法延迟。 Delay 方法仅延迟可观察集合的第一个元素。通常这种行为是正确的一次，因为如果你想缓冲每个人，你将不得不缓冲无限数量的元素，但是由于延迟对 Autorefesh 的调用 200 毫秒也会延迟 _refreshFinished 的输出 200 毫秒，因此不会有缓冲开销. 基本上我想要一个 Observable，它每 MaxTime(some_call,1000ms) 触发一次，然后延迟 200ms 甚至更好，一些动态值。在这一点上，我什至并不真正关心贯穿其中的价值观，尽管将来可能会改变。

我愿意接受任何建议

【问题讨论】：

标签： c# .net system.reactive

【解决方案1】：

Observable.Generate() 有许多重载，可让您动态调整创建下一个项目的时间。

例如

IScheduler schd = Scheduler.TaskPool;
var timeout = TimeSpan.FromSeconds(1);
var shortDelay = TimeSpan.FromMilliseconds(200);
var longerDelay = TimeSpan.FromMilliseconds(500);
Observable.Generate(schd.Now, 
                    time => true, 
                    time => schd.Now, 
                    time => new object(), // your code here
                    time => schd.Now.Subtract(time) > timeout  ? shortDelay : longerDelay ,
                    schd);

【讨论】：

您在倒数第二行有错字。它应该是超时而不是超时。但我真的很喜欢你的解决方案，它比我在下面发布的要简单得多

【解决方案2】：

这听起来更像是新的异步框架 http://msdn.microsoft.com/en-us/vstudio/gg316360 的工作

【讨论】：

请详细说明为什么这是新异步框架的工作？
由于他基本上是在等待一些计算完成，听起来这样做会更容易，等到那个完成。此外，在我看来，“等待其他东西赶上来”听起来像是等待多个异步作业完成，这正是异步框架的用途。如果作业没有在给定时间内完成，只需抛出一些超时并相应地处理它们。

【解决方案3】：

有办法做到这一点。这不是最简单的事情，因为必须根据每个值动态计算等待时间，但它可以工作并且非常通用。

当您使用此代码时，您只需在 YOURCODE 中插入应该调用的代码，其他一切都会自动运行。基本上每个 Max(yourCodeTime+extraDelay,usualCallTime+extraDelay) 都会调用您的代码。这意味着 yourCode 不会同时被调用两次，并且应用程序将始终有额外的延迟时间来执行其他操作。如果有一些更简单/其他的方法可以做到这一点，我很乐意听到它。

double usualCallTime = 1000;
double extraDealy = 100;
var subject = new Subject<double>();
var subscription =
    sub.TimeInterval()
        .Select(x =>
            {
                var processingTime = x.Interval.TotalMilliseconds - x.Value;
                double timeToWait = 
                     Math.Max(0, usualCallTime - processingTime) + extraDelay;
                return Observable.Timer(TimeSpan.FromMilliseconds(timeToWait))
                    .Select(ignore => timeToWait);
            })
        .Switch()
        .Subscribe(x => {YOURCODE();sub.OnNext(x)});
sub.OnNext(0);

private static void YOURCODE()
{
    // do stuff here
    action.Invoke();
}

【讨论】：

【解决方案4】：

如果我正确理解了您的问题，那么您有一个长期运行的计算功能，例如：

static String compute()
{
    int t = 300 + new Random().Next(1000);
    Console.Write("[{0}...", t);
    Thread.Sleep(t);
    Console.Write("]");
    return Guid.NewGuid().ToString();
}

并且您希望每秒至少调用一次此函数，但没有重叠调用，并且调用之间的恢复时间至少为 200 毫秒。下面的代码适用于这种情况。

我从更实用的方法开始（使用 Scan() 和 Timestamp()），更多的是 Rx 的风格——因为我正在寻找一个好的 Rx 练习——但最后，这种非聚合方法更简单。

static void Main()
{
    TimeSpan period = TimeSpan.FromMilliseconds(1000);
    TimeSpan recovery = TimeSpan.FromMilliseconds(200);

    Observable
        .Repeat(Unit.Default)
        .Select(_ =>
        {
            var s = DateTimeOffset.Now;
            var x = compute();
            var delay = period - (DateTimeOffset.Now - s);
            if (delay < recovery)
                delay = recovery;

            Console.Write("+{0} ", (int)delay.TotalMilliseconds);

            return Observable.Return(x).Delay(delay).First();
        })
        .Subscribe(Console.WriteLine);
}

这是输出：

[1144...]+200 a7cb5d3d-34b9-4d44-95c9-3e363f518e52
[1183...]+200 359ad966-3be7-4027-8b95-1051e3fb20c2
[831...]+200 f433b4dc-d075-49fe-9c84-b790274982d9
[766...]+219 310c9521-7bee-4acc-bbca-81c706a4632a
[505...]+485 0715abfc-db9b-42e2-9ec7-880d7ff58126
[1244...]+200 30a3002a-924a-4a64-9669-095152906d85
[1284...]+200 e5b1cd79-da73-477c-bca0-0870f4b5c640
[354...]+641 a43c9df5-53e8-4b58-a0df-7561cf4b0483
[1094...]+200 8f25019c-77a0-4507-b05e-c9ab8b34bcc3
[993...]+200 840281bd-c8fd-4627-9324-372636f8dea3

[编辑：此示例使用 Rx 2.0(RC) 2.0.20612.0]

【讨论】：

【解决方案5】：

假设您有一个现有的 'IObservable' ，那么以下将起作用

var delay = TimeSpan.FromSeconds(1.0);
var actual = source.Scan(
    new ConcurrentQueue<object>(),
    (q, i) =>
        {
            q.Enqueue(i);
            return q;
        }).CombineLatest(
            Observable.Interval(delay),
            (q, t) =>
                {
                    object item;
                    if (q.TryDequeue(out item))
                    {
                        return item;
                    }

                    return null;
                }).Where(v => v != null);

'actual' 是你的结果 observable。但是请记住，如果上面的代码还不是很热的话，它已经把它变成了一个热的可观察对象。所以你不会被调用'OnCompleted'。

【讨论】：