Valkey: 避免 "Connection Closed by Server" 錯誤
本文件是使用 Python指南的一部分。請在此處查看完整指南：如何將 Valkey 與 Python 連接。
👋 歡迎來到 Stackhero 文件！
Stackhero 提供現成的 Valkey cloud 解決方案，帶來多種好處，包括：
- 包含
Redis Commander網頁介面。- 無限制的訊息大小和傳輸。
- 只需一鍵即可輕鬆更新。
- 由專用私有 VM提供的最佳效能和強大安全性。
節省時間並簡化您的生活：只需 5 分鐘即可嘗試 Stackhero 的 Valkey cloud hosting 解決方案！
如果您的應用程式在一段時間內未與 Valkey 互動，可能會出現 redis.exceptions.ConnectionError: Connection closed by server 錯誤，導致自動斷開連接。為了避免這種情況，您可以設置 health_check_interval 參數，如下所示：
r = redis.from_url(
'rediss://default:<password>@<XXXXXX>.stackhero-network.com:<PORT_TLS>',
health_check_interval=10,
socket_connect_timeout=5,
retry_on_timeout=True,
socket_keepalive=True
)
使用 Valkey 的 Pub/Sub 功能時，redis-py 庫期望像 get_message() 或 listen() 這樣的函數被調用的頻率高於 health_check_interval。在我們的範例中，我們將此間隔設置為 10 秒，因此請確保至少每 10 秒調用一次 get_message() 或 listen()（參考 redis-py 官方文件）。
如果不這樣做，您可能會遇到相同的連接錯誤。為了避免這種情況，考慮定期調用 check_health()。
以下是實現方法：
import redis
import threading
# 連接到 Valkey
r = redis.from_url(
'rediss://default:<password>@<XXXXXX>.stackhero-network.com:<PORT_TLS>',
health_check_interval=10,
socket_connect_timeout=5,
retry_on_timeout=True,
socket_keepalive=True
)
# 創建一個 PubSub 實例
p = r.pubsub()
# 訂閱頻道 "test"
p.subscribe('test')
# 創建一個函數，每 5 秒調用一次 `check_health`
def valkey_auto_check(p):
t = threading.Timer(5, valkey_auto_check, [p])
t.start()
p.check_health()
# 調用 valkey_auto_check 函數
valkey_auto_check(p)