如何在 Akka Stream 中记录流速? [英] How to log flow rate in Akka Stream?

问题描述

我有一个带有单个流/图的 Akka Stream 应用程序.我想测量源头的流速并每 5 秒记录一次，例如在过去 5 秒内收到 3 条消息".我试过了，

I have an Akka Stream application with a single flow/graph. I want to measure the flow rate at the source and log it every 5 seconds, like 'received 3 messages in the last 5 seconds'. I tried with,

someOtherFlow
  .groupedWithin(Integer.MAX_VALUE, 5 seconds)
  .runForeach(seq => 
    log.debug(s"received ${seq.length} messages in the last 5 seconds")
  )

但只在有消息时输出，当有0条消息时没有空列表.我也想要0.这可能吗?

but it only outputs when there are messages, no empty list when there are 0 messages. I want the 0's as well. Is this possible?

推荐答案

你可以尝试类似的事情

  src
    .conflateWithSeed(_ ⇒ 1){ case (acc, _) ⇒ acc + 1 }
    .zip(Source.tick(5.seconds, 5.seconds, NotUsed))
    .map(_._1)

它应该批处理您的元素，直到刻度线释放它们.这受到文档中的示例的启发.

which should batch your elements until the tick releases them. This is inspired from an example in the docs.

另一方面，如果您需要将其用于监控目的，您可以为此目的利用第三方工具 - 例如Kamon.

On a different note, if you need this for monitoring purposes, you could leverage a 3rd party tool for this purpose - e.g. Kamon.

这篇关于如何在 Akka Stream 中记录流速?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！