测量消息到达 Akka Streams Flow 或 Sink 的速率

Question

我已经编写了一个 Akka Streams 应用程序并且它工作正常。

我想做的是将我的 JMX 控制台连接到运行 Akka Streams 应用程序的 JVM 实例，然后研究进入我的 Sink 和 Flows 的消息量。

这可能吗？我用谷歌搜索但没有找到具体的方法。

我的应用程序的最后阶段是汇入 Cassandra 数据库。我想知道每秒消息进入 Sink 的速率。

我还想在我的图表中选择一个随机流，然后知道每秒流经该流的消息数。

有什么开箱即用的吗？或者我应该在我的每个流程中编写类似 dropwizard 的代码来衡量速率。

Answer 1

目前没有什么“开箱即用”可以用来监控 Akka Stream 内的费率。

但是，这是一个非常简单的工具，您可以在监控 Flow 中提取它，您可以将其放置在适合您需要的任何位置。

下面的例子是基于Kamon，但是你可以看到它可以很容易地移植到Dropwizard：

  def meter[T](name: String): Flow[T, T, NotUsed] = {
    val msgCounter = Kamon.metrics.counter(name)

    Flow[T].map { x =>
      msgCounter.increment()
      x
    }
  }

  mySource
    .via(meter("source"))
    .via(myFlow)
    .via(meter("sink"))
    .runWith(mySink)

以上是您可以在this repo 找到的演示的一部分。 像这样的临时解决方案具有为您的应用程序完美定制的优势，同时保持简单。