0

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

这来自我的另一个问题：

如何将akka流kafka（reactive-kafka）集成到akka http应用程序中？

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

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)))
          }
        }
      }
    }
  }

但是，onComplete(producer send producerRecord)正在生成以下类型不匹配错误：

[错误] 发现：Future[org.apache.kafka.clients.producer.RecordMetadata]（在 java.util.concurrent 中）[错误] 需要：Future[org.apache.kafka.clients.producer.RecordMetadata]（在 scala.并发）[错误] onCompleteRecordMetadata { _ =>

有没有办法解决这个问题，也许通过使用 Producer 作为接收器（http://doc.akka.io/docs/akka-stream-kafka/current/producer.html#producer-as-a-sink）而不是java producer.send函数？

4

2 回答 2

3

您可以利用Cake 的基于 Scala 的 Kafka 客户端，它将完成运行 Java 期货的工作并为您提供 Scala 期货。一旦确定创建的是 acakesolutions.kafka.KafkaProducer而不是 a org.apache.kafka.clients.producer.KafkaProducer，其余代码实际上应该保持不变。

或者，您可以利用Reactive Kafka解决这个问题，同时继续使用高级 Akka HTTP DSL。您可以通过将您的生产者记录运行到 Kafka Sink 来做到这一点，这样：

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)))
        }
于 2017-09-18T09:49:02.690 回答
3

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

具体来说，您可以使用FutureConverters.toScala(producer.send(producerRecord))将java.util.concurrent.Futurescala.concurrent.Future

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

于 2017-09-22T11:10:21.300 回答