基于下游条件的RX过滤：用现有操作符替代可变变量实现串行处理

嘿，你完全不用靠可变变量来折腾这个逻辑！Rx里正好有个专门的操作符exhaustMap，完美匹配你想要的行为——仅当当前内部Observable完成后，才会处理下一个上游项，期间的上游项直接被过滤忽略。

为什么exhaustMap是最优解？ #

它的核心逻辑就是：一旦订阅了某个内部Observable（也就是你通过createProcess启动的外部进程输出流），就会暂时忽略上游的所有新发射项，直到这个内部Observable完成（进程执行结束），之后才会响应下一个上游参数。完全不需要手动维护processNotRunning这种可变状态，纯函数式实现，更安全也更符合Rx的设计思想。

替换后的极简代码 #

直接把你原来的map + switch换成exhaustMap就行：

let mapProcess obs = obs |> Observable.exhaustMap createProcess

对比其他操作符的区别（帮你理清适用场景） #

为了避免混淆，顺便说下几个类似操作符的差异：

• switchMap：你的初始示例用的就是这个，它会取消当前运行的内部Observable，立即切换到新的，这和你要“等待当前进程完成再处理后续”的需求完全相反
• concatMap：它会把上游的项排队等待，前一个内部Observable完成后再执行下一个，而你要的是直接过滤掉进程运行期间的上游项，不是排队，所以exhaustMap才是正确选择

验证行为逻辑 #

用exhaustMap的话，整个流程会是：

1. 上游发射第一个参数，exhaustMap调用createProcess启动进程，订阅其输出流
2. 进程运行期间，上游发射的任何新参数都会被直接忽略
3. 进程执行完成（输出流结束），exhaustMap恢复监听上游，准备处理下一个参数

这样既满足了“等待进程完成再处理后续”的要求，又完全避免了可变状态带来的潜在问题（比如并发场景下的竞态条件）。

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