【问题标题】:How to receive Redis publish message in Go如何在 Go 中接收 Redis 发布消息
【发布时间】:2021-10-20 16:29:12
【问题描述】:

我正在尝试在 Go 中使用 Redis PubSub 来传递/发布消息并在订阅期间检索它。

我已经能够正确设置代码的发布和订阅/PubSub 部分。下面是我的代码。我希望在订阅期间检索的(字符串)消息是 test message. 但是,我的代码输出给出了频道、种类和计数,并且没有显示预期的消息 (test message)。

在 Go 中使用 Redis 发布 / 订阅 Publish 后如何获得预期的消息(test message)? 我觉得我很接近,但我可能在这里遗漏了一个小东西。我对 Redis 很陌生。感谢您的帮助。

以下是我的代码:

package main

import (
    "fmt"
    "log"
    "time"

    "github.com/gomodule/redigo/redis"
)

func main() {
    fmt.Println("Start redis test.")

    c, err := redis.Dial("tcp", "localhost:6379")
    if err != nil {
        log.Println(err)
    } else {
        log.Println("No error during redis.Dial.")
    }
    // defer c.Close()



    /// Publisher.
    c.Do("PUBLISH", "example", "test message")
    /// End here

    /// Subscriber.
    psc := redis.PubSubConn{Conn: c}
    psc.Subscribe("example")
    for {
        switch v := psc.Receive().(type) {
        case redis.Message:
            fmt.Printf("%s: message: %s\n", v.Channel, v.Data)
        case redis.Subscription:
            fmt.Printf("%s: %s %d\n", v.Channel, v.Kind, v.Count)
        case error:
            fmt.Println(v)
        }
    }
    /// End here

}

以下是我的输出: example: subscribe 1

【问题讨论】:

    标签: go redis publish-subscribe redigo


    【解决方案1】:

    我相信你的代码很好;问题是您在订阅有效之前发布了一条消息。例如,试试这个,它将您的发布者放入一个每秒发布一次消息的 goroutine:

    package main
    
    import (
        "fmt"
        "log"
        "time"
    
        "github.com/gomodule/redigo/redis"
    )
    
    func main() {
        fmt.Println("Start redis test.")
    
        c, err := redis.Dial("tcp", "localhost:6379")
        if err != nil {
            log.Println(err)
        } else {
            log.Println("No error during redis.Dial.")
        }
        // defer c.Close()
    
        /// Publisher.
        go func() {
            c, err := redis.Dial("tcp", "localhost:6379")
            if err != nil {
                panic(err)
            }
    
            count := 0
            for {
                c.Do("PUBLISH", "example",
                    fmt.Sprintf("test message %d", count))
                count++
                time.Sleep(1 * time.Second)
            }
        }()
        /// End here
    
        /// Subscriber.
        psc := redis.PubSubConn{Conn: c}
        psc.Subscribe("example")
    
        for {
            switch v := psc.Receive().(type) {
            case redis.Message:
                fmt.Printf("%s: message: %s\n", v.Channel, v.Data)
            case redis.Subscription:
                fmt.Printf("%s: %s %d\n", v.Channel, v.Kind, v.Count)
            case error:
                fmt.Println(v)
            }
    
            time.Sleep(1)
        }
        /// End here
    
    }
    

    运行此程序,您会看到您的订阅者收到一条消息 其次,产生如下输出:

    Start redis test.
    2021/08/18 19:01:29 No error during redis.Dial.
    example: subscribe 1
    example: message: test message 0
    example: message: test message 1
    example: message: test message 2
    example: message: test message 3
    example: message: test message 4
    example: message: test message 5
    

    【讨论】:

      猜你喜欢
      • 2018-02-26
      • 1970-01-01
      • 1970-01-01
      • 2020-06-21
      • 2022-11-11
      • 2011-05-04
      • 1970-01-01
      • 2018-08-22
      • 2021-06-29
      相关资源
      最近更新 更多