akka 使用Scala节流器改变每个消息之间的时间

gfttwv5a  于 2022-11-23  发布在  Scala
关注(0)|答案(1)|浏览(196)

我正在学习Akka,并编写了下面的代码来限制发送到TradeAction演员的消息数量。每秒最多发送3条消息。可以修改限制器以设置每条消息之间的时间吗?例如,在message1和message2之间延迟2秒,在message2和message3之间延迟1秒。

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

import scala.concurrent.duration.DurationInt

case class Action(side: String)

object PlaceTrade {

  val actorSystem = ActorSystem("firstActorSystem")
  println(actorSystem.name)

  object TradeAction {
    def props(action : String) = Props(new TradeAction(action))
  }

  class TradeAction(actorName: String) extends Actor {
    override def receive: Receive = {
      case "Buy" => {
        val r = requests.get("http://www.google.com")
        println("r status code is "+r.statusCode)
        println("Buy")
        println("")
      }
      case "Sell" => {
        val r = requests.get("http://www.google.com")
        println("r status code is "+r.statusCode)
        println("Sell")
        println("")
      }
      case _ =>
    }
  }

  implicit val materializer = ActorMaterializer.create(ActorSystemConfig.getActorSystem)

  def getThrottler(ac: ActorRef) = Source.actorRef(bufferSize = 1000, OverflowStrategy.dropNew)
    .throttle(1, 3.second)
    .to(Sink.actorRef(ac, NotUsed))
    .run()

  def main(args: Array[String]): Unit = {
    val tradeAction = actorSystem.actorOf(TradeAction.props("TradeAction"))
    val throttler = getThrottler(tradeAction)

    val l = List(Action("Buy"),Action("Buy"),Action("Buy"),Action("Sell"))
    l.foreach(action => {
      throttler ! action.side
    })
  }

}
axzmvihb

axzmvihb1#

您可以使用delayWith来实现这一点,它允许一个(可能是有状态的)方法定义延迟此元素的时间长度(无需重新排序元素),例如:

import akka.stream.scaladsl.{ DelayStrategy, DelayOverflowStrategy }
import scala.concurrent.duration.FiniteDuration

def decliningDelay(): DelayStrategy[Any] =
  new DelayStrategy {
    var nextDelaySeconds: Option[Int] = None

    def nextDelay(elem: Any): FiniteDuration =
      nextDelaySeconds match {
        case None =>
          nextDelaySeconds = 2
          0.seconds
        case Some(delay) if delay > 0 =>
          nextDelaySeconds = delay - 1
          delay.seconds
        case _ => 0.seconds
      }
  }

def getThrottler(ac: ActorRef): ActorRef =
  Source.actorRef(bufferSize = 1000, OverflowStrategy.dropNew)
    .delayWith(() => decliningDelay(), DelayOverflowStrategy.backpressure)
    .to(Sink.actorRef(ac, NotUsed))
    .run()

您可以将delayWiththrottle组合在一起;我可能会把delayWith放在throttle之前。

相关问题