akka alppaka producer如何与akka http集成

pgvzfuti 于 2021-06-06 发布在 Kafka

关注(0)|答案(1)|浏览(414)

大家好，我是如何通过http akka ad获取数据并通过alpakka-kafka连接器发送到kafka的，这是我的来源：

object WebTrack extends App  with  Directives with LazyLogging {
val host = "localhost"
val port = 7070

val authorization = "Authorization"

val route = withSizeLimit(96239727) {
  post {
    headerValueByName("Authorization") { auth =>
      entity(as[data]) { trans =>
        //**************************I need send data to Alpakka Kafka connector********************************

        complete(HttpEntity(ContentTypes.`text/html(UTF-8)`, "<h1>Say hello to akka-http</h1>"))
      }
    }
  }
}

val bindingFuture = Http().bindAndHandleAsync(Route.asyncHandler(route), host, port)
.onComplete {
  case Success(_) => {
    logger.debug("This is very convenient ;-)")
    println(s"Server online at http://localhost:7070 \\nPress RETURN to stop...")
  }
  case Failure(e) => {
    println("Error Bind Http().bindAndHandleAsync")
  }
}

}

scala akka apache-kafka

来源：https://stackoverflow.com/questions/53285060/how-can-akka-alppaka-producer-integration-with-akka-http

1条答案

按热度按时间

np8igboo1#

有这么多不同的事情，你可以做与阿克卡流
这里只有一个使用 Producer Flume，根据文件https://doc.akka.io/docs/akka-stream-kafka/current/producer.html. 一定要退房 passThrough 它可以做的一些事情：

implicit val system: ActorSystem = ActorSystem("my-system")
  implicit val materializer: ActorMaterializer = ActorMaterializer()
  implicit val executionContext: ExecutionContextExecutor = system.dispatcher

  case class User(firstName:String, lastName:String)
  object UserJsonSupport extends DefaultJsonProtocol with SprayJsonSupport {
    implicit val PortofolioFormats = jsonFormat2(User)
  }

  import UserJsonSupport._
  val config = system.settings.config.getConfig("akka.kafka.producer")
  val producerSettings = ProducerSettings(config, new StringSerializer, new StringSerializer)

  val route = withSizeLimit(96239727) {
    post {
      headerValueByName("Authorization") { auth =>
        entity(as[User]) { user: User =>
          Source.single(user)
            .map(user => new ProducerRecord[String, String]("last_names_topic", user.lastName, user.firstName))
            .runWith(Producer.plainSink(producerSettings))
          complete(HttpEntity(ContentTypes.`text/html(UTF-8)`,
            "<h1>Say hello to akka-http</h1>"))
        }
      }
    }
  }

赞(0）回复(0）举报 2021-06-07

我来回答

akka alppaka producer如何与akka http集成

1条答案

相关问题

热门标签

最新问答