登录
首页 >  Golang >  Go问答

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

来源:stackoverflow

时间:2024-04-03 21:57:34 484浏览 收藏

对于一个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学习网公众号!

声明:本文转载于:stackoverflow 如有侵犯,请联系study_golang@163.com删除
相关阅读
更多>
最新阅读
更多>
课程推荐
更多>