时间戳

首页 > 解决方案 > .NET ReactiveExtensions：使用具有可变时间跨度的 Sample()

c# - .NET ReactiveExtensions：使用具有可变时间跨度的 Sample()

问题描述

给定一个高频可观察的数据流，我只想每 XX 秒发出一个项目。

这通常在 RX 中使用 .Sample(TimeSpan.FromSeconds(XX)) 完成

但是......我希望时间间隔根据数据的某些属性而变化。

假设我的数据是：

类位置 { ... public int 速度；}

如果速度小于 100，我想每 5 秒发出一次数据。如果速度高于 100，则应每 2 秒一次。

使用现成的 Sample() 是否可行，还是我需要自己构建一些东西？

标签： c#system.reactiverx.net

解决方案

这是一个低级实现，将System.Reactive.Concurrency.Scheduler.SchedulePeriodic扩展方法用作计时器。

public static IObservable<TSource> Sample<TSource>(this IObservable<TSource> source,
    Func<TSource, TimeSpan> intervalSelector, IScheduler scheduler = null)
{
    if (source == null) throw new ArgumentNullException(nameof(source));
    if (intervalSelector == null)
        throw new ArgumentNullException(nameof(intervalSelector));
    scheduler = scheduler ?? Scheduler.Default;

    return Observable.Create<TSource>(observer =>
    {
        TimeSpan currentInterval = Timeout.InfiniteTimeSpan;
        IDisposable timer = null;
        TSource latestItem = default;
        bool latestEmitted = true;
        object locker = new object();

        Action periodicAction = () =>
        {
            TSource itemToEmit;
            lock (locker)
            {
                if (latestEmitted) return;
                itemToEmit = latestItem;
                latestItem = default;
                latestEmitted = true;
            }
            observer.OnNext(itemToEmit);
        };

        return source.Subscribe(onNext: item =>
        {
            lock (locker)
            {
                latestItem = item;
                latestEmitted = false;
            }
            var newInterval = intervalSelector(item);
            if (newInterval != currentInterval)
            {
                timer?.Dispose();
                timer = scheduler.SchedulePeriodic(newInterval, periodicAction);
                currentInterval = newInterval;
            }
        }, onError: ex =>
        {
            timer?.Dispose();
            observer.OnError(ex);
        }, onCompleted: () =>
        {
            timer?.Dispose();
            observer.OnCompleted();
        });
    });
}

使用示例：

observable.Sample(x => TimeSpan.FromSeconds(x.Speed < 100 ? 5.0 : 2.0));

每次intervalSelector回调返回不同的时间间隔时，计时器都会重新启动。在每个新项目都会更改间隔的极端情况下，此自定义运算符的行为将更像内置而Throttle不是内置Sample。

与 不同Sample，Throttle的周期是一个滑动窗口。每次Throttle接收到一个值，窗口都会被重置。（引用）

推荐阅读