Redis golang 客户端定时清理无效的 PubSub 连接

使用 go-redis 客户端订阅 Redis PubSub 通道时，可能会遇到连接断开问题，导致接收消息失败。此问题是由内部错误“redis: 丢弃坏的 pubsub 连接: eof”引起的。通过将通道范围从 `receive()` 返回的通道更改为 `pubsub.channel()` 返回的通道，并设置 `readtimeout` 为 `-1`，可以解决此连接断开问题。

我做了什么：

我正在使用 github.com/go-redis/redis 中的 golang 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))
                }

            default:
                rootLogger.Warn("Received unknown input over REDIS PubSub control channel", zap.Any("received", in))
            }
        }
    }()
    return nil
}

解决方案

我通过范围从 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())
        }
    }
}

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

redisClient := redis.NewClient(&redis.Options{
    Addr:        addr,
    Password:    redisConf.Password,
    DB:          0, // Default DB
    ReadTimeout: -1,
})

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

理论要掌握，实操不能落！以上关于《Redis golang 客户端定时清理无效的 PubSub 连接》的详细介绍，大家都掌握了吧！如果想要继续提升自己的能力，那么就来关注golang学习网公众号吧！