Golang Redis PubSub 超时

Question

到目前为止我一直在这样做：

import (
    _redis "gopkg.in/redis.v3"
    "strconv"
    "time"
)

type Redis struct {
    Connector   *_redis.Client
    PubSub      *_redis.PubSub
}

var redis *Redis = nil

func NewRedis() bool {
    if redis == nil {
        redis = new(Redis)
        redis.Connector = _redis.NewClient(&_redis.Options{
            Addr: config.RedisHostname + ":" + strconv.FormatInt(config.RedisPort, 10),
            Password: "",
            DB: 0,
        })
        Logger.Log(nil, "Connected to Redis")
        err := redis.Init()
        if err != nil {
            Logger.Fatal(nil, "Cannot setup Redis:", err.Error())
            return false
        }
        return true
    }
    return false
}

func (this *Redis) Init() error {
    pubsub, err := this.Connector.Subscribe("test")
    if err != nil {
        return err
    }
    defer pubsub.Close()
    this.PubSub = pubsub
    for {
        msgi, err := this.PubSub.ReceiveTimeout(100 * time.Millisecond)
        if err != nil {
            Logger.Error(nil, "PubSub error:", err.Error())
            err = this.PubSub.Ping("")
            if err != nil {
                Logger.Error(nil, "PubSub failure:", err.Error())
                break
            }
            continue
        }
        switch msg := msgi.(type) {
            case *_redis.Message:
                Logger.Log(nil, "Received", msg.Payload, "on channel", msg.Channel)
        }
    }
    return nil
}

我的连接器是一个 redis.Client，它正在工作，因为我也能够发布消息。

当我运行我的程序时，出现以下错误： PubSub error: WSARecv tcp 127.0.0.1:64505: i/o timeout

你知道我做错了什么吗？我正在使用这个包： https: //github.com/go-redis/redis

Answer 1

需要注意的一些事项：

• （实现细节）当redis进入PubSub模式时，之后该socket上发生的唯一事情就是PubSub事件，这就是为什么go-redis中的PubSub被抽象成自己的类型
• PubSub 客户端可能会订阅单个订阅者中的多个主题，因此始终存在订阅/取消订阅事件。
• 该接口具有Receive()和ReceiveTimeout(duration)方法，两者都返回线路上的下一个事件；可以是订阅/取消订阅事件和消息事件；（你不一定知道哪个）它们之间的唯一区别是 Receive 会永远阻塞，直到有新消息为止，而 ReceiveTimeout 会在超时时出错。

考虑到这一点，除非您的消息始终超过 10 条/秒（换句话说，消息之间的间隔 <100 毫秒），否则使用这么短的超时是低效的；我认为，由于 golang 有 goroutine，你几乎不应该将其用于ReceiveTimeout实际应用程序，或者使用足够长的超时（例如一分钟）。

考虑到这一点，您的接收循环应该如下所示：

    for {
        msgi, err := this.PubSub.Receive()
        if err != nil {
            Logger.Error(nil, "PubSub error:", err.Error())
            return err
        }
        switch msg := msgi.(type) {
            case *_redis.Message:
                Logger.Log(nil, "Received", msg.Payload, "on channel", msg.Channel)
            default:
                Logger.Log(nil, "Got control message", msg)
        }
    }

如果您的应用程序确实需要使用超时，那么您应该使用类型断言来断言表示*net.OpError超时并将其与其他更严重的错误区分开来。