如何将 Http 管理到 Akka Stream 并将消息发送到 Kafka？答案

【问题标题】：How to manage Http into Akka Stream and send the message to Kafka?如何将 Http 管理到 Akka Stream 并将消息发送到 Kafka？
【发布时间】：2019-03-15 03:21:03
【问题描述】：

我从 Akka Streams 开始，我想构建一个服务器作为 Stream 接收 Http.IncomingConnection 并将接收到的消息作为 plainSink 发送到 Kafka。

我声明我的来源如下：

val serverSource: Source[Http.IncomingConnection, Future[Http.ServerBinding]] =
    Http().bind(interface = "localhost", port = "8080")

然后，我想从 HttpRequest 的正文中提取消息（字符串），最后将字符串发送到 Kafka。流程如下所示：

val bindingFuture: Future[Http.ServerBinding] = serverSource
     .map(???) //Here, I need to extract the message
     .map(message => new ProducerRecord[String, String](topic, message.result(2 seconds)))
     .runWith(akka.kafka.scaladsl.Producer.plainSink(producerSettings))

但是，我不知道如何提取消息。我想做这样的事情：

val requestHandler: HttpRequest => HttpResponse = {
    case HttpRequest(POST, Uri.Path("/publish"), _, _, _) => {
      HttpResponse(202, entity = "Message sent to Kafka!")
    }
    case r: HttpRequest =>
      r.discardEntityBytes() // important to drain incoming HTTP Entity stream
      HttpResponse(404, entity = "Unknown resource!")
  }

但是，使用connection handleWithSyncHandler requestHandler 我无法让消息跟随流处理。而且，我还想在/publish URI 下获取任何请求，或者在流中的其他情况下返回 404。

可以这样做吗？

【问题讨论】：

标签： scala apache-kafka akka akka-stream akka-http

【解决方案1】：

改用指令

Routing DSL 比手动处理 HttpRequest 更容易使用：

val route : Route = 
  post {
    path("publish") {
      extractRequestEntity { entity =>
        onComplete(entity.toStrict(10.seconds).map(_.data.utf8String){ message =>
          Producer.plainSink(producerSettings)(
            new ProducerRecord[String, String](topic, message.result(2 seconds))
          )
          complete(StatusCodes.OK)
        } 
      }
    }
  }

现在可以传入它来处理传入的请求：

Http().bindAndHandle(
  route,
  "localhost",
  8080
)

【讨论】：