在网页上显示 Kafka 消息答案

【问题标题】：Display Kafka messages on web page在网页上显示 Kafka 消息
【发布时间】：2019-09-30 19:41:55
【问题描述】：

我有一个带有 Tomcat 服务器的 Java Spring 应用程序，它监听 kafka 主题。我想在网页上以实时模式显示所有消息。因此，当 kafka 消息到达后端时，我想在我的网页上看到它。我不知道将kafka消息直接推送到前端并显示在网页上的好方法。有人可以帮助我提供解决方案和一些可以提供帮助的示例吗？谢谢！

【问题讨论】：

标签： javascript java apache-kafka frontend

【解决方案1】：

我已经为我的上一个雇主用 Java 实现了一个这样的系统，尽管没有使用 Spring/Tomcat。它正在使用来自 Kafka 的消息并在 Web 套接字上提供它们以显示在浏览器中。我遵循的方法是使用akka-stream-kafka 和akka-http 来支持网络套接字。这样做的好处是两者都基于 akka-streams，这使得它很容易适合流式数据。虽然您可以将 akka-http 嵌入到在 tomcat 中运行的 spring 应用程序中，但它可能不再是最自然的选择，因为 spring 框架已经对 kafka 和 websockets 都有自己的支持。但是，如果您不熟悉其中任何一个，那么使用 akka 方法可能是最简单的，并且核心逻辑遵循这些思路（我无法分享工作中的代码，所以只是从文档中的示例中将它们放在一起，未经测试）：

public Route createRoute(ActorSystem system) {
  return path("ws", () -> {
    ConsumerSettings<byte[], String> consumerSettings = ConsumerSettings.create(system, new ByteArrayDeserializer(), new StringDeserializer())
      .withBootstrapServers("localhost:9092")
      .withGroupId(UUID.randomUUID().toString()) //this is so that each client gets all messages. To be able to resume from where a client left off in case of disconnects, you can generate in on the client side and pass in the request
      .withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")

    return handleWebSocketMessages(
      Flow.fromSinkAndSourceCoupled(
        Sink.ignore(),
        Consumer.committableSource(consumerSettings, Subscriptions.topics("topic1"))
          .map(msg -> TextMessage.create(msg.record().value()))
      )
    );
  }
}

要公开这条路线，您可以按照minimalistic example，唯一的区别是您定义的路线需要ActorSystem：

final Http http = Http.get(system);
final ActorMaterializer materializer = ActorMaterializer.create(system);

final Flow<HttpRequest, HttpResponse, NotUsed> routeFlow = createRoute(system).flow(system, materializer);
final CompletionStage<ServerBinding> binding = http.bindAndHandle(routeFlow,
    ConnectHttp.toHost("localhost", 8080), materializer);

一旦您将消息发布到 websocket，前端的代码当然取决于您选择的 UI 框架，从 javascript 使用 ws 消息的最简单代码是：

this.connection = new WebSocket('ws://url-to-your-ws-endpoint');
this.connection.onmessage = evt => { 
  // display the message

为了在 UI 中轻松显示消息，您希望格式更方便，例如 JSON。如果您的 Kafka 消息还不是 JSON，这就是第一个 sn-p 中的反序列化器出现的地方，您可以在反序列化器中将其转换为方便的 JSON 字符串，或者稍后在 Source 对象上调用的 .map() 中执行此操作.

或者，如果轮询是一个选项，您也可以考虑使用现成的Kafka Rest Proxy，那么您只需要构建前端。

【讨论】：