当akka actor系统终止时如何运行预定的功能

Question

我有一个演员系统，我用它来安排一个函数的执行，该函数从 flink 的 map 操作符在 Kafka 主题中产生一个事件。如果出现异常，actor 系统将被终止，并在 akka 文档中说明（请参阅https://doc.akka.io/docs/akka/current/scheduler.html#from-akka-actor-actorsystem）所有计划任务都应执行。在我的情况下，当函数执行时，会抛出与函数内部使用的类相关的 java.lang.NoClassDefFoundError。

new RichMapFunction[String, String] {
      implicit lazy val executor: ExecutionContext = ExecutionContext.fromExecutor(Executors.directExecutor())
      var myActorSystem: ActorSystem = _
      var kafkaProducer: KafkaProducer[String, String] = _
      var runtimeContext: RuntimeContext = _

      override def map(value: String): String = {
        value match {
          case "stop" =>
            throw new Exception("Stop command received")
          case _ =>
            myActorSystem.scheduler.scheduleOnce(FiniteDuration(5L, MINUTES)){
              kafkaProducer.send(new ProducerRecord[String, String]("test", value.reverse))
            }
        }

        s"scheduled function on event $value"
      }

      override def open(parameters: Configuration): Unit = {
        myActorSystem = ActorSystem("testSystem")
        kafkaProducer = {
          val props = new Properties()
          props.put("bootstrap.servers", "localhost:9092")
          // props.put("acks", "all")
          props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
          props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
          new KafkaProducer[String, String](props)
        }
        runtimeContext = getRuntimeContext
      }

      override def close(): Unit = {
        println("Terminate actor system...")
        myActorSystem.terminate()
      }
    }

Answer 1

Actor 系统终止是异步的，所以我使用了下面的代码。

Await.result(myActorSystem.terminate(), scala.concurrent.duration.Duration.Inf)