Rxjs异步播放队列的暂停与恢复问题求助

可以通过BehaviorSubject管理播放状态结合switchMap来优雅实现中断/恢复逻辑，中断期间自动忽略所有音频数据，恢复后重新启动串行播放队列，同时避免冗余的外部变量判断。

实现思路 #

1. 用BehaviorSubject维护播放状态（允许/中断），状态变化时自动切换音频处理逻辑
2. 中断时：暂停当前播放的音频、清理资源，同时断开对WebSocket音频流的处理
3. 恢复时：重新建立音频流的串行播放队列，从新接收的音频开始播放

完整代码 #

import { BehaviorSubject, EMPTY, switchMap, concatMap } from 'rxjs';

const audio = new Audio();
// 播放控制状态：true=允许播放，false=中断播放，初始值为允许
const playControl$ = new BehaviorSubject(true);

playControl$.pipe(
  switchMap(isAllowed => {
    // 中断状态：暂停当前音频、清理资源，返回空Observable忽略所有音频
    if (!isAllowed) {
      audio.pause();
      if (audio.src) {
        URL.revokeObjectURL(audio.src);
        audio.src = '';
      }
      return EMPTY;
    }

    // 允许播放状态：处理WebSocket音频流，保持串行播放
    return subject.pipe(
      concatMap(audioBlob => {
        return new Observable(resolve => {
          const objectUrl = URL.createObjectURL(audioBlob);
          audio.src = objectUrl;

          // 播放结束后清理资源并完成当前任务
          const onEnded = () => {
            URL.revokeObjectURL(objectUrl);
            audio.removeEventListener('ended', onEnded);
            resolve(audioBlob);
          };

          audio.addEventListener('ended', onEnded);

          // 处理播放失败（如浏览器自动播放限制）
          audio.play().catch(err => {
            URL.revokeObjectURL(objectUrl);
            audio.removeEventListener('ended', onEnded);
            resolve(audioBlob);
          });

          // 清理函数：取消订阅时移除事件监听
          return () => {
            audio.removeEventListener('ended', onEnded);
          };
        });
      })
    );
  })
).subscribe();

状态控制调用 #

• 开始录音（中断播放）：playControl$.next(false);
• 结束录音（恢复播放）：playControl$.next(true);

方案优势 #

• 逻辑集中：通过单一状态源管理播放状态，避免分散的变量判断
• 资源安全：中断时自动清理音频URL，避免内存泄漏
• 无队列阻塞：中断时直接断开音频流处理，不会留下pending的播放任务
• 自动忽略数据：中断期间WebSocket发送的音频会被EMPTY Observable直接忽略，无需额外过滤

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