RxSwift：如何构建Observable调用链替代回调解决异步顺序操作问题

嘿，完全能理解你被回调地狱折腾的烦躁！RxSwift里的**flatMap操作符**就是专门用来处理这种有依赖关系的异步顺序任务的，能把嵌套的回调彻底改成线性的清晰代码。

核心思路 #

你的三个操作是严格的顺序依赖：OriginalData → operationA → OutputA → operationB → OutputB → operationC，RxSwift可以通过链式调用flatMap来串联这些异步操作——每个flatMap都会等待前一个Observable完成并发出结果，再用这个结果启动下一个异步任务。

代码示例 #

首先假设你的三个操作都被封装成返回Observable的函数（这是RxSwift处理异步的标准方式）：

// 先定义你的数据类型和操作函数
struct OriginalData { /* 你的原始数据结构 */ }
struct OutputA { /* operationA的输出结构 */ }
struct OutputB { /* operationB的输出结构 */ }

func operationA(data: OriginalData) -> Observable<OutputA> {
    // 这里是operationA的异步逻辑，比如网络请求、耗时计算等
    // 示例：模拟异步操作，1秒后返回OutputA
    return Observable.create { observer in
        DispatchQueue.global().asyncAfter(deadline: .now() + 1) {
            observer.onNext(OutputA())
            observer.onCompleted()
        }
        return Disposables.create()
    }
}

func operationB(data: OutputA) -> Observable<OutputB> {
    // operationB的异步逻辑，接收OutputA作为输入
    return Observable.create { observer in
        DispatchQueue.global().asyncAfter(deadline: .now() + 1) {
            observer.onNext(OutputB())
            observer.onCompleted()
        }
        return Disposables.create()
    }
}

func operationC(data: OutputB) -> Observable<Void> {
    // operationC的异步逻辑，接收OutputB作为输入，无返回值
    return Observable.create { observer in
        DispatchQueue.global().asyncAfter(deadline: .now() + 1) {
            observer.onNext(())
            observer.onCompleted()
        }
        return Disposables.create()
    }
}

接下来在你的类里，用RxSwift串联这些操作：

class YourViewModel {
    let originalData: OriginalData = OriginalData()
    private let disposeBag = DisposeBag()

    func startSequentialOperations() {
        // 1. 把原始数据包装成Observable作为序列起点
        Observable.just(originalData)
            // 2. 执行operationA，用其输出作为下一个操作的输入
            .flatMap { [weak self] data in
                guard let self = self else { return Observable.empty() }
                return self.operationA(data: data)
            }
            // 3. 用operationA的输出执行operationB
            .flatMap { outputA in
                self.operationB(data: outputA)
            }
            // 4. 用operationB的输出执行operationC
            .flatMap { outputB in
                self.operationC(data: outputB)
            }
            // 5. 订阅序列，处理最终结果或错误
            .subscribe(
                onNext: { _ in
                    print("所有操作都执行完成啦！")
                },
                onError: { error in
                    print("某个操作出错了：\(error.localizedDescription)")
                },
                onCompleted: {
                    print("序列正常结束")
                }
            )
            .disposed(by: disposeBag) // 记得用disposeBag管理订阅生命周期
    }
}

关键细节解释 #

• flatMap的作用：它会订阅前一个Observable，当前一个Observable发出元素时，用这个元素创建一个新的Observable（也就是你的下一个操作），并将这个新Observable的元素转发出去。这样就保证了操作的顺序性——只有前一个操作完成并发出结果，下一个操作才会启动。
• 错误处理：如果序列中任何一个操作发出error事件，整个序列会立即终止，错误会传递到subscribe的onError闭包里。你也可以用catchError操作符来捕获错误并继续序列，比如：

  .flatMap { outputA in
      self.operationB(data: outputA)
  }
  .catchError { error in
      print("operationB出错了：\(error)")
      return Observable.just(OutputB()) // 返回一个默认值继续执行operationC
  }
  

• 生命周期管理：一定要用DisposeBag来管理订阅，避免内存泄漏——当持有disposeBag的对象（比如ViewModel）被销毁时，所有订阅都会自动取消。

如果你的操作原本不是返回Observable的，比如是闭包回调的形式，也可以先把它们包装成Observable，比如：

// 把基于回调的operationA包装成Observable
func wrappedOperationA(data: OriginalData) -> Observable<OutputA> {
    return Observable.create { observer in
        // 调用原来的回调式operationA
        oldOperationA(data: data) { outputA, error in
            if let error = error {
                observer.onError(error)
            } else if let outputA = outputA {
                observer.onNext(outputA)
                observer.onCompleted()
            }
        }
        return Disposables.create()
    }
}

这样就能无缝把旧的回调代码接入RxSwift的序列中啦！

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