我做了什么：我正在使用golang來自 的 Redis 庫github.com/go-redis/redis。我的客戶端監(jiān)聽一個名為“control”的 PubSub 通道。每當(dāng)消息到達時，我都會處理它并繼續(xù)接收下一條消息。我沒完沒了地聽著，信息經(jīng)常出現(xiàn)，有時好幾天都沒有。我的期望：我希望 Redis 通道能夠無限地保持打開狀態(tài)并在發(fā)送消息時接收消息。我的經(jīng)歷：通常它會運行幾天，但偶爾client.Receive()會返回EOF錯誤。發(fā)生此錯誤后，客戶端不再在該通道上接收消息。在內(nèi)部，redis 客戶端向 stdout 打印以下消息：redis: 2019/08/29 14:18:57 pubsub.go:151: redis: 丟棄壞的 PubSub 連接: EOF免責(zé)聲明：我不確定這個錯誤是導(dǎo)致我停止接收消息的原因，它只是看起來相關(guān)。附加問題：我想了解為什么會發(fā)生這種情況，如果這是正常的，并且client.Subscribe()每當(dāng)我遇到這種行為時重新連接到頻道是否是一個很好的補救措施，或者我應(yīng)該解決根本問題，無論它是什么。代碼：這是處理我的客戶端的完整代碼（連接到 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)) }
3 回答
慕容708150
TA貢獻1831條經(jīng)驗 獲得超4個贊
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())
}
}
}
繁星點點滴滴
TA貢獻1803條經(jīng)驗 獲得超3個贊
我不知道如果這是正確的方法，但在創(chuàng)建新的 Redis 客戶端時，將ReadTimeout屬性設(shè)置為-1解決了我的問題。
redisClient := redis.NewClient(&redis.Options{
Addr: addr,
Password: redisConf.Password,
DB: 0, // Default DB
ReadTimeout: -1,
})
注意：我使用的是 go-redis/v9
冉冉說
TA貢獻1877條經(jīng)驗 獲得超1個贊
我的看法是，如果 Redis 認為客戶端空閑，它可能會斷開你的客戶端的連接。
解決這個問題的方法似乎是這樣的：
使用
ReceiveTimeout而不是Receive.如果操作超時，則發(fā)出
Ping并等待回復(fù)。沖洗，重復(fù)。
這樣，您就可以確保連接上存在一些流量，無論是否實際發(fā)布了任何數(shù)據(jù)。
- 3 回答
- 0 關(guān)注
- 509 瀏覽
添加回答
舉報
0/150
提交
取消