RxJS链中如何在保留订阅的同时让错误冒泡？

要同时实现订阅持续活跃和错误冒泡至全局处理器，核心是把错误处理逻辑下沉到concatMap内部的HTTP请求流中，避免用全局retry吞掉错误。具体实现如下：

核心思路 #

将catchError操作符放在每个HTTP请求的Observable管道内：

1. 捕获单个请求的错误，调用全局错误处理器完成上报
2. 返回一个不会发射错误的空流（比如EMPTY），避免错误向上传递导致主订阅终止

修改后的代码示例 #

import { EMPTY } from 'rxjs';
import { catchError, concatMap, tap } from 'rxjs/operators';

const onEditSubscription = onEdit$.pipe(
    tap(x => console.log("on Edit")),
    concatMap((data) => 
        this.svc.getAggregate(data as number).pipe(
            catchError((err) => {
                // 调用你的全局错误处理器，比如：
                // this.globalErrorHandler.handleError(err);
                console.log("错误已上报至全局处理器", err);
                // 返回EMPTY，让当前请求流正常结束，不终止主订阅
                return EMPTY;
            })
        )
    )
).subscribe({
    next : (data) => console.log(data),
    error : (err) => console.log("error") // 仅onEdit$自身的错误会触发这里
});

可选：给单个请求添加重试逻辑 #

如果需要对失败的HTTP请求进行重试，只需在catchError前添加retry操作符即可——重试仅作用于当前请求，不会影响主订阅的活跃性：

concatMap((data) => 
    this.svc.getAggregate(data as number).pipe(
        retry(3), // 单个请求最多重试3次
        catchError((err) => {
            this.globalErrorHandler.handleError(err);
            return EMPTY;
        })
    )
)

原理说明 #

• 原代码中全局的retry会捕获所有错误并重试整个流，导致错误无法传递到下游的error回调
• 将错误处理放在concatMap内部后，每个HTTP请求的错误被局部捕获处理，主流（onEdit$的订阅）不会因单个请求失败而终止
• 返回EMPTY确保当前请求流正常完成，不会向上抛出错误，同时主订阅保持活跃

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