如何计算 Akka Streams 中 GraphStage 内的聚合？

Question

我在 Akka 流中有一个 operator/component，旨在在 5 秒的 window 内计算一个值。因此，我使用 TimerGraphStageLogic 创建了我的 operator/component，您可以在下面的代码中看到它。为了测试它，我创建了 2 个源，一个递增，另一个递减，然后我使用 Merge 形状合并它们，然后我使用我的 windowFlowShape，最后在 [=] 中发射它们15=]形状。我确保 TimerGraphStageLogic 正常工作，因为我在另一个 PoC 中对其进行了测试。在此示例中，我只是将通用类型 T 替换为 Int，因为我必须指定我的 window 将聚合的内容。

但是，我的问题是我无法在 window 阶段运算符中聚合 Int 值。当我尝试执行 sum = sum + elem 时，我在运行时收到错误消息：

overloaded method value + with alternatives:
  (x: scala.Int)scala.Int <and>
  (x: Char)scala.Int <and>
  (x: Short)scala.Int <and>
  (x: Byte)scala.Int
 cannot be applied to (Int(in class WindowProcessingTimerFlow))
                sum = sum + elem

这是我编译但在运行时抛出上述错误的代码：

import akka.actor.ActorSystem
import akka.stream._
import akka.stream.scaladsl.{Flow, GraphDSL, Merge, RunnableGraph, Sink, Source}
import akka.stream.stage._
import scala.collection.mutable
import scala.concurrent.duration._

object StreamOpenGraphWindow {
  def main(args: Array[String]): Unit = {
    run()
  }
  def run() = {
    implicit val system = ActorSystem("StreamOpenGraphWindow")
    val sourceNegative = Source(Stream.from(0, -1)).throttle(1, 1 second)
    val sourcePositive = Source(Stream.from(0)).throttle(1, 1 second)

    // Step 1 - setting up the fundamental for a stream graph
    val windowRunnableGraph = RunnableGraph.fromGraph(
      GraphDSL.create() { implicit builder =>
        import GraphDSL.Implicits._
        // Step 2 - create shapes
        val mergeShape = builder.add(Merge[Int](2))
        val windowFlow = Flow.fromGraph(new WindowProcessingTimerFlow[Int](5 seconds))
        val windowFlowShape = builder.add(windowFlow)
        val sinkShape = builder.add(Sink.foreach[Int](x => println(s"sink: $x")))

        // Step 3 - tying up the components
        sourceNegative ~> mergeShape.in(0)
        sourcePositive ~> mergeShape.in(1)
        mergeShape.out ~> windowFlowShape ~> sinkShape

        // Step 4 - return the shape
        ClosedShape
      }
    )
    // run the graph and materialize it
    val graph = windowRunnableGraph.run()
  }

  // step 0: define the shape
  class WindowProcessingTimerFlow[Int](silencePeriod: FiniteDuration) extends GraphStage[FlowShape[Int, Int]] {
    // step 1: define the ports and the component-specific members
    val in = Inlet[Int]("WindowProcessingTimerFlow.in")
    val out = Outlet[Int]("WindowProcessingTimerFlow.out")

    // step 3: create the logic
    override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new TimerGraphStageLogic(shape) {
      // mutable state
      val batch = new mutable.Queue[Int]
      var open = false

      // step 4: define mutable state implement my logic here
      setHandler(in, new InHandler {
        override def onPush(): Unit = {
          try {
            val nextElement = grab(in)
            batch.enqueue(nextElement)
            if (open) {
              pull(in) // send demand upstream signal, asking for another element
            } else {
              var sum: scala.Int = 0
              val set: Iterable[Int] = batch.dequeueAll(_ => true).to[collection.immutable.Iterable]
              set.toList.foreach { elem =>
                sum = sum + elem // ************* WHY I CANNOT DO IT? *************
              }
              push(out, sum)
              open = true
              scheduleOnce(None, silencePeriod)
            }
          } catch {
            case e: Throwable => failStage(e)
          }
        }
      })
      setHandler(out, new OutHandler {
        override def onPull(): Unit = {
          pull(in)
        }
      })
      override protected def onTimer(timerKey: Any): Unit = {
        open = false
      }
    }
    // step 2: construct a new shape
    override def shape: FlowShape[Int, Int] = FlowShape[Int, Int](in, out)
  }
}

Answer 1

因为您正在创建一个名为 Int 的类型参数，它隐藏了类型 Int 定义，其中定义：

class WindowProcessingTimerFlow[Int](silencePeriod: FiniteDuration) extends GraphStage[FlowShape[Int, Int]] {

尝试从中删除泛型：

class WindowProcessingTimerFlow(silencePeriod: FiniteDuration) extends GraphStage[FlowShape[Int, Int]] {

如何计算 Akka Streams 中 GraphStage 内的聚合？

How do I compute an aggregation inside a GraphStage in Akka Streams?

scala

akka

akka-stream