Redis golang 客户端定期丢弃错误的 PubSub 连接 (EOF)答案

【问题标题】：Redis golang client periodically discarding bad PubSub connection (EOF)Redis golang 客户端定期丢弃错误的 PubSub 连接 (EOF)
【发布时间】：2019-09-03 15:25:23
【问题描述】：

我做了什么：

我正在使用来自github.com/go-redis/redis 的golang Redis 库。我的客户在名为“控制”的 PubSub 频道上收听。每当有消息到达时，我都会处理它并继续接收下一条消息。我没完没了地听，消息可能经常来，有时几天都没有。

我的期望：

我希望 redis 通道能够无休止地保持打开状态并在消息发送时接收消息。

我的经历：

通常它运行好几天，但每隔一段时间client.Receive() 返回EOF 错误。在此错误之后，客户端不再在该通道上接收消息。在内部，redis 客户端向标准输出打印以下消息：

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
}

【问题讨论】：

您需要检查几件事。 1. 处理断网 2. 处理IdleTimeout 看看是否还有报错

标签： go redis publish-subscribe

【解决方案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())
        }
    }
}

【讨论】：

pubsub 频道是如何工作的？如果使用这种方式超过 2 次是否会降低性能？
@DarmawanZulkifli Redis PubSub 应该可以在多对多的环境中执行。 redis.io/topics/pubsub

【解决方案2】：

我的看法是，如果 Redis 认为客户端处于空闲状态，它可能会断开您的客户端。

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

使用ReceiveTimeout 而不是Receive。
如果操作超时，请发送Ping 并等待回复。
冲洗，重复。

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

我会开始here。

【讨论】：