json复制编辑{
  "message": "消息内容",
  "to": "接收者ID或群ID",
  "type": "消息类型，0-私聊，1-群聊",
  "from": "发送者ID",
  "group_id": 1, 
  "createTime": "2025-06-18T13:24:00Z"
}

go复制编辑// 绑定 routing key（加入群聊）
func BindRoutingKey(ch *amqp.Channel, queueName, exchangeName, routingKey string) error {
    return ch.QueueBind(queueName, routingKey, exchangeName, false, nil)
}

// 解绑 routing key（退出群聊）
func UnbindRoutingKey(ch *amqp.Channel, queueName, exchangeName, routingKey string) error {
    return ch.QueueUnbind(queueName, routingKey, exchangeName, nil)
}

go复制编辑r.GET("/ws", func(c *gin.Context) {
    // WebSocket 升级
    upgrader := websocket.Upgrader{CheckOrigin: func(r *http.Request) bool { return true }}
    conn, err := upgrader.Upgrade(c.Writer, c.Request, nil)
    if err != nil {
        panic(err)
    }
    defer conn.Close()

    // 假设从请求参数或认证中获取用户ID
    userId := c.Query("user_id")
    queueName := "queue.user." + userId
    exchangeName := "message.topic"

    ch, _ := rabbitMQConn.Channel()
    defer ch.Close()

    // 订阅用户专属队列，消费消息推送到 WebSocket
    go func() {
        msgs, _ := ch.Consume(queueName, "", true, false, false, false, nil)
        for d := range msgs {
            err := conn.WriteMessage(websocket.TextMessage, d.Body)
            if err != nil {
                break
            }
        }
    }()

    // 读取客户端消息并发布到交换机
    for {
        _, msg, err := conn.ReadMessage()
        if err != nil {
            break
        }
        // 解析消息，确定 routing key
        var payload MessagePayload
        json.Unmarshal(msg, &payload)

        var routingKey string
        if payload.Type == "0" { // 私聊
            routingKey = "private.id." + payload.To
        } else if payload.Type == "1" { // 群聊
            routingKey = "group.id." + strconv.Itoa(payload.GroupId)
        }

        ch.Publish(exchangeName, routingKey, false, false, amqp.Publishing{
            ContentType: "application/json",
            Body:        msg,
        })
    }
})