RXJS如何使用OR条件：两个Observable的触发逻辑问题

你的问题核心是当前代码是串行等待两个流，必须先等WebSocket流过滤出符合条件的值，才会去发起HTTP请求，这和你“并行等待任一返回且满足条件就继续”的需求不符。要实现并行监听、取第一个符合条件的流，用RxJS的race操作符是最直接的方案。

修改后的代码 #

switchMap(() => {
  // 预处理WebSocket流：过滤出管理员且状态活跃的数据
  const validWebSocket$ = webSocket.access.pipe(
    filter(data => data.access.role === "ADMIN" && data.access.status === "ACTIVE")
  );

  // 预处理HTTP请求流：过滤出access为true的响应，同时处理请求失败的情况
  const validHttp$ = httpService.access.pipe(
    filter(res => res.access === true),
    catchError(() => EMPTY) // HTTP失败时返回空流，避免终止整个监听
  );

  // 用race并行监听两个流，取第一个发出符合条件值的流
  return race(validWebSocket$, validHttp$);
})
// 这里接后续需要执行的流程

关键说明 #

1. race操作符的作用：它会同时订阅传入的所有Observable，当其中任意一个Observable先发出值时，立即取消对其他Observable的订阅，并将这个值传递给下游。完美匹配你“任一请求返回且满足条件就继续”的需求。
2. 为什么原代码会停滞：原代码的两个switchMap是串行执行的，只有第一个switchMap里的WebSocket流发出符合条件的值后，才会进入第二个switchMap发起HTTP请求，相当于强制先等WebSocket，再等HTTP，完全不是并行等待。
3. HTTP错误处理：给HTTP流加上catchError(() => EMPTY)是为了避免HTTP请求失败时直接终止整个监听流程——如果HTTP失败，空流不会发出值，race会继续等待WebSocket的符合条件的值。

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