如何在Angular & RxJS中实现丢弃旧元素的队列处理？

你的原始方案仅通过isProcessing标记是否正在处理，但未保存process运行期间收到的最新请求值，导致这些请求（包括最后一次）直接被丢弃，无法得到处理。以下是两种可靠的解决方案：

方案一：手动维护待处理状态（直观易理解） #

通过维护变量存储最新待处理值，在每次send时更新该值，当前process完成后自动检查并处理最新待办：

private pendingValue: any = null;
private isProcessing = false;

send(value: any) {
  // 每次调用都更新为最新待处理值，确保最后一次请求不丢失
  this.pendingValue = value;
  
  // 当前未处理时，启动处理流程
  if (!this.isProcessing) {
    this.processNext();
  }
}

private processNext() {
  // 无待处理值时，标记为非处理状态并退出
  if (this.pendingValue === null) {
    this.isProcessing = false;
    return;
  }

  this.isProcessing = true;
  // 取出当前待处理值并清空，避免重复处理
  const currentValue = this.pendingValue;
  this.pendingValue = null;

  process(currentValue).subscribe({
    next: (result) => {
      // 处理process返回结果
      // ...do something
    },
    complete: () => {
      // 处理完成后，递归检查是否有新的待处理值
      this.processNext();
    },
    error: (error) => {
      // 错误处理逻辑（可选）
      // ...handle error
      // 即使出错也要继续检查待处理值，避免阻塞后续请求
      this.processNext();
    }
  });
}

原理说明： #

• 每次send都会覆盖pendingValue为最新值，确保最后一次调用的内容被保留。
• processNext负责启动process，并在完成（或出错）后自动检查待处理值，实现"处理完当前任务后立即处理最新待办"的逻辑。
• 同一时间仅运行一个process实例，避免并发执行。

方案二：RxJS响应式实现（优雅简洁） #

利用RxJS的Subject和操作符，用响应式方式管理请求流：

import { Subject } from 'rxjs';
import { exhaustMap, tap } from 'rxjs/operators';

private valueSubject = new Subject<any>();

constructor() {
  this.valueSubject.pipe(
    // exhaustMap确保同一时间仅处理一个process，运行期间忽略新值
    // Subject会保留最后一次传入的值，当前process完成后自动处理该值
    exhaustMap((value) => {
      return process(value).pipe(
        tap((result) => {
          // 处理process返回结果
          // ...do something
        })
      );
    })
  ).subscribe({
    error: (error) => {
      // 全局错误处理（可选）
      // ...handle error
    }
  });
}

send(value: any) {
  // 每次send将值传入Subject
  this.valueSubject.next(value);
}

原理说明： #

• Subject作为请求中转站，收集所有send传入的值。
• exhaustMap在内部process运行时忽略外部新值，但Subject会保留最后一次传入的值，当前process完成后自动触发下一次处理。
• 全程通过响应式流管理状态，无需手动维护isProcessing和pendingValue变量。

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