基于Riverpod的异步请求队列执行器及循环依赖疑问

核心问题分析 #

你当前的实现思路方向是对的，但确实存在循环依赖的潜在风险——当队列中的任务调用了依赖于futureExecutor的Provider时，可能触发循环引用导致异常。另外，当前asyncExecute在任务失败后直接返回，后续任务会被卡住，不符合「失败后等待重试再执行后续」的需求。

优化后的实现方案 #

1. 重构FutureExecutor为独立Notifier #

避免直接传递外部Provider的ref，改为内部管理执行逻辑，通过自身的ref读取其他Provider，消除隐式依赖：

@riverpod
class FutureExecutor extends _$FutureExecutor {
  final Queue<Future<void> Function(AsyncNotifierProviderRef ref)> _queue = Queue();
  bool _isExecuting = false;

  @override
  FutureOr<void> build() {
    // 初始化无需返回业务状态，仅用于持有自身ref
    return;
  }

  void addTask(Future<void> Function(AsyncNotifierProviderRef ref) task) {
    _queue.addLast(task);
    if (!_isExecuting) {
      _executeQueue();
    }
  }

  Future<void> _executeQueue() async {
    _isExecuting = true;
    while (_queue.isNotEmpty) {
      try {
        await _queue.first(ref);
        _queue.removeFirst();
      } catch (e, stacktrace) {
        ref.read(errorProvider.notifier).setError("FutureExecutor._executeQueue", e, stacktrace);
        // 等待重试信号（比如用户触发重试操作）
        await _waitForRetry();
        // 不移除当前任务，循环会自动重试
      }
    }
    _isExecuting = false;
  }

  Future<void> _waitForRetry() {
    // 通过监听重试信号Provider实现等待逻辑
    return ref.watch(retrySignalProvider.future);
  }
}

// 配套的重试信号Provider
@riverpod
class RetrySignal extends _$RetrySignal {
  @override
  bool build() => false;

  void triggerRetry() {
    state = !state; // 通过状态变化触发future完成
  }
}

2. 消除循环依赖的关键措施 #

• 禁止外部ref传递：所有任务都使用FutureExecutor自身的ref进行操作，避免引入外部Provider的隐式依赖。
• 谨慎使用订阅：队列任务中如需访问其他Provider，优先用ref.read而非ref.watch，减少不必要的订阅导致的循环引用。
• 明确依赖边界：确保队列任务依赖的Provider不反向依赖futureExecutor，从业务层面切断循环链。

3. 任务失败后的重试逻辑 #

优化后解决了原实现的阻塞问题：

• 捕获异常后记录错误信息
• 等待外部重试信号（比如用户点击重试按钮）
• 重试当前失败任务，成功后再继续执行后续队列任务

使用示例 #

在业务Provider中调用FutureExecutor添加异步任务：

@riverpod
class ExampleProvider extends _$ExampleProvider {
  @override
  ExampleState build() => ExampleState.initial();

  Future<void> fetchData() async {
    // 本地修改状态（加载中）
    state = state.copyWith(isLoading: true);
    // 将异步任务加入队列
    ref.read(futureExecutorProvider.notifier).addTask((executorRef) async {
      try {
        final response = await http.get(Uri.parse("https://your-api.com/data"));
        // 应用远程结果到状态
        state = state.copyWith(data: response.body, isLoading: false);
      } catch (e) {
        state = state.copyWith(isLoading: false);
        rethrow; // 抛出异常交给FutureExecutor处理重试
      }
    });
  }
}

总结 #

• 重构后的FutureExecutor通过自身管理ref，彻底规避了循环依赖风险
• 完善的重试机制确保失败任务重试后再执行后续任务
• 所有任务严格串行执行，完全匹配你的需求

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