我有一个像这样的演员
class TcpClientActor(target: Target) extends Actor with Logger {
override def preStart(): Unit = {
self ! TestConnection
}
override def receive: Receive = {
case TestConnection =>
IO(Tcp) ! Connect(remoteAddress = new InetSocketAddress(target.endpoint, target.port), localAddress = None, options = Nil, timeout = Some(timeout), pullMode = false)
case failed@CommandFailed(_: Connect) =>
info(s"Failure: $target.endpoint:$target.port")
shutdown()
case Connected(_, _) =>
info(s"Success: $target.endpoint:$target.port")
sender() ! Close
shutdown()
}
def shutdown(): Unit = {
context stop self
}
}
我正在迭代一个带有端点的文件来测试并创建其中一个actor,每一行都是Target
类型的构造函数参数。我希望能够限制并行TCP连接的数量以启动到某个设置数量,是否有我可以在Akka中使用的内置机制,以确保我不会因为每一行立即创建一个TcpClientActor
而使系统过载输入并启动套接字连接?
我会使用Akka Stream来限制消息
import scala.concurrent.duration._
import akka.NotUsed
import akka.actor.ActorRef
import akka.stream.{ ActorMaterializer, OverflowStrategy, ThrottleMode }
import akka.stream.scaladsl.{ Sink, Source }
object TcpThrottle {
def throttler(ratePerSecond: Int, burstRate: Option[Int], bufferSize: Int = 1000)(implicit materializer: ActorMaterializer): ActorRef =
Source.actorRef(bufferSize = bufferSize, OverflowStrategy.dropNew)
.throttle(ratePerSecond, 1.second, burstRate.getOrElse(ratePerSecond), ThrottleMode.Shaping)
.to(Sink.actorRef(IO(Tcp), NotUsed)
.run()
}
class TcpClientActor(target: Target) extends Actor with Logger {
val throttler = TcpThrottle.throttler(1, Some(5))
// otherwise identical
def receive: Receive = {
case TestConnection => throttler ! Connect(remoteAddress = new InetSocketAddress(target.endpoint, target.port), localAddress = None, options = Nil, timeout = Some(timeout), pullMode = false)
// other cases identical
}
}
改编自The Akka 2.5 migration guide。可能有必要