Scala - 回调的返回值

Question

我对 scala 编程相当陌生。有人可以帮我处理回调的返回值吗？如何从调用方法返回回调值作为 JsObject？我正在使用带有演员系统的 Play2 框架。如果我的返回类型错误，请告诉我，与 SendToKafka 方法中的 JsObject 相比，我应该返回 Future。

我有以下代码

override def SendToKafka(data: JsValue): Option[JsObject] = {
  val props: Map[String, AnyRef] = Map(
    "bootstrap.servers" -> "localhost:9092",
    "group.id" -> "CountryCounter",
    "key.serializer" -> "io.confluent.kafka.serializers.KafkaAvroSerializer",
    "value.serializer" -> "io.confluent.kafka.serializers.KafkaAvroSerializer",
    "schema.registry.url" -> "http://localhost:8081"
  )

  val schema: Schema = new Parser().parse(Source.fromURL(getClass.getResource("/test.avsc")).mkString)

  val gRecord: GenericRecord = new GenericData.Record(schema)
  gRecord.put("emp_id", request.emp_id)

  val producer = new KafkaProducer[Int, GenericRecord](props.asJava)
  val record = new ProducerRecord("Emp", 1, gRecord)

  val promise = Promise[RecordMetadata]()

  producer.send(record, producerCallback(promise))
  val f = promise.future
  val returnValue : Some[JsObject] =null
  val con = Future {
    f onComplete {
      case Success(r) => accessLogger.info("r" + r.offset())
      case Failure(e) => accessLogger.info("e "+ e)
    }

    // I would like to return offset as JsObject or exception ( if any )
  }

  private def producerCallback(promise: Promise[RecordMetadata]): Callback = {
    new Callback {
      override def onCompletion(metadata: RecordMetadata, exception: Exception): Unit = {

      val result = if (exception == null) {
        //accessLogger.info("offset - " + metadata.offset())
        // I would like to return this offset as JsObject 
        Success(metadata)
      }
      else {
        accessLogger.error(exception.printStackTrace().toString)
        Failure(exception)
        // I would like to return exception (if any ) as JsObject 
      }
      promise.complete(result)
    }
  }
}

Answer 1

因为promise 是Promise[RecordMetadata] 类型，而f 是promise.future，f 是Future[RecordMetadata] 类型。未来将拥有result 的任何内容，在promise.complete(result) 中。

未来可能最终会包含失败（即回调中的Failure(exception)），因此需要处理（下面使用匹配/案例来说明）

Await.ready 可用于等到未来有Success 或Failure ——但如果没有这样的阻塞调用，在同一个方法中，未来可能还不会完成。

import scala.concurrent.duration._
import scala.concurrent._
...
// arbitrary time -- set an appropriate wait time
val fReady: Future[RecordMetadata] = Await.ready(f, 4.seconds)

// After Await.ready is called, *up to* the duration (4s here) has elapsed and the future should have a result
// you probably need to change the return type to Either if you use this approach,
// or change this to Option type and ignore the failure, assuming that the exception is logged already 
val result: Either[Throwable, Int] = fReady.value match {
  case Some(Success(a)) => Right(a) // you can edit this to compute a JsValue from `a` if you want
  case Some(Failure(b)) => Left(b)
  case None => Left(new RuntimeException("Unexpected"))
}

// can be return type or edit this
result