Flutter中使用StreamTransformer优雅处理Stream错误恢复

Stream错误处理的常见思路

在使用Stream构建异步数据流时，如果中途发生异常，默认的`catchError`通常会直接中断整个Stream。虽然可以使用try-catch块包裹订阅逻辑，但在需要特定数据节点失败时提供降级默认值（如继续运行而非崩溃）的场景下，代码往往变得冗长且难以维护。

为了简化这一过程，我们可以将错误处理逻辑封装为独立的StreamTransformer，并结合Dart的扩展语法，使其使用更加直观。

封装StreamErrorHandle转换器

首先，我们需要定义一个泛型类StreamErrorHandler<T>，它继承自StreamTransformerBase<T, T>。该转换器的核心目标是拦截Stream中的错误，并通过回调函数获取替代值。

在初始化时，我们创建一个广播类型的StreamController用于管理内部状态，同时接收一个错误处理回调onError。当错误发生时，如果回调返回了非空值，我们将该值添加到输出流中。

class StreamErrorHandler<T> extends StreamTransformerBase<T, T> {
    final _controller = StreamController<T>.broadcast();
    final T? Function(Object error) onError;
    
    StreamErrorHandler({
        required this.onError,
    });
    
    @override
    Stream<T> bind(Stream<T> stream) {
        // 实现细节见下文
    }
}

实现bind方法监听错误

在bind方法中，我们利用Dart标准库提供的handleError方法来捕获源Stream抛出的异常。具体步骤如下：

1. 调用stream.handleError((error) {...})注册错误处理器。
2. 在处理器中调用传入的onError回调函数。
3. 如果onError返回了有效值(value != null)，则将该值写入_controller.sink。
4. 监听原始Stream的正常数据，并将其转发至_controller.sink.add。
5. 设置取消监听逻辑，确保资源释放。

@override
Stream<T> bind(Stream<T> stream) {
    final sub = stream.handleError((error) {
        final value = onError(error);
        if (value != null) {
            _controller.sink.add(value);
        }
    }).listen(_controller.sink.add);
    
    _controller.onCancel = (){
        sub.cancel();
    };
    
    return _controller.stream;
}

通过Extension简化调用

为了让API更易用，我们为Stream<T>创建一个扩展方法onErrorRecoverWith。该方法接受一个错误处理函数，并应用之前定义的StreamErrorHandler转换器。

extension Recover<T> on Stream<T> {
    Stream<T> onErrorRecoverWith(T? Function(Object error) onError) => transform(StreamErrorHandler<T>(onError: onError));
}

实际测试示例

以下是一个完整的测试场景。模拟一个异步流getNames，在产生前两个数据后抛出异常。通过onErrorRecoverWith，我们在捕获到异常后返回字符串'Baz'作为默认值，从而使循环能够完整执行完毕。

Stream<String> getNames() async* {
    yield 'Foo';
    yield 'Bar';
    throw Exception('Something went wrong');
}

Future<void> testIt() async {
    final names = getNames().onErrorRecoverWith(
        (error) {
            print(error); // 记录日志
            return 'Baz';  // 返回默认值以恢复流
        },
    );
    
    await for(final name in names){
        print(name); // 预期输出: Foo, Bar, Baz
    }
}

执行上述代码后，控制台将按顺序打印出 Foo、Bar 和 Baz，证明Stream在错误发生后成功恢复了运行。

辅助工具：简易Log扩展

在上述示例中，使用了print(error)来查看异常信息。在实际工程中，为了更清晰地标识日志来源，可以参考以下封装的log()扩展方法（基于dart:developer），它能提供更完善的调试功能：

import 'dart:developer' as devtools show log;
extension Log on Object {
    void log() => devtools.log(toString());
}

以上就是Flutter Recovering Stream Errors小技巧的详细内容，更多关于Flutter Stream错误处理与扩展方法的资料请关注本站其它相关文章！