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

对于一个Golang开发者来说，牢固扎实的基础是十分重要的，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))
}

正确答案

如果所有消息都表示您不希望丢失任何消息，我建议使用 Redis Streams 而不是 pub/sub。这将确保您不会丢失消息，并且可以在必要时返回流历史记录。

This is an example 使用 Go、Streams 和 websockets 的经验应该可以帮助您朝这个方向开始

以上就是《Golang：如何确保redis订阅者收到来自redis pubsub的所有消息？》的详细内容，更多关于的资料请关注golang学习网公众号！