Akka-streams 基于时间的分组

Question

我有一个监听事件流的应用程序。这些事件往往成块出现：10 到 20 个在同一秒内发生，它们之间有几分钟甚至几小时的沉默。这些事件被处理并产生一个聚合状态，并且这个更新的状态被进一步发送到下游。

在伪代码中，它看起来像这样：

kafkaSource()
  .mapAsync(1)((entityId, event) => entityProcessor(entityId).process(event)) // yields entityState
  .mapAsync(1)(entityState => submitStateToExternalService(entityState))
  .runWith(kafkaCommitterSink)

问题是下游 submitStateToExternalService 每秒更新 10-20 个状态是没有用的 - 只发出最后一个状态并只处理那个状态会更有效。

考虑到这一点，我开始研究是否不能在处理后立即不发出状态，而是等待一段时间看看是否有更多事件进入。

在某种程度上，它类似于conflate，但它会在下游停止背压时立即发出元素，而且我的处理速度实际上足以跟上进来的事件，所以我不能依赖背压.

我遇到了groupedWithin，但只要窗口结束（或达到最大元素数），它就会发出元素。理想情况下，我想要的是一个时间窗口，其中在向下游发射之前的等待时间由组中的每个新元素重置。

在我自己实现一些东西来做这件事之前，我想确保我没有忽略 akka-streams 中已经存在的做这件事的方法，因为这似乎是一个相当普遍的事情。

Answer 1

老实说，我会把entityProcessor 变成一个集群分片的持久参与者。

case class ProcessEvent(entityId: String, evt: EntityEvent)

val entityRegion = ClusterSharding(system).shardRegion("entity")

kafkaSource()
  .mapAsync(parallelism) { (entityId, event) =>
    entityRegion ? ProcessEvent(entityId, event)
  }
  .runWith(kafkaCommitterSink)

这样，您可以安全地增加并行度，以便您可以同时处理多个实体的事件，而不必担心对任何特定实体的事件进行错误排序。

然后，您的实体参与者将更新其状态以响应流程命令，并使用合适的持久性插件持久化事件，发送回复以完成询问模式。获得您正在寻找的压缩效果的一种方法是让他们在一段时间后安排外部服务的更新（在取消任何以前安排的更新之后）。

这种方案有一个潜在的缺陷（自制的 Akka Stream 解决方案也存在一个潜在问题，即允许在更新状态之前处理 n > 1 个事件）：如果服务会发生什么情况在更新本地状态视图和更新外部服务之间失败？

您可以处理此问题的一种方法是在实体的状态中编码实体是否脏（具有尚未传播到外部服务的状态），并在启动时构建实体列表并遍历它们以具有脏实体更新外部状态。

如果实体所做的不仅仅是跟踪状态以发布到单个外部数据存储，那么使用 Akka Persistence Query 构建一个完整的读取端视图来更新外部服务可能会很有用。但是，在这种情况下，由于读取端视图的 (State, Event) => State 转换与实体处理器的转换相同，因此采用这种方式可能没有意义。

中途替代方案是将调度等卸载到不同的参与者或参与者组，这些参与者被告知“该实体更新了它的状态”，然后安排一个询问实体的当前状态，时间戳为状态已在本地更新。收到响应后，如果时间戳比上次更新更新，则更新外部服务。