如何在 RXJS 中链接承诺列表？答案

【问题标题】：How to chain a list of promises in RXJS?如何在 RXJS 中链接承诺列表？
【发布时间】：2017-08-30 08:55:30
【问题描述】：

如何在 RXJS 中链接承诺列表？每个 Promise 都需要在前一个 Promise 解决后执行（工作 todo 是有状态的）。

我现在的做法感觉很原始：

const workTodo = []; // an array of work
const allWork = Observable.create(observer => {
  const next= () => {
    const currentTodo = workTodo.shift();
    if (currentTodo ) {
      doTodoAsync(currentTodo)
        .then(result => observer.onNext(result))
        .then(next);
    } else {
      observer.onCompleted();
    }
  };
  next();
});

我在想这样的事情：

const workTodo = []; // an array of work
const allWork = Observable
                  .fromArray(workTodo)
                  .flatMap(doTodoAsync);

但这基本上是一次执行所有的承诺。

【问题讨论】：

标签： javascript asynchronous rxjs rxjs5

【解决方案1】：

看来你的尝试已经很接近了。

您可以将.flatMap 的最大并发指定为 1，例如：

Observable.fromArray(workTodo)
 .flatMap(doTodoAsync, 1)

或等效地使用.concatMap 而不是.flatMap：

Observable.fromArray(workTodo)
 .concatMap(doTodoAsync)

我会使用concatMap，因为它感觉更惯用。

更新： DEMO

【讨论】：

我已经测试了concatMap，但它似乎做了一些不同的事情。它并行执行doTodoAsync，然后按原始顺序组合结果。也在这里描述：reactivex.io/documentation/operators/flatmap.html。 flatMap 上的第二个参数number 似乎不存在。
@nicojs 不确定您的确切意思。使用正在运行的演示更新了答案。它清楚地表明任务是一个接一个地执行，并且mergeMap/flatMap 接受concurrency 参数，如下所述：reactivex.io/rxjs/class/es6/…。

【解决方案2】：

一些递归怎么样？

首先创建一个递归函数并命名为recursiveDoToDo：

const recursiveDoToDo = (currentTodo, index) =>
    Observable
        .fromPromise(doTodoAsync(currentTodo))
        .map(resolved => ({resolved, index}));

上面的代码只是简单地将你的 doTodoAsync 包装到一个 Observable 中，然后我们将结果映射为返回已解析的承诺和数组的 index，以供以后递归使用。

接下来，我们将使用.expand() 运算符递归调用recursiveDoToDo。

recursiveDoToDo(worktodo[0], 0)
    .expand(res => recursiveDoToDo(worktodo[res.index + 1], res.index + 1))
    .take(worktodo.length)

您需要为递归做的只是将索引增加 1。因为.expand() 将永远递归运行，.take() 操作符用于告诉 observable 何时结束流，即长度你的worktodo。

现在您可以简单地订阅它：

recursion.subscribe(x => console.log(x));

这里是working JS Bin

【讨论】：

它似乎工作，但它也似乎很原始。我想我更喜欢我的递归调用。我觉得应该有一个简单的运营商来做到这一点。