
本文探讨了在rxjs管道中实现高效缓存和条件式api调用的策略,旨在避免使用外部状态,同时确保在输入参数未变时重放最新值,并在参数变化时触发新的异步操作。文章详细阐述了如何利用`scan`操作符结合`switchall`来构建一个内部状态管理机制,即使面对延迟的异步操作也能保持缓存的准确性和一致性,从而提供一个纯粹且健壮的rxjs解决方案。
需求场景与挑战
在构建响应式应用时,我们经常遇到这样的场景:需要根据输入参数的变化来触发异步操作(例如API调用),但如果参数没有改变,我们希望重用之前的结果,而不是重新发起请求。同时,无论参数是否变化,管道的下游都应该始终接收到最新的值。理想情况下,我们希望所有逻辑都封装在RxJS管道内部,避免使用外部变量来维护状态,以提高代码的可读性和纯粹性。
考虑一个具体例子:我们有一个上游Observable,它会根据用户交互(如点击)发出数值。我们希望:
- 当发出的数值与上一次不同时,触发一个新的“API调用”Observable(模拟异步操作)。
- 当发出的数值与上一次相同时,重放之前“API调用”的结果。
- 整个过程中,管道的订阅者始终能接收到最新的值。
最初,我们可能会尝试使用switchMap结合一个外部变量来缓存内部Observable,如下所示:
const { fromEvent, scan, map, startWith, pairwise, switchMap, tap, of, shareReplay, delay } = rxjs;
const events = [1, 1, 1, 1, 1, 2, 2, 3, 4, 4, 4, 4];
let cache$; // 外部状态变量
fromEvent(document, 'click')
.pipe(
scan((count) => count + 1, -1),
map((count) => events[count % events.length]),
startWith(undefined),
pairwise(), // 获取当前值和前一个值
switchMap(([prev, curr]) => {
if (prev === curr) {
// 如果值未变,返回缓存的Observable
return cache$;
}
// 如果值改变,创建新的Observable并缓存
cache$ = of(curr).pipe(
tap((val) => console.log(`New subscription for ${val}`)),
delay(1000), // 模拟异步延迟
shareReplay(1) // 缓存结果并重放给新订阅者
);
return cache$;
})
)
.subscribe((val) => console.log(`Received ${val}`));这段代码通过外部变量cache$实现了预期功能:只有当curr与prev不同时,才会创建新的内部Observable并模拟API调用(通过delay和tap)。shareReplay(1)确保了API调用完成后,其结果会被缓存,并在后续请求相同值时立即重放。然而,使用外部变量cache$打破了RxJS管道的纯粹性,降低了代码的可维护性和可测试性。
延迟异步操作带来的挑战
在上述场景中,内部Observable通常是异步的,例如模拟API调用的delay(1000)。如果内部Observable是同步的,我们可以考虑使用switchScan等操作符来管理内部状态。然而,当内部Observable存在延迟时,switchScan可能会导致问题。如果上游Observable在内部异步Observable完成之前发出了新值,switchScan会切换到新的内部Observable,可能导致之前正在进行的异步操作被取消,其结果无法被缓存,从而使缓存处于不确定或陈旧的状态。因此,我们需要一个能够稳健处理异步延迟的解决方案。
解决方案:利用 scan 和 switchAll 实现内部状态缓存
为了在不使用外部状态的情况下实现条件式缓存和重放,同时有效处理异步延迟,我们可以巧妙地结合使用scan和switchAll操作符。
核心思想:
- scan 维护内部状态: 使用scan来累积和存储内部Observable本身,而不是其发出的值。scan的累加器将持有上一个被创建或重用的内部Observable。
-
条件式创建/重用: 在scan的回调函数中,我们检查当前值curr是否与前一个值prev相同。
- 如果相同,scan就重用并发出累加器中已有的内部Observable。
- 如果不同,scan就创建一个新的内部Observable(包含shareReplay(1)来缓存其结果),并将其作为新的累加器值发出。
- switchAll 订阅最新Observable: scan发出的不是数据值,而是Observable。switchAll操作符的作用是订阅scan发出的最新Observable,并将其发出的值扁平化到主管道中。它会自动取消对前一个Observable的订阅,并切换到新的Observable。
下面是使用scan和switchAll实现的优化代码:
const { fromEvent, scan, map, startWith, pairwise, of, tap, delay, shareReplay, switchAll, EMPTY } = rxjs;
const events = [1, 1, 1, 1, 1, 2, 2, 3, 4, 4, 4, 4];
fromEvent(document, 'click')
.pipe(
scan((count) => count + 1, -1),
map((count) => events[count % events.length]),
startWith(undefined),
pairwise(), // [prev, curr]
scan(
(cachedObservable, [prev, curr]) => {
if (prev === curr && cachedObservable !== EMPTY) {
// 如果值未变且已有缓存Observable,则重用它
return cachedObservable;
}
// 如果值改变,或首次运行,创建新的Observable并作为缓存
return of(`${curr}:Result`).pipe(
tap((val) => console.log(`New subscription for ${val}`)),
delay(1000), // 模拟异步延迟
shareReplay(1) // 缓存结果并重放给新订阅者
);
},
EMPTY // scan的初始累加器值,表示没有缓存的Observable
),
switchAll() // 订阅并扁平化scan发出的最新Observable
)
.subscribe((val) => console.log(`Received ${val}`));点击页面任意位置,观察控制台输出
代码解析:
- pairwise(): 这一步与之前相同,用于获取当前值curr和前一个值prev。
-
scan((cachedObservable, [prev, curr]) => { ... }, EMPTY):
- 这是关键所在。scan的第一个参数是累加器函数,它接收当前的累加值(这里是cachedObservable,代表上一次缓存的Observable)和源Observable发出的值(这里是[prev, curr])。
- EMPTY是scan的初始累加值,表示开始时没有缓存的Observable。
- 在累加器函数内部:
- if (prev === curr && cachedObservable !== EMPTY):如果当前值与前一个值相同,并且我们已经有了一个缓存的cachedObservable,那么我们就直接返回这个cachedObservable。这意味着scan会再次发出这个旧的Observable。
- 否则(即prev !== curr或cachedObservable是EMPTY),我们创建一个新的Observable:of(${curr}:Result).pipe(tap(...), delay(1000), shareReplay(1))。这个新的Observable会被scan发出,并成为下一次迭代的cachedObservable。
- shareReplay(1)在这里至关重要。它确保了:
- 当内部Observable完成时,其结果会被缓存。
- 后续对这个Observable的订阅会立即收到缓存的结果,而不会重新执行delay(1000)和tap(即不会重新发起API调用)。
- 即使内部Observable仍在进行中(例如,正在delay),shareReplay(1)也会缓存其未来的结果,并将其提供给所有订阅者,包括在延迟期间切换过来的订阅者。
-
switchAll(): scan操作符现在发出的是Observable流(Observable
>)。switchAll订阅scan发出的每一个内部Observable,并将它们的值扁平化到主管道中。当scan发出一个新的Observable时,switchAll会自动取消对前一个Observable的订阅,并切换到新的Observable。由于shareReplay(1)的存在,即使切换发生在一个正在进行中的异步操作上,新的订阅者也能在操作完成后收到结果。
优点与注意事项
- 消除外部状态: 整个缓存逻辑完全封装在RxJS管道内部,提升了代码的纯粹性、可读性和可测试性。
- 处理异步延迟: scan结合shareReplay(1)和switchAll的模式能够稳健地处理内部Observable的异步延迟。shareReplay(1)确保了即使在切换过程中,异步操作的结果也能被正确缓存并重放。
- 高效的资源利用: 只有当参数真正改变时,才会创建新的内部Observable并触发潜在的API调用。对于重复的参数,直接重放缓存的结果。
- 灵活性: 这种模式可以很容易地扩展到更复杂的缓存策略,例如基于时间或大小的缓存。
注意事项:
- shareReplay(1) 的重要性: 如果没有shareReplay(1),每次scan发出相同的Observable时,switchAll都会重新订阅它,导致内部的异步操作(如API调用)被重复执行。shareReplay(1)是实现结果缓存的关键。
- EMPTY 初始值: 使用EMPTY作为scan的初始累加值是一个好的实践,它明确表示在管道开始时没有可用的缓存Observable。
- 理解 switchAll: switchAll是switchMap的泛化形式。当你的上游Observable发出的是Observable本身时,使用switchAll比使用switchMap(obs => obs)更简洁。
总结
通过巧妙地结合scan和switchAll操作符,我们成功地构建了一个RxJS管道,它能够在不引入任何外部状态的情况下,实现对异步操作的条件式缓存和重放。这种模式不仅解决了参数变化时触发新操作、参数不变时重放最新值的需求,还优雅地处理了内部异步操作可能带来的延迟问题。它体现了RxJS在处理复杂异步流和状态管理方面的强大能力,是编写纯粹、高效且可维护的响应式代码的典范。










