akka timer 提供消息 Cancel timer [Message] with generation [583]答案

【问题标题】：akka timer giving message Cancel timer [Message] with generation [583]akka timer 提供消息 Cancel timer [Message] with generation [583]
【发布时间】：2020-02-15 21:08:17
【问题描述】：

我正在使用akka timer

我正在使用twitter streams，我正在尝试在 5 秒内获取推文数量这是我的代码

case class PerSecond(tweet:Tweet)
case class TweetPerSecondCount(tweet:Tweet)

class TweetPerSecondActor extends Actor with Timers{
var counter=0
 def receive: PartialFunction[Any, Unit] = {
    case PerSecond(tweet) =>
      log.info("Actor TweetPerSecondActor recevied the message PerSecond")
      timers.startPeriodicTimer("perSecond", TweetPerSecondCount(tweet), 5.second)

    case TweetPerSecondCount(tweet) =>
      log.info("Actor TweetPerSecondActor recevied the message TweetPerSecondCount")
      log.info("got the tweet {}",getCounter+"in 5 seconds")

    case message =>
      log.warn("Actor TweetPerSecondActor: Unhandled message received : {}", message)
      unhandled(message)
  }
}

在播放框架的控制器动作中，我从 twitter 流中获取 tweets 对象（连续流而不弯腰）

    class Mycontroller extends Controller {

    val tweetPerSecondActor = system.actorOf......//create actor


    def tweetAveragePerSecond = Action {
        log.debug("in the action tweetAveragePerSecond")

        def getTweet: PartialFunction[StreamingMessage, Unit] = {
          case tweet: Tweet =>
            val future = ask(tweetPerSecondActor, PerSecond(tweet))
        }
        val streaming: Future[TwitterStream] = handleTwitterStreamClient.getStreamingCLient.sampleStatuses(stall_warnings = true)(getTweet)

        Ok("tweet average per second")
      }
    }

当我到达路线时，日志显示

TweetPerSecondActor INFO - Actor TweetPerSecondActor recevied the message PerSecond
16:42:44.335 28939 [ArteciateActorSystem-akka.actor.default-dispatcher-5] TimerScheduler DEBUG - Cancel timer [perSecond] with generation [758]
16:42:44.335 28939 [ArteciateActorSystem-akka.actor.default-dispatcher-5] TimerScheduler DEBUG - Start timer [perSecond] with generation [759]

    6:42:44.335 28939 [TwitterActorSystem-akka.actor.default-dispatcher-6] TweetPerSecondActor INFO - Actor TweetPerSecondActor recevied the message PerSecond
    16:42:44.335 28939 [TwitterActorSystem-akka.actor.default-dispatcher-5] TimerScheduler DEBUG - Cancel timer [perSecond] with generation [758]
    16:42:44.335 28939 [TwitterActorSystem-akka.actor.default-dispatcher-5] TimerScheduler DEBUG - Start timer [perSecond] with generation [759]

如果我传递虚拟字符串值而不是运行流并传递 twitter 对象，则计时器可以正常工作像下面案例类 PerSecond(tweet:String) 案例类 TweetPerSecondCount(tweet:String)

class TweetPerSecondActor extends Actor with Timers{
var counter=0
 def receive: PartialFunction[Any, Unit] = {
    case PerSecond(tweet) =>
      log.info("Actor TweetPerSecondActor recevied the message PerSecond")
      timers.startPeriodicTimer("perSecond", TweetPerSecondCount(tweet), 5.second)

    case TweetPerSecondCount(tweet) =>
      log.info("Actor TweetPerSecondActor recevied the message TweetPerSecondCount")
      log.info("got the tweet {}",getCounter+"in 5 seconds")

    case message =>
      log.warn("Actor TweetPerSecondActor: Unhandled message received : {}", message)
      unhandled(message)
  }
}

    class Mycontroller extends Controller {

    val tweetPerSecondActor = system.actorOf......//create actor


    def tweetAveragePerSecond = Action {
        log.debug("in the action tweetAveragePerSecond")

            val future = ask(tweetPerSecondActor, PerSecond("dummy value"))


        Ok("tweet average per second")
      }
    }

【问题讨论】：

标签： scala timer akka twitter4j twitter4s

【解决方案1】：

Akka 中的计时器是键控的（在您的情况下，键是 "perSecond"）并且 API 保证

每个定时器都有一个键，如果启动了具有相同键的新定时器，则取消前一个定时器

所以每次调用 getTweet 函数（考虑到它是为效果而不是价值而调用的，我更喜欢“过程”，但这可能是特殊的）被调用时，计时器将被取消。

根据您要完成的任务，解决方案可能包括：

每个请求的参与者。您需要让请求返回一些可以传递给未来请求以从参与者那里获取信息的内容。 Actor 生命周期管理也可能是可取的。
为整个控制器保留一个参与者，但每个请求使用唯一的计时器键。如果这些是周期性定时器，您必须跟踪哪些定时器键处于活动状态并取消不再需要的键（保持任意多个周期性定时器运行可能最终会降低性能）

【讨论】：