**文章首段摘要：** 在 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 片段：

   
     
       
       
    
     
       
test websocket
       

      

            
    
  

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

1. 仅显示第一条消息，之后连接会快速关闭
2. 要查看第二个，我必须刷新页面，这不是 websocket 方式
3. 因为连接已关闭，所以通道不是红色的，因此它卡在 cosume 函数中，直到我读取它。我不能有这种行为
4. 为了避免第 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知识！