RXJS Observables中处理Promise React Native场景下如何替代async/await写法

核心解法 #

RxJS原生支持对Promise的适配，不需要借助async/await就能实现等待异步操作完成的逻辑，改造后的纯Observable写法如下：

const myDownloadEpic = (
  action$,
  state$
) =>
  action$.pipe(
    ofType(myDownloadActionType), // 捕获对应action
    map(action => action.payload),
    mergeMap(payload => downloadManager.download(payload)), // 下载文件得到Blob
    mergeMap(response => 
      // 无依赖的两个异步操作并行执行，forkJoin会等全部完成后返回结果数组
      forkJoin([
        from(RNFS.readDir(RELEVANT_TARGET_PATH)),
        from(convertToBase64(response))
      ])
    ),
    mergeMap(([files, base64Data]) => 
      // 写入文件，等待操作完成
      from(RNFS.writeFile(FULL_PATH, base64Data, 'base64'))
    ),
    // 后续业务逻辑，比如返回成功action
    map(() => downloadSuccessAction()),
    // 统一捕获全链路异常
    catchError(error => of(downloadFailAction(error)))
  )

关键说明 #

• 所有返回Promise的异步方法，都可以用from()直接转换为Observable，RxJS会自动等待Promise状态变更后再向下传递结果，等效于await的效果
• 对于无依赖的并行异步操作，用forkJoin包裹，仅当所有输入的Observable都完成时才会向下发射结果数组，完全满足你等待所有异步执行完毕再做后续处理的需求
• 对于有先后依赖的串行异步操作，直接链式调用mergeMap/concatMap即可，整个数据流是扁平的链式结构，没有嵌套，不会出现冗余代码
• 全链路的异常可以统一用最外层的catchError捕获处理，比async/await的多层try/catch写法更简洁易维护

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