RxJS管道中无外部状态的条件式缓存与重放策略

本文探讨了在rxjs管道中实现高效缓存和条件式api调用的策略，旨在避免使用外部状态，同时确保在输入参数未变时重放最新值，并在参数变化时触发新的异步操作。文章详细阐述了如何利用`scan`操作符结合`switchall`来构建一个内部状态管理机制，即使面对延迟的异步操作也能保持缓存的准确性和一致性，从而提供一个纯粹且健壮的rxjs解决方案。

需求场景与挑战

在构建响应式应用时，我们经常遇到这样的场景：需要根据输入参数的变化来触发异步操作（例如API调用），但如果参数没有改变，我们希望重用之前的结果，而不是重新发起请求。同时，无论参数是否变化，管道的下游都应该始终接收到最新的值。理想情况下，我们希望所有逻辑都封装在RxJS管道内部，避免使用外部变量来维护状态，以提高代码的可读性和纯粹性。

考虑一个具体例子：我们有一个上游Observable，它会根据用户交互（如点击）发出数值。我们希望：

1. 当发出的数值与上一次不同时，触发一个新的“API调用”Observable（模拟异步操作）。
2. 当发出的数值与上一次相同时，重放之前“API调用”的结果。
3. 整个过程中，管道的订阅者始终能接收到最新的值。

最初，我们可能会尝试使用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操作符。

有道智云AI开放平台

有道智云AI开放平台

16

查看详情

核心思想：

1. scan 维护内部状态： 使用scan来累积和存储内部Observable本身，而不是其发出的值。scan的累加器将持有上一个被创建或重用的内部Observable。
2. 条件式创建/重用： 在scan的回调函数中，我们检查当前值curr是否与前一个值prev相同。
   • 如果相同，scan就重用并发出累加器中已有的内部Observable。
   • 如果不同，scan就创建一个新的内部Observable（包含shareReplay(1)来缓存其结果），并将其作为新的累加器值发出。
3. 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>

代码解析：

1. pairwise()： 这一步与之前相同，用于获取当前值curr和前一个值prev。
2. 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)也会缓存其未来的结果，并将其提供给所有订阅者，包括在延迟期间切换过来的订阅者。
3. 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在处理复杂异步流和状态管理方面的强大能力，是编写纯粹、高效且可维护的响应式代码的典范。

以上就是RxJS管道中无外部状态的条件式缓存与重放策略的详细内容，更多请关注php中文网其它相关文章！