交易引擎中的 RxJava答案

【问题标题】：RxJava in trading engine交易引擎中的 RxJava
【发布时间】：2015-02-25 18:01:03
【问题描述】：

我将编写伪代码以避免不必要的样板。我是 Rx 的新手，但我真的很想使用它而不是 Features 和类似的东西......

这是交易引擎的简化模型

我们有一个提供人们可能感兴趣的所有事件流（可观察对象），例如

market.updateStream // stream of orderbook update events

我们有所谓的 s，它们将订阅所需的流并在发生任何变化时“做出反应”

market.updateStream.subscribe(bot1.marketUpdateAction)
market.updateStream.subscribe(bot2.marketUpdateAction)

我们可能有许多机器人将订阅一个市场，每个机器人要么开始计算要么忽略更改事件。

现在我们有 N 个机器人对 1 个事件做出反应，我们需要有人比较他们的计算并决定哪一个是最有利可图的，如果某个机器人正在减速并且超出我们准备等待的时间限制，我们跳过它并继续进行比较和执行......为此我们有一个 botController 订阅了所有机器人事件，以便它知道机器人何时决定对事件做出反应......就像这样

bot1.calculationStream.subscribe(botController.botActivityAction)

bot 将依次发出 2 个不同的事件（calculationStarted 和 calculationEnded，其中包含实际结果）

当机器人发出开始计算的事件时（如果市场事件对机器人感兴趣，因此并非所有机器人都会发出启动事件，就会发生这种情况）botController 应执行以下操作，从第一个机器人启动事件开始计算时间，并且等待所有发出类似事件的机器人，如果注册的机器人提前完成，比较立即开始......

对不起，如果这个问题太抽象了，但我真的不知道如何使用 RxJava 实现 botControllers 行为......任何努力都值得赞赏......有这么多 rx 转换，我真的不知道我可以在那里使用什么。

更新

假设我们的控制器订阅了 N 个机器人事件，每个机器人可以发出 2 个事件（STARTED、COMPLETED）...

现在，当控制器收到第一个 STARTED 事件时，它开始倒计时 T，而 T 时间未到期，它将接受来自机器人的新事件，当 T 到期或所有机器人返回 COMPLETED 事件时，它会进行一些计算并返回单个结果。 ..

我不明白的部分：据我所知，rx 是孤立地处理每个事件，因此可以避免典型的并发问题，现在我有几个相互关联的事件，我不知道如何使用rx...我只是需要一些指导。

【问题讨论】：

您的问题是什么？我认为你想知道如何构建一个“机器人控制器”来订阅 N 个 observables，然后在 T 时间后获取所有可用的结果，对它们做一些事情，然后传递？跨度>
几乎是的，当第一个“botStartedCalculation”发出 botController 注册该机器人并知道该机器人将返回“botResult”（或者如果经过 T 时间并且该机器人没有任何东西到达时被忽略），我的问题/问题是我真的不明白如何使用 Rx 转换来实现这种行为，我需要类似伪代码之类的东西，如果这可以通过 Rx 给我一个线索......我将更新问题以更清楚地解释问题...
我什至试图找人帮忙，但没有人回应:(

标签： java-8 reactive-programming rx-java

【解决方案1】：

我不完全理解您的问题，但这里有一些设计理念可以向您展示如何思考“Rx 方式”：

我不会订阅机器人，但它们应该是更新流上的 map 或 flatMap，以便它们将更新流转换为答案流。

我会创建一个有 4 个子类的 BotAnswer 类：Result、NotInterested、Timeout、BotError。

然后对于每个机器人：

Observable<BotAnswer> bot1Answers = market.updateStream.flatMap(event -> 
  Observable.just(event).map({ if (event is interesting)
    return new Result(doBotCalculations())
  else
    return new NotInterested()
  })
  .timeout(T, TimeUnit.SECONDS)
  .onErrorResumeNext(
     error -> if (error instanceof TimeoutException)
        return Observable.just(new Timeout())
      else
        return Observable.just(new BotError(error))
   )
)

控制器将对所有机器人答案执行zip：

Observable.zip(bot1Ansers, bot2Answers, ... (a1, a2, ...) -> {
  // compare the answers a1, a2, ... do some calculations, return result
})

【讨论】：

非常感谢，我从这个开始。
但是我们这里有一个大问题，每个机器人都会开始自己的超时计算......但是我们需要控制器从第一个机器人开始......你认为它可以使用 RX 吗？跨度>

【解决方案2】：

这是我想出的...

每个机器人都订阅市场更新，当市场更新时，机器人开始计算，当它完成发射完成事件时，如果它从主管那里得到终止事件，它停止计算，如果它得到另一个市场更新它的终止当前执行并重新开始......

主管也订阅市场更新，当它得到一个它期望所有机器人计算结果时（没有结果也是结果:)，如果某个机器人通过发送终止事件（并将该机器人视为未返回任何结果）来忽略他的时间过长，一旦它得到所有的机器人结果，它就会意识到它开始比较它们并将最终值发送给它的订阅者，如果它在执行时获得另一个市场更新，它将终止其计算并重新开始（等待机器人）...

关于实现，Bot 将设法处理对其方法的适当并发访问，例如（启动、终止）以及 Supervisor，最终这些对象确保不会通过并发执行对对象进行不安全操作，使用传统同步，但是触发什么以及周围发生什么的主要流程是由 Rx 控制的......

请注意，在这种情况下超时由主管控制...

我在这里想念什么吗？如果我必须将传统同步与 rx 结合使用，也许我做错了什么？

【讨论】：