RxJS实现：基于前Observable最新状态切换新数据流

嘿，这个场景我太熟了！你一开始想到用scan+高阶流的思路方向完全没错，不过卡在了怎么获取前一个流的最新状态上对吧？我给你两个方案，首选那个超直观的，顺便也补一下你想的scan方案~

首选方案：用BehaviorSubject维护状态 + switchMap切换流 #

这个方案最容易理解，完美匹配你的需求——每次新请求来的时候，自动停掉之前的流，用最新状态+新请求创建新流，同时实时更新状态：

完整代码示例 #

import { Subject, BehaviorSubject, interval } from 'rxjs';
import { switchMap, tap } from 'rxjs/operators';

// 初始化状态和请求流
const initialState = 10;
const request$ = new Subject<number>();

// 你定义的stream函数（简化版）
function stream(request: number, state: number) {
  return interval(1000).pipe(
    map(i => state + (i + 1) * request)
  );
}

// 用BehaviorSubject维护全局最新状态——它的核心优势是随时能通过.value拿到当前最新值
const latestState$ = new BehaviorSubject(initialState);

// 核心逻辑：处理请求流
const output$ = request$.pipe(
  // switchMap会自动取消前一个流的订阅，正好符合"新请求来就停掉旧流"的需求
  switchMap(request => {
    // 拿到当前最新的状态值
    const currentState = latestState$.value;
    // 创建新的状态流
    return stream(request, currentState).pipe(
      // 每次新状态发出时，更新全局最新状态
      tap(newState => latestState$.next(newState))
    );
  })
);

// 测试订阅输出
output$.subscribe(val => console.log('输出:', val));

// 模拟请求发送：0秒发2，3.5秒发5
setTimeout(() => request$.next(2), 0);
setTimeout(() => request$.next(5), 3500);

为什么这个方案可行？ #

1. BehaviorSubject：它会一直保存最新的状态值，不管什么时候都能通过.value直接获取，完美解决你之前拿不到前一个流最新状态的问题。
2. switchMap：每次新请求进来时，它会自动取消上一个stream的订阅，同时订阅新创建的流，正好实现“替换流”的需求。
3. tap操作符：在新流的每个状态输出时，同步更新latestState$，确保下一次请求进来时能拿到最准确的最新状态。

运行这段代码，输出完全符合你的预期：

输出: 12
输出: 14
输出: 16
输出: 21
输出: 26
输出: 31
...

补充方案：用scan实现你的初始思路 #

如果你坚持想用scan结合高阶流的方式，我们可以调整scan的积累值，让它同时跟踪最新状态和当前流，代码如下：

import { Subject, interval, of } from 'rxjs';
import { scan, switchAll, tap } from 'rxjs/operators';

const initialState = 10;
const request$ = new Subject<number>();

function stream(request: number, state: number) {
  return interval(1000).pipe(
    map(i => state + (i + 1) * request)
  );
}

const output$ = request$.pipe(
  // scan的积累值是一个对象：包含最新状态和当前流
  scan((acc, request) => {
    // 用当前最新状态创建新流
    const newStream$ = stream(request, acc.latestState);
    // 订阅新流，实时更新最新状态
    newStream$.subscribe(newState => acc.latestState = newState);
    return {
      stream$: newStream$,
      latestState: acc.latestState
    };
  }, {
    stream$: of(initialState),
    latestState: initialState
  }),
  // 切换到最新的流，自动取消之前的流
  switchMap(acc => acc.stream$)
);

// 测试代码和之前一致
output$.subscribe(val => console.log('输出:', val));
setTimeout(() => request$.next(2), 0);
setTimeout(() => request$.next(5), 3500);

这个方案的核心是让scan积累的不是单纯的流，而是一个包含状态和流的对象，这样就能在每次新请求进来时，直接用积累的最新状态创建新流，同时通过订阅新流来更新状态。

内容的提问来源于stack exchange，提问作者Joe Daley