如何使用Rx通知实现异常（含超时）后持续订阅数据

嘿，这个问题我太熟悉了！在Rx系列库（比如RxJava、RxJS）里，默认的流规则是一旦Observable抛出异常，整个订阅链就会终止，这确实是很多开发者踩过的坑。要实现超时后订阅者不停止接收数据，核心就是精准拦截超时异常，不让它终止整个流，下面给你几个实用的针对性方案：

场景1：整个流超时（指定时间内没收到任何数据） #

如果是因为Observable长时间没发射数据触发超时，导致订阅终止，你可以用retryWhen来捕获超时异常并重试订阅，让流能继续监听后续数据：

originalObservable
    .retryWhen(errors -> errors.flatMap(error -> {
        // 只处理超时异常
        if (error instanceof TimeoutException) {
            // 超时后延迟1秒重新订阅，也可以直接返回Observable.empty()立即重试
            return Observable.timer(1, TimeUnit.SECONDS);
        }
        // 其他异常正常抛出，终止流（避免吞掉真正的错误）
        return Observable.error(error);
    }))
    .subscribe(subscriber);

这个逻辑是：每当超时异常发生时，我们不终止流，而是等待一小段时间后重新订阅原Observable，这样后续发射的数据依然能被订阅者接收。

场景2：单个元素处理超时（处理某条数据耗时过长） #

如果是处理单个元素时触发超时（比如map里的业务逻辑耗时超了），那要把每个元素的处理包装成独立的子流，这样单个元素的超时不会影响整个主流：

originalObservable
    .flatMap(item -> 
        // 把每个元素包装成独立的Observable，单独处理超时
        Observable.just(item)
            .map(this::processHeavyTask) // 你的耗时处理方法
            .timeout(2, TimeUnit.SECONDS) // 单个元素的超时时间
            .onErrorResumeNext(error -> {
                if (error instanceof TimeoutException) {
                    // 处理单个元素超时：可以跳过、返回默认值或者记录日志
                    System.out.println("处理元素[" + item + "]超时，跳过该元素");
                    return Observable.empty(); // 跳过这个元素，流继续走
                    // 也可以返回Observable.just(defaultValue)给订阅者一个默认结果
                }
                return Observable.error(error);
            })
    )
    .subscribe(subscriber);

这里每个元素的处理都是独立的，哪怕某个元素超时失败，主流依然会继续处理下一个元素，订阅者不会被中断。

关键注意事项 #

• 一定要区分超时异常和其他异常：别把所有异常都吞掉，否则真正的业务错误会被隐藏，只针对TimeoutException做特殊处理。
• 根据你的实际场景选方案：流级超时用retryWhen，元素级超时用flatMap+独立子流。

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