在 Akka 上调用 Actor 之前累积消息答案

【问题标题】：Accumulate messages before call Actor on Akka在 Akka 上调用 Actor 之前累积消息
【发布时间】：2017-05-17 01:32:36
【问题描述】：

我正在使用 Akka Stream 实现一个节流阀，它几乎可以工作了。它几乎可以工作，因为我想每次发送一个消息列表而不是一个。我不确定 Akka 的 Actors 是否允许在调用 Actor 之前累积消息。

我知道下面的例子很简单，但在实际场景中，我会收到来自 RabbitMQ 的大量消息，我想在几秒钟后调用 actor 以避免过高的速率 - 收到消息后，我会调用网络服务。

这是我所做的：

import akka.NotUsed
import akka.actor.{Actor, ActorRef, ActorSystem, Props}
import akka.stream.{ActorMaterializer, OverflowStrategy, ThrottleMode}
import akka.stream.scaladsl.{Sink, Source}

import scala.concurrent.duration._

object AkkaStream {
  def main(args:Array[String]): Unit = {
    val system = ActorSystem("HelloSystem")
    val actor = system.actorOf(Props[ToughActor])
    implicit val materializer = ActorMaterializer.create(system)
    val throttler: ActorRef =
      Source.actorRef(bufferSize = 1000, OverflowStrategy.dropNew)
        .throttle(2, 5.second, 2, ThrottleMode.Shaping)
        .to(Sink.actorRef(actor, NotUsed))
        .run()

    throttler ! CoolThing("message 1")
    throttler ! CoolThing("message 2")
    throttler ! CoolThing("message 3")
    throttler ! CoolThing("message 4")
    throttler ! CoolThing("message 5")
  }
}

class ToughActor extends Actor {
  def receive = {
    case x => println(x.getClass)
  }
}

case class CoolThing(name: String)

在我的示例中，我想在 receive 函数上接收最多 2 个对象的集合，而不是 receive 被调用两次。

如果我发送了 2 条消息（例如：CoolThing("message 1") 和 CoolThing("message 2")），是否可以在 Actor 上接收 List(CoolThing("message 1"), CoolThing("message 2")？

【问题讨论】：

标签： akka akka-stream

【解决方案1】：

我认为您想要一个分组功能而不是节流阀。或者一个接一个。

例如，

 Source.actorRef(bufferSize = 1000, OverflowStrategy.dropNew)
    .groupedWithin(2, 2.seconds)
    .to(Sink.actorRef(actor, NotUsed))
    .run()

【讨论】：