在java.util.concurrent中找到。未来需要scala.concurrent.Future

Question

相关:java.util.concurrent.Future的scala.concurrent.Future包装器

这来自我的另一个问题：

如何将akka streams Kafka(reactive-Kafka)集成到akka http应用中？

我有一个AKKA HTTP应用程序，我想在路由中的onComplete函数中向Kafka发送消息/产品记录，如下所示：

val producer : KafkaProducer = new KafkaProducer(producerSettings)

val routes : Route = 
  post {
    entity(as[User]) { user =>
      val createUser = userService.create(user)
      onSuccess(createUser) {
        case Invalid(y: NonEmptyList[Err]) =>  
          complete(BadRequest -> "invalid user")
        case Valid(u: User) => { 
          val producerRecord = 
            new ProducerRecord[Array[Byte], String]("topic1","some message")

          onComplete(producer.send(producerRecord)) { _ =>
            complete(ToResponseMarshallable((StatusCodes.Created, u)))
          }
        }
      }
    }
  }

但是，on complete(producer send producer record)正在生成以下类型不匹配错误:

[错误]已找到：未来[org.apache.kafka.clients.producer.RecordMetadata]（java.util.concurrent）[错误]必填：未来[org.apache.kafka.clients.producer.RecordMetadata]（scala.concurrent）[错误]onCompleteRecordMetadata{_=

有没有办法解决这个问题，也许可以把制作人当作水槽(http://doc.akka.io/docs/akka-stream-kafka/current/producer.html#producer-as-a-sink）而不是java生产者。发送函数？

Answer 1

为了回答您的特定问题，scala-java8-compat库提供了java8和Scala Futures之间的转换器。

具体来说，您可以使用 FutureConverters.toScala（producer.send（producerRecord）） 将 java.util.concurrent.Future 转换为 scala.concurrent.Future

但是，使用本身具有 Scala 友好 API 的客户端库（如上面的 Stefano 所建议的那样）可能会为您带来最佳结果。

Answer 2

你可以利用Cake的基于Scala的Kafka客户端，它将完成运行Java期货的工作，并给你Scala期货。一旦你确保你创建了一个 cakesolutions.kafka.KafkaProducer 而不是 org.apache.kafka.clients.producer.KafkaProducer，你的代码的其余部分实际上应该保持不变。

或者，您可以利用反应式Kafka解决这个问题，同时继续使用高级Akka HTTP DSL。你可以通过将你的制作人记录运行到一个Kafka接收器来实现，如下所示:

val producerSink = Producer.plainSink(producerSettings)

...
        // inside the route
        val producerRecord =
          new ProducerRecord[Array[Byte], String]("topic1", "some message")

        onComplete(Source.single(producerRecord).runWith(producerSink)) { _ =>
          complete(ToResponseMarshallable((StatusCodes.Created, u)))
        }