Scala - 从回调中返回值

Question

我对scala编程相当陌生。谁能帮助我解决回调的返回值问题。我如何从调用方法中以JsObject的形式返回一个回调值？我使用的是Play2框架的actor系统。请告诉我，如果我的返回类型是错误的，我应该返回Future，而不是SendToKafka方法中的JsObject。

我有以下代码

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