本文探讨了在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}`));<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/7.8.1/rxjs.umd.min.js"></script> <div>点击页面任意位置，观察控制台输出</div>
代码解析：
- 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<Observable<string>>）。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在处理复杂异步流和状态管理方面的强大能力，是编写纯粹、高效且可维护的响应式代码的典范。
