Golang：如何确保redis订阅者收到来自redis pubsub的所有消息？

2024-04-04 23:44

短信预约 -IT技能 免费直播动态提醒

对于一个Golang开发者来说，牢固扎实的基础是十分重要的，编程网就来带大家一点点的掌握基础知识点。今天本篇文章带大家了解《Golang：如何确保redis订阅者收到来自redis pubsub的所有消息？》，主要介绍了，希望对大家的知识积累有所帮助，快点收藏起来吧，否则需要时就找不到了！

问题内容

我正在尝试将消息发布到 redis 中动态生成的通道，同时订阅现有通道中的所有消息。以下似乎可行，但根据客户端（浏览器）请求的时间，它无法接收一些消息。

我尝试在 select 语句中对两个 go 通道进行“扇入”，但效果不佳。

package main

import (
    ...
    "github.com/go-redis/redis/v8"
    "github.com/gorilla/websocket"
)

var upgrader = websocket.Upgrader{}
var rd = redis.NewClient(&redis.Options{
    Addr: "localhost:6379",
})
var ctx = context.Background()

func echo(w http.ResponseWriter, r *http.Request) {
    conn, err := upgrader.Upgrade(w, r, nil)
    if err != nil {
        log.Println("websocket connection err:", err)
        return
    }
    defer conn.Close()

    room := make(chan string)
    start := make(chan string)

    go func() {
    loop:
        for {
            sub := rd.Subscribe(ctx)
            defer sub.Close()
            
            channels := []string{}

            for {
                select {
                //it seems that messages are not received when executing this case sometimes
                case channel := <-room:
                    log.Println("channel", channel)
                    channels = append(channels, channel)
                    sub = rd.Subscribe(ctx, channels...)
                    start <- "ok"
                case msg := <-sub.Channel():
                    log.Println("msg", msg)
                    err := conn.WriteMessage(websocket.TextMessage, []byte(msg.Payload))
                    if err != nil {
                        log.Println("websocket write err:", err)
                        break loop
                    }
                }
            }

        }
    }()

    for {
        _, msg, err := conn.ReadMessage()
        if err != nil {
            log.Println("websocket read err:", err)
            break
        }
        log.Println(string(msg))

        chPrefix := strings.Split(string(msg), ":")[0]
        ch := chPrefix + "-channel"
        if string(msg) == "test" || string(msg) == "yeah" {
            room <- ch
            log.Println(ch)
            log.Println(<-start)
        }

        if err := rd.Publish(ctx, ch, msg).Err(); err != nil {
            log.Println("redis publish err:", err)
            break
        }
    }

}

func main() {
    http.Handle("/", http.FileServer(http.Dir("./js")))
    http.HandleFunc("/ws", echo)

    log.Println("server starting...", "http://localhost:5000")
    log.Fatal(http.ListenAndServe("localhost:5000", nil))
}