如何获取链式Observable发射的最后两个值并优化关联Observable订阅逻辑

现有实现的问题 #

• 变量名笔误：定义的流变量为first，首次订阅误用了未声明的source
• 冷流重复订阅风险：interval是冷Observable，两次独立订阅会触发两次独立的流执行，生成两个完全独立的发射序列，无法保证第二个订阅拿到的序列和第一个一致
• bufferCount(2)的边界缺陷：如果上游发射1个值后就抛出错误，缓存长度不足2时不会触发发射，无法拿到仅有的1个有效值+错误通知

最优实现方案 #

根据需求可选择以下两种方案：

方案1：多播共享流（适合两个逻辑独立解耦的场景） #

通过share操作符将冷流转为热流，所有订阅共享同一个发射序列，避免重复执行上游逻辑：

import { interval, share, materialize, bufferCount, last } from 'rxjs';
import { take } from 'rxjs/operators';

const first = interval(1000).pipe(
  take(5),
  share() // 核心：多播共享，多次订阅只执行一次上游流
);

// 处理单次发射值的响应逻辑
first.subscribe({
  next(response) {
    console.log("First: ", response);
    // 你的业务处理逻辑
  }
});

// 获取最后两个通知（含错误场景）
const second = first.pipe(
  materialize(), // 将所有next/error/complete事件转为Notification对象，错误不会中断流
  bufferCount(2, 1), // 滑动窗口，步长为1，每新增1个值就发射当前最新的2个值
  last() // 取流终止前的最后一个窗口
);

second.subscribe({
  next(notifications) {
    console.log("Second: ", notifications);
  }
});

方案2：单订阅整合逻辑（适合逻辑关联度高的场景，最简实现） #

不需要声明第二个Observable，直接在一次订阅中维护最近两个值的队列，逻辑更直观，性能开销更低：

import { interval, materialize } from 'rxjs';
import { take } from 'rxjs/operators';

const first = interval(1000).pipe(
  take(5),
  materialize()
);

// 存储最近两个通知的队列
const lastTwoNotifications: any[] = [];

first.subscribe({
  next(notification) {
    // 处理正常值的响应逻辑
    if (notification.kind === 'N') {
      console.log("First: ", notification.value);
      // 你的业务处理逻辑
    }
    // 更新最近两个通知队列
    lastTwoNotifications.push(notification);
    if (lastTwoNotifications.length > 2) {
      lastTwoNotifications.shift();
    }
  },
  complete() {
    console.log("Second: ", lastTwoNotifications);
  },
  error() {
    console.log("Second: ", lastTwoNotifications);
  }
});

关于flatMap/concat的疑问 #

你提到的flatMap（mergeMap）和concat都不适合当前场景：

• flatMap用于将上游发射的每个值映射为另一个Observable，再合并发射下游Observable的值
• concat用于按顺序依次订阅多个独立的Observable
  你的两个逻辑都是基于同一个上游流的不同处理，不需要使用这两个操作符。

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