Redis®*: 避免 Redis 和 Python 中的 "Connection Closed by Server" 錯誤
此文件屬於使用 Python指南的一部分。請在此處查看完整指南：如何將 Redis 與 Python 連接。
👋 歡迎來到 Stackhero 文件！
Stackhero 提供一個即用型的 Redis cloud 解決方案，帶來多項好處，包括：
- 包含
Redis Commander網頁介面。- 無限制的訊息大小和傳輸。
- 只需一鍵即可輕鬆進行 更新。
- 由 專用私有 VM 提供的最佳 效能 和強大 安全性。
節省時間，簡化生活：只需 5 分鐘 即可嘗試 Stackhero 的 Redis cloud hosting 解決方案！
您可能會遇到 redis.exceptions.ConnectionError: Connection closed by server 錯誤，這可能是由於您的 Python 應用程式不活動，導致連接自動關閉。當您的應用程式嘗試重新連接時，可能會失敗，從而導致此錯誤。
為了減輕這種情況，考慮在您的 Redis 連接中設置 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
)
在使用 Redis 的 Pub/Sub 功能時，確保您的應用程式比指定的 health_check_interval（在此範例中為每 10 秒）更頻繁地調用 get_message() 或 listen()。您可以參考 redis-py 官方文檔 以獲取更多詳細信息。
如果這些調用未在間隔內完成，您可能仍會遇到 "Connection closed by server" 錯誤。一個實用的解決方案是定期使用 check_health() 函數。
以下是如何實現它：
import redis
import threading
# 連接到 Redis
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 redis_auto_check(p):
t = threading.Timer(5, redis_auto_check, [p])
t.start()
p.check_health()
# 調用 redis_auto_check 函數
redis_auto_check(p)