如何在Akka Stream中记录流量？

Question

我有一个带有单个流/图的Akka Stream应用程序。我想测量源处的流速并每5秒记录一次，就像“最近5秒钟收到3条消息”一样。我尝试过

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

但仅在有消息时输出，而在有0条消息时则没有空白列表。我也想要0。这可能吗？

Answer 1

示例 akka 流日志记录。

  implicit val system: ActorSystem = ActorSystem("StreamLoggingActorSystem")
  implicit val materializer: ActorMaterializer = ActorMaterializer()
  implicit val adapter: LoggingAdapter = Logging(system, "customLogger")
  implicit val ec: ExecutionContextExecutor = system.dispatcher

  def randomInt = Random.nextInt()

  val source = Source.repeat(NotUsed).map(_ ? randomInt)


  val logger = source
    .groupedWithin(Integer.MAX_VALUE, 5.seconds)
    .log(s"in the last 5 seconds number of messages received : ", _.size)
    .withAttributes(
      Attributes.logLevels(
        onElement = Logging.WarningLevel,
        onFinish = Logging.InfoLevel,
        onFailure = Logging.DebugLevel
      )
    )

  val sink = Sink.ignore

  val result: Future[Done] = logger.runWith(sink)

  result.onComplete{
    case Success(_) =>
      println("end of stream")
    case Failure(_) =>
      println("stream ended with failure")
  }

源代码在这里。

Answer 2

您可以尝试类似

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

这应该批处理您的元素，直到刻度线释放它们为止。这是从docs中的示例得到启发的。

另一方面，如果您需要监视目的，则可以使用第三方工具（例如Kamon）来实现此目的。

Answer 3

稍微扩展 Stefano 的回答，我创建了以下流程：

def flowRate[T](metric: T => Int = (_: T) => 1, outputDelay: FiniteDuration = 1 second): Flow[T, Double, NotUsed] =
  Flow[T]
  .conflateWithSeed(metric(_)){ case (acc, x) ? acc + metric(x) }
  .zip(Source.tick(outputDelay, outputDelay, NotUsed))
  .map(_._1.toDouble / outputDelay.toUnit(SECONDS))

def printFlowRate[T](name: String, metric: T => Int = (_: T) => 1,
                     outputDelay: FiniteDuration = 1 second): Flow[T, T, NotUsed] =
  Flow[T]
    .alsoTo(flowRate[T](metric, outputDelay)
              .to(Sink.foreach(r => log.info(s"Rate($name): $r"))))

第一个将流量转换为每秒的速率。您可以提供 a metric，它为通过的每个对象提供一个值。假设您想测量字符串流中字符的比率，那么您可以通过_.length. 第二个参数是流量报告之间的延迟（默认为一秒）。

第二个流可用于内联打印流量以用于调试目的，而无需修改通过流的值。例如

stringFlow
  .via(printFlowRate[String]("Char rate", _.length, 10 seconds))
  .map(_.toLowercase) // still a string
  ...

这将每 10 秒显示一次字符的平均速率（每秒）。