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

Posted 2023-02-16

【中文标题】Redis golang 客户端定期丢弃错误的 PubSub 连接 (EOF)

【英文标题】：Redis golang client periodically discarding bad PubSub connection (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 看看是否还有报错

【参考方案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。