使用 RxJS 进行业务事件的复杂事件处理答案

【问题标题】：Using RxJS for Complex Event Processing of business events使用 RxJS 进行业务事件的复杂事件处理
【发布时间】：2013-11-03 12:00:39
【问题描述】：

我有一个我们的 Web 应用程序产生的业务事件数据库。
我想编写 JavaScript 应用程序来处理这些事件（最多数百个），并使用复杂事件处理会在其中找到一些模式。

例如，如果我们在 15 分钟内有登录失败事件而没有登录事件，我们想知道它。我们使用会话 ID 关联事件。

我更喜欢使用现有库，从我看到的 RxJS 看来是正确的工具。
我的问题是如何使用原始事件的时间戳而不是当前计算机时间来构建事件流？

我看到 RxJS 有时间运算符，但看起来它使用的是当前时间。

更新 1
我发现 HistoricalScheduler 类看起来不错，但没有关于它的文档，我不确定它是否存在于 RxJS 中。

【问题讨论】：

因为这些事件已经发生，并且您很可能打算将此查询作为一次性交易运行，我怀疑 Rx（或任何可观察的）是您想要的方式.自从使用 Ix（交互式扩展）以来，它会做得更多，因为您要查询的事件已经存在。
另外，你是在什么环境下运行的？可能有更好的方法来进行这种查询。例如，您最好使用简单的数据库查询。是否有理由需要将所有结果拉入内存，就像使用 RxJS（没有查询提供程序）那样？
最后，由于前面提到的原因，我相信调度程序与这个逻辑几乎没有关系。调度程序控制事件/通知将处理的时间和地点，并不打算指示数据处理的时间。事实上，每个事件的时间戳应该已经存储在你的数据库中了。
我理解并同意。我认为使用 Rx 来查询存在的事件应该很容易，显然我错了。在观看和阅读 Rx 时，我读到他们使用虚拟调度程序进行测试，这似乎是重新运行现有事件的好方法，就好像它们真的发生了一样。最后一件事是，如果这个系统存在，把它变成实时处理器应该就像改变它的输入一样简单，不是吗？
啊，我明白你的意思了。实际上，如果您想这样做，我建议您将事件的实际时间戳（而不是系统时间或调度程序时间）用于您的逻辑。因为那个时间戳已经存在（如果它不存在，你就不能使用 Rx 来获取过去事件的准确时间戳），你可以直接使用它......你不需要调度程序或类似的东西。只需在产生事件之前按时间对事件进行排序。

标签： rxjs complex-event-processing

【解决方案1】：

检查 rxmarbles 示例代码。

特别是：https://github.com/staltz/rxmarbles/blob/master/src/controllers/utils.coffee，它使用 Rx.VirtualTimeScheduler。

据我了解，您应该能够适应您的用例

【讨论】：

Thomas，链接好像坏了

【解决方案2】：

据推测，您应该能够按时间戳对现有事件进行排序，以了解它们发生的时间，并将其与未来的事件合并（因为它们显然已经被排序了）。之后，您可以以相同的方式处理事件。

var pastEvents = [
    { userId: 1, time: 100, status: 'failure' },
    { userId: 2, time: 400, status: 'success' },
    { userId: 1, time: 300, status: 'success' },
    { userId: 4, time: 200, status: 'success' },
  ]
  .sort(function (a, b) { return b.time - a.time; });

var futureEvents = getFutureEventObservable();

var events = Rx.Observable.fromArray(pastEvents).merge(futureEvents);

var groups = events.groupBy(function (x) { return x.userId; });

// do "complex event processing" here.

不幸的是，如果事件还没有时间戳，那么数据就会丢失，并且 Rx 没有特殊的方式来重现它。

【讨论】：