RxSwift中有序执行写入任务出错时如何触发Reset回滚Observable逻辑

实现方案 #

你可以用RxSwift自带的catchError操作符实现需求，这是Rx体系内处理错误分支的标准实践，完全不需要嵌套订阅，也能满足你"任意写入任务失败立即触发回滚"的要求。

核心逻辑说明 #

1. concat操作符本身就具备"任意子序列报错，整个序列立即终止并抛出错误"的特性，刚好匹配你按顺序执行、失败立即终止的需求
2. catchError可以拦截上游抛出的错误事件，返回一个全新的Observable替代原序列继续执行，你可以在这个闭包里执行回滚逻辑
3. 整个流程用原生操作符串联，不需要手动创建Observable和管理Disposable，符合Rx最佳实践

优化后代码 #

enum BatchTasksResult{
    case elapsedTime(Double)
    case failedFatal
    case rolledback
    case success
}

func writeBlocks(tasks: [WriteTask]) -> Observable<BatchTasksResult> {
    // 按顺序执行所有写入任务
    return Observable.concat(writeSomewhere(tasks))
        // 把每个写入任务的返回结果转成进度事件
        .map { writeResult in
            BatchTasksResult.elapsedTime(writeResult.totalTime)
        }
        // 所有写入任务成功完成后，追加成功事件
        .concat(Observable.just(.success))
        // 捕获写入流程中的任意错误，触发回滚
        .catch { [weak self] error -> Observable<BatchTasksResult> in
            guard let self = self else {
                return Observable.just(.failedFatal)
            }
            // 执行回滚任务，根据回滚结果返回对应事件
            return self.getResetTaskObservable() // 替换为你返回Observable<ResetTaskResult>的回滚方法
                .map { _ in BatchTasksResult.rolledback }
                // 捕获回滚过程的错误，返回致命失败事件
                .catch { _ in Observable.just(.failedFatal) }
        }
}

常见问题说明 #

你之前用flatMap无法实现的原因是：flatMap只处理序列正常发出的Next元素事件，错误属于序列终止信号，会直接绕过所有普通转换操作符向下传递，所以需要用专门处理错误事件的catchError来拦截分支。

这个方案的事件流完全符合你的预期：

• 每完成一个写入任务，就会发出elapsedTime事件
• 所有写入任务成功完成，会发出success事件，序列正常终止
• 任意写入任务失败，会立即终止写入流程，执行回滚：
  • 回滚成功，发出rolledback事件，序列终止
  • 回滚失败，发出failedFatal事件，序列终止

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