javascript - 如何在 RxJS（或一般在响应式扩展中）实现时间过期的热观察者

我想用 RxJs 实现带有时间过期的缓存。这里是一个“普通”缓存的例子：

//let this represents "heavy duty job"
var data = Rx.Observable.return(Math.random() * 1000).delay(2000);

//and we want to cache result
var cachedData = new Rx.AsyncSubject();
data.subscribe(cachedData);

cachedData.subscribe(function(data){
    //after 2 seconds, result is here and data is cached
    //next subscribe returns immediately data
    cachedData.subscribe(function(data2){ /*this is "instant"*/ });
});

当第一次对 cachedData 调用 subscribe 时，会调用“重负荷任务”，并且在 2 秒后结果会保存在 cachedData（AsyncSubject）中。任何其他后续对 cachedData 的 subscribe 调用都会立即返回保存的结果（因此实现了缓存）。

我想要实现的是在cachedData有效的时间段内给它“加点料”，当该时间段结束后，我想重新运行“重度任务”以获取新数据，并再次缓存以供新的时间段使用，等等……

期望的行为：

//pseudo code
cachedData.youShouldExpireInXSeconds(10);


//let's assume that all code is sequential from here

//this is 1.st run
cachedData.subscribe(function (data) {
    //this first subscription actually runs "heavy duty job", and
    //after 2 seconds first result data is here
});

//this is 2.nd run, just after 1.st run finished
cachedData.subscribe(function (data) {
    //this result is cached
});

//15 seconds later
// cacheData should expired
cachedData.subscribe(function (data) {
    //i'm expecting same behaviour as it was 1.st run:
    // - this runs new "heavy duty job"
    // - and after 2 seconds we got new data result
});


//....
//etc

我刚接触 Rx(Js)，无法弄清楚如何实现带有冷却时间的热可观察对象。

答案＃1

你所缺少的只是安排一个任务，在一段时间后将你的 cachedData 替换为一个新的 AsyncSubject。以下是将其作为新的 Rx.Observable 方法来实现的方法：

Rx.Observable.prototype.cacheWithExpiration = function(expirationMs, scheduler) {
    var source = this,
        cachedData = undefined;

    // Use timeout scheduler if scheduler not supplied
    scheduler = scheduler || Rx.Scheduler.timeout;

    return Rx.Observable.create(function (observer) {

        if (!cachedData) {
            // The data is not cached.
            // create a subject to hold the result
            cachedData = new Rx.AsyncSubject();

            // subscribe to the query
            source.subscribe(cachedData);

            // when the query completes, start a timer which will expire the cache
            cachedData.subscribe(function () {
                scheduler.scheduleWithRelative(expirationMs, function () {
                    // clear the cache
                    cachedData = undefined;
                });
            });
        }

        // subscribe the observer to the cached data
        return cachedData.subscribe(observer);
    });
};

用法：

// a *cold* observable the issues a slow query each time it is subscribed
var data = Rx.Observable.return(42).delay(5000);

// the cached query
var cachedData = data.cacheWithExpiration(15000);

// first observer must wait
cachedData.subscribe();

// wait 3 seconds

// second observer gets result instantly
cachedData.subscribe();

// wait 15 seconds

// observer must wait again
cachedData.subscribe();

答案＃2

一个简单的解决方案是创建一个自定义可管道操作符，在经过一段时间后 repeatWhen。这是我想出来的办法：

 export const refreshAfter = (duration: number) => (source: Observable<any>) =>
                                  source.pipe(
                                     repeatWhen(obs => obs.pipe(delay(duration))),
                                     publishReplay(1), 
                                     refCount());

然后我这样使用它：

const serverTime$ = this.environmentClient.getServer().pipe(map(s => s.localTime))
const cachedServerTime$ = serverTime.pipe(refreshAfter(5000)); // 5s cache

重要提示：这里使用了 publishReplay(1)、refCount()，因为 shareReplay(1) 不会取消订阅源可观察对象，所以它会一直访问你的服务器。不幸的是，这会导致一个后果，即在出现错误时，该错误将从 publishReplay(1)、refCount() 重播。很快会有一个“新改进”的 shareReplay。关于类似问题的说明见这里。一旦这个“新”版本可用，这个答案应该更新——但自定义操作符的优点是你可以在一个地方修复它们。