消息处理节流/背压

Question

我有消息的来源，即Observable。对于每条消息，我想进行一个 HTTP 调用，这将产生另一个 Observable，因此我将它们与 flatMap 组合在一起，然后将它们发送给某个订阅者。这里是这个场景的代码：

Rx.Observable.interval(1000)
.flatMap (tick) ->
  // returns an `Observable`
  loadMessages()
.flatMap (message) ->
  // also returns and `Observable`
  makeHttpRequest(message)
.subscribe (result) ->
  console.info "Processed: ", result

这个例子是用 coffeescript 写的，但我认为问题陈述对任何其他 Rx 实现都是有效的。

我使用这种方法遇到的问题是loadMessages 会非常快速地生成大量消息。这意味着，我在很短的时间内发出了很多 HTTP 请求。这在我的情况下是不可接受的，所以我想将并行 HTTP 请求的数量限制在 10 个左右。换句话说，当我发出 HTTP 请求时，我想限制管道或应用某种背压。

Rx 是否有任何标准方法或最佳实践来处理这种情况？

目前我实现了非常简单（并且非常次优）的背压机制，如果系统处理的按摩过多，则忽略滴答声。它看起来像这样（简化版）：

Rx.Observable.interval(1000)
.filter (tick) ->
  stats.applyBackpressureBasedOnTheMessagesInProcessing()
.do (tick) ->
  stats.messageIn()
.flatMap (tick) ->
  // returns an `Observable`
  loadMessages()
.flatMap (message) ->
  // also returns and `Observable`
  makeHttpRequest(message)
.do (tick) ->
  stats.messageOut()
.subscribe (result) ->
  console.info "Processed: ", result

但我不确定这是否可以做得更好，或者 Rx 可能已经有一些机制来处理这种需求。

Answer 1

这不是严格意义上的背压，这只是限制并发。这是一个简单的方法（忽略我可能错误的语法，通过 TextArea 编码）：

Rx.Observable.interval(1000)
    .flatMap (tick) ->
        // returns an `Observable`
        loadMessages()
    .map (message) ->
        // also returns and `Observable`, but only when
        // someone first subscribes to it
        Rx.Observable.defer ->
            makeHttpRequest(message)
    .merge 10 // at a time
    .subscribe (result) ->
        console.info "Processed: ", result

在 C# 中，等效的想法是 SelectMany，而不是 Select(Defer(x)).Merge(n)。 Merge(int) 最多订阅n in-flight Observables，并将其余的缓存到以后。我们拥有Defer 的原因是为了让我们在Merge(n) 订阅我们之前不做任何工作。

Answer 2

在 RXJS 中你可以使用背压子模块

http://rxjs.codeplex.com/SourceControl/latest#src/core/backpressure/

disclaimer我从未使用过 JS 的 RX 版本，但您确实要求一种实现背压的标准方法，并且核心库似乎支持它。 RX c# 尚无此支持。不知道为什么。

Answer 3

听起来您想从队列中拉取而不是推送您的 http 请求。 Rx 真的是这里正确的技术选择吗？

编辑：

一般来说，我不会使用 Rx 设计一个我对源事件有完全命令式控制的解决方案。这不是一个被动的场景。

Rxjs 中的背压模块显然是为处理您不拥有源流的情况而编写的。给你。

TPL 数据流听起来更适合这里。

如果你必须使用 RX，你可以像这样设置一个循环：如果你想限制 X 个并发事件，设置一个主题作为你的消息源并强制推送（@ 987654323@) X 条消息。在您的订阅者中，您可以在 OnNext 处理程序的每次迭代中将新消息推送到主题，直到源耗尽。这保证了最多 X 条消息在传输中。