Redis golang 客户端定期丢弃错误的 PubSub 连接（EOF）

Question

我做了什么：我正在使用golang来自 的 Redis 库github.com/go-redis/redis。我的客户端监听一个名为“control”的 PubSub 通道。每当消息到达时，我都会处理它并继续接收下一条消息。我没完没了地听着，信息经常出现，有时好几天都没有。我的期望：我希望 Redis 通道能够无限地保持打开状态并在发送消息时接收消息。我的经历：通常它会运行几天，但偶尔client.Receive()会返回EOF错误。发生此错误后，客户端不再在该通道上接收消息。在内部，redis 客户端向 stdout 打印以下消息：redis: 2019/08/29 14:18:57 pubsub.go:151: redis: 丢弃坏的 PubSub 连接: EOF免责声明：我不确定这个错误是导致我停止接收消息的原因，它只是看起来相关。附加问题：我想了解为什么会发生这种情况，如果这是正常的，并且client.Subscribe()每当我遇到这种行为时重新连接到频道是否是一个很好的补救措施，或者我应该解决根本问题，无论它是什么。代码：这是处理我的客户端的完整代码（连接到 redis、订阅频道、无休止地接收消息）：func InitAndListenAsync(log *log.Logger, sseHandler func(string, string) error) error {    rootLogger = log.With(zap.String("component", "redis-client"))    host := env.RedisHost    port := env.RedisPort    pass := env.RedisPass    addr := fmt.Sprintf("%s:%s", host, port)    tlsCfg := &tls.Config{}    client = redis.NewClient(&redis.Options{        Addr:      addr,        Password:  pass,        TLSConfig: tlsCfg,    })    if _, err := client.Ping().Result(); err != nil {        return err    }    go func() {        controlSub := client.Subscribe("control")        defer controlSub.Close()        for {            in, err := controlSub.Receive()  // *** SOMETIMES RETURNS EOF ERROR ***            if err != nil {                rootLogger.Error("failed to get feedback", zap.Error(err))                break            }            switch in.(type) {            case *redis.Message:                cm := comm.ControlMessageEvent{}                payload := []byte(in.(*redis.Message).Payload)                if err := json.Unmarshal(payload, &cm); err != nil {                    rootLogger.Error("failed to parse control message", zap.Error(err))                } else if err := handleIncomingEvent(&cm); err != nil {                    rootLogger.Error("failed to handle control message", zap.Error(err))                }

Answer 1

pubsub.Channel()我通过遍历从而不是返回的通道来解决断开连接Receive()。

这是新代码：

func listenToControlChannel(client *redis.Client) {

    pubsub := client.Subscribe("control")

    defer pubsub.Close()

    if _, err := pubsub.Receive(); err != nil {

        rootLogger.Error("failed to receive from control PubSub", zap.Error(err))

        return

    }

    controlCh := pubsub.Channel()

    fmt.Println("start listening on control PubSub")

    // Endlessly listen to control channel,

    for msg := range controlCh {

        cm := ControlMessageEvent{}

        payload := []byte(msg.Payload)

        if err := json.Unmarshal(payload, &cm); err != nil {

            fmt.Printf("failed to parse control message: %s\n", err.Error())

        } else if err := handleIncomingEvent(&cm); err != nil {

            fmt.Printf("failed to handle control message: %s\n", err.Error())

        }

    }

}

Answer 2

我不知道如果这是正确的方法，但在创建新的 Redis 客户端时，将ReadTimeout属性设置为-1解决了我的问题。

redisClient := redis.NewClient(&redis.Options{

    Addr:        addr,

    Password:    redisConf.Password,

    DB:          0, // Default DB

    ReadTimeout: -1,

})

注意：我使用的是 go-redis/v9

Answer 3

我的看法是，如果 Redis 认为客户端空闲，它可能会断开你的客户端的连接。

解决这个问题的方法似乎是这样的：

1. 使用ReceiveTimeout而不是Receive.

2. 如果操作超时，则发出Ping并等待回复。

3. 冲洗，重复。

这样，您就可以确保连接上存在一些流量，无论是否实际发布了任何数据。