如何在 Akka Streams 中的 GraphStage 内计算聚合? [英] How do I compute an aggregation inside a GraphStage in Akka Streams?
问题描述
我在 Akka 流中有一个运算符/组件,旨在在 5 秒的窗口内计算一个值.因此,我使用 TimerGraphStageLogic
创建了我的操作符/组件,您可以在下面的代码中看到它.为了测试它,我创建了 2 个源,一个递增,另一个递减,然后使用 Merge
形状合并它们,然后使用 windowFlowShape
,最后以 Sink
形状发射它们.我确保 TimerGraphStageLogic 正在工作,因为我在另一个 PoC 中对其进行了测试.在这个例子中,我只是将泛型类型 T
替换为 Int
,因为我必须指定我的窗口将聚合什么.
I have an operator/component in Akka stream that aims to compute a value within a window of 5 seconds. So, I created my operator/component using TimerGraphStageLogic
which you can see on the code below. In order to test it I created 2 sources, one that increments and the other that decrements, then I merge them using the Merge
shape, then I use my windowFlowShape
, and finally emit them in a Sink
shape. I ensure that the TimerGraphStageLogic is working because I tested it in another PoC. In this example I am justing replacing the generic type T
to Int
since I have to specify what my window will aggregate.
但是,我的问题是我无法在 window stage 运算符中聚合 Int
值.当我尝试执行 sum = sum + elem
时,我在运行时收到一个错误:
However, my problem is that I cannot aggregate the Int
values inside the window stage operator. I receive an error in runtime when I try to execute sum = sum + elem
that says:
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
这是我编译但在运行时抛出上述错误的代码:
Here is my code which compiles but throws the above error in runtime:
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)
}
}
推荐答案
因为您正在创建一个名为 Int
的类型参数,它隐藏了类型 Int
定义,其中定义:
Because you are creating a type parameter named Int
which shadows the type Int
definition where defining:
class WindowProcessingTimerFlow[Int](silencePeriod: FiniteDuration) extends GraphStage[FlowShape[Int, Int]] {
尝试从中删除泛型:
class WindowProcessingTimerFlow(silencePeriod: FiniteDuration) extends GraphStage[FlowShape[Int, Int]] {
这篇关于如何在 Akka Streams 中的 GraphStage 内计算聚合?的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!