如何使用 Kafka、Alpakka Kafka、Play Framework 和 Websocket 处理 POST 请求？答案

【问题标题】：How handle a POST request with Kafka, Alpakka Kafka, Play Framework and Websocket?如何使用 Kafka、Alpakka Kafka、Play Framework 和 Websocket 处理 POST 请求？
【发布时间】：2020-07-29 12:58:04
【问题描述】：

假设我有两个 kafka 主题，request_topic 用于我的 Post 请求，response_topic 用于我的回复。

这是模型：

case class Request(requestId: String, body: String)
case class Response(responseId: String, body: String, requestId: String)

这是我的套接字处理程序

def socket = WebSocket.accept[String, String] { req =>
  val requestId = ??? // Generate a unique requestId

  val in: Sink[String, Future[Done]] = Sink.foreach[String]{ msg =>
    val record = new ProducerRecord[String, Request]("request_topic", "key", Request(requestId, msg))
    val producer: KafkaProducer[String, Request] = ???
    Future { producer.send(record).get }
  }

  // Once produced, some stream processing apps will manage to process request and publish the reponse to response_topic
  // The Request and Response object are linked by the requestId field.

  val consumerSettings = ???
  val out: Source[ConsumerRecord[String, Response], _] = Consumer
    .plainSource(consumerSettings, Subscriptions.topics("response_topic"))
    .filter(cr => cr.value.requestId == requestId)
    .map(cr => someResponseString(cr.value))

  Flow.formSinkAndSource(in, out)
}

def someResponseString(res: Response): String = ???

基本上，对于每条传入的消息，我都会向 Kafka 发布一个 Request 对象，然后该请求由一些流处理应用程序（此处未显示）进行处理，并希望将响应发布回 Kafka。

我有一些顾虑：

1 - Alpakka Kafka 连接器会为每条传入消息创建一个新的连接器实例，还是会在 Play 运行时使用相同的实例？

2 - 根据单个 requestId 过滤响应是个好主意，还是应该将整个流发送回每个客户端，让他们根据他们感兴趣的 requestId 过滤响应。

3 - 我错了吗？（我是 Websocket 的真正新手）

提前致谢。

【问题讨论】：

标签： scala websocket apache-kafka playframework alpakka

【解决方案1】：

1) 取决于您如何配置它。例如，在in: Sink 正文中，您将为每条消息创建一个新的KafkaProducer。相反，您应该为整个应用程序设置一个生产者。

我不确定 Akka / Play 的线程模型是如何工作的，但大多数网络服务器为每个传入连接启动一个新线程，线程池中的线程数最多为固定数量。

2) 我想尽快进行过滤，并且尽可能多地在服务器端进行过滤。这节省了返回客户端的带宽。

此外，如果您只想将数据从网络服务器上的 Kafka 推送到一个方向的客户端，您可能需要SSE, not Websocket

【讨论】：