如何在 rxjs 中链接 observables

Question

我有一个 observable 从服务器拉取事件，过滤应用程序类型的事件，然后订阅并将事件分派给一个或多个处理程序来处理。

然后，处理程序开始对数据库进行一些异步更新，我发现 observable 会快速生成事件，以至于更新会相互影响。我应该预料到的。

所以我认为我需要我的处理程序每​​个都使用自己的 observable 充当队列，该队列将处理一个事件并等待确认。

所以我的问题是，如何创建一个连续接收消息并一次发送一条消息以等待确认，然后再发布下一条消息的 observable？

我认为 observables 也需要很冷，因为我不能丢失消息。

Answer 1

我认为运营商concatMap 所做的事情与您正在寻找的内容很接近。您可以在 SO 上查看以前的答案，以说明 concatMap 的类似用例： RxJS queueing dependent tasks

它很接近但不完全是您想要的，因为没有等待ACK 信号来释放下一个值。相反，concatMap 使用当前“执行”的 observable 的完成信号来订阅下一个。如果您的 observable 在某处包含在数据库上执行更新，那么这些更新将按顺序执行。例如：

function handler (source$) {
  // source$ is your source of events from which you generate the update calls
  return source$.concatMap(function (event){
    return updateDB(event);
  })
}

function updateDB(event) {
  return Rx.Observable.create(function(observer){
    // do the update in the db
    // you probably have a success and error handler 
    // you plug the observer notification into those handlers
    if (success) {
      // if you need to pass down some value from the update
      observer.onNext(someValue);
      // In any case, signal completion to allow concatMap to move to next update
      observer.onCompleted();
    }
    if (error) {observer.onError(error);}
  })
}

这是专门针对您正在使用的库的通用代码。您可能可以直接使用运算符fromNodeCallback 或fromCallback，具体取决于您的数据库更新函数的API。

同样，请注意，在执行当前 observable 时，可能会涉及一些缓冲来保持下一个 observable，并且该缓冲区只能是有限的，所以如果您在生产者之间的速度确实存在显着差异和消费者或内存限制，您可能希望以不同的方式处理事情。

另外，如果你使用的是 RxJS v5，onError 变为 error，onComplete 变为 complete，onNext 变为 next（参见new observer interface）。

最后评论，您的流的有损/无损性质是一个不同于流的冷热性质的概念。您可以查看illustrated subscription and data flows 了解这两种类型的流。