当前位置：首页 > 文章列表 > Golang > Go问答 > 从 kafka 发送消息到 websocket

从 kafka 发送消息到 websocket

来源：stackoverflow 2024-03-17 12:18:31 0浏览收藏

**文章首段摘要：** 在 Golang 中使用 Gin 框架创建的 Web 服务中，用户想要将从 Kafka 主题接收的消息流式传输到 WebSocket 中。他们使用通道来接收消息，并在 Gin 框架中创建 WebSocket 端点，以使用通道将消息发送到连接的客户端。但是，他们的实现遇到了几个问题，包括仅显示第一条消息、连接快速关闭以及通道阻塞的问题。本文提供了详细的解决方案，包括更改客户端读取泵以丢弃接收到的消息、更改 Kafka 读取循环以将消息发送到 Hub 广播通道、删除合并消息的代码，以及使用 select 语句和保持活动 WebSocket 连接来解决问题。

问题内容

我在 golang 中使用 gin-framework 创建了一个 web 服务。在这个项目中，我还使用了一些来自特定主题的 kafka 消息。我想要实现的目标是将我从主题收到的消息倒入 websocket 中。因此，通信只是一种方式，超过 1 个人可以连接到 web 套接字并查看传入的消息。我想使用通道，所以在接收 kafka 消息的函数中我有这样的东西：

ch <- kafkamessage

在 gin 框架中，我创建了这样的东西：

requestrouterrpcv1.get("wf-live", wsworkflowlive)

func wsworkflowlive(c *gin.context) {
    ws, err := upgrader.upgrade(c.writer, c.request, nil)
    if err != nil {
        log.println("error get connection")
        log.fatal(err)
    }
    defer ws.close()

    err = ws.writejson(<-ch)

    if err != nil {
        log.println("error write message: " + err.error())
    }
}

var upgrader = websocket.upgrader{
    checkorigin: func(r *http.request) bool {
        return true
    },
}

这里是我用来测试 websocket 的 html 片段：

<!DOCTYPE html>
   <html lang="en">
     <head>
       <meta charset="UTF-8" />
       <title>index</title>
    </head>
     <body>
       <h1>test websocket</h1>
       <p id="message-json"></p>
      <p id="message-text"></p>
            <script>
        function jsonWS() {
          var ws = new WebSocket("ws://ws.acme.com/ws/v1/wf-live");

          ws.onmessage = function (evt) {
            console.log("Received Message: " + evt.data);
            document.getElementById("message-json").innerText += evt.data;
          };

          ws.onclose = function (evt) {
            console.log("Connection closed.");
          };
        }

        // Start websocket
        jsonWS();
      </script>
    </body>
  </html>

但是我错过了一些东西，我是一个新手，因为一旦收到第一条kafka消息，我就会出现以下错误行为：

仅显示第一条消息，之后连接会快速关闭
要查看第二个，我必须刷新页面，这不是 websocket 方式
因为连接已关闭，所以通道不是红色的，因此它卡在 cosume 函数中，直到我读取它。我不能有这种行为
为了避免第 3 点，我认为我必须有一种机制，仅当连接一个或多个 ws 时才将消息发送到通道。

正确答案

Gorilla chat example 很接近您的需求。该示例将从任何客户端收到的消息广播到所有连接的客户端。请执行以下操作以使代码适应您的用例：

更改 client read pump 以丢弃收到的消息，而不是向中心发送消息。

func (c *client) readpump() {
    defer func() {
        c.hub.unregister <- c
        c.conn.close()
    }()
    c.conn.setreadlimit(maxmessagesize)
    c.conn.setreaddeadline(time.now().add(pongwait))
    c.conn.setponghandler(func(string) error { 
    c.conn.setreaddeadline(time.now().add(pongwait)); return nil })
    for {
        if _, _, err := c.nextreader(); err != nil {
          break
        }
    }
}

更改 kafka 读取循环以将消息发送到 Hub.broadcast 通道。

for {
    msg, err := c.ReadMessage(xxx)
    if err != nil {
          // handle error
    }
    hub.broadcast <- msg
}

删除 the code that coalesces messages in the client send queue to a single websocket message 或调整客户端以在单个 websocket 消息中处理多个 kafka 消息。

你很接近了。只是缺少两件事。

1- 使用 select 语句不断从 kafka 通道接收新消息。

2- 保持活跃的 websocket 连接。这个answer有更多细节

请告诉我这是否适合您。

好了，本文到此结束，带大家了解了《从 kafka 发送消息到 websocket》，希望本文对你有所帮助！关注golang学习网公众号，给大家分享更多Golang知识！

版本声明

本文转载于：stackoverflow 如有侵犯，请联系study_golang@163.com删除

如何使用 json marshal 获取地图值

上一篇: 如何使用 json marshal 获取地图值

下一篇: 更改默认的 MongoDB ObjectID 生成器

查看更多