Valkey: 避免 "Connection Closed by Server" 错误
本文档属于使用 Python指南的一部分。您可以在此处查看完整指南：如何将 Valkey 与 Python 连接。
👋 欢迎来到Stackhero文档！
Stackhero提供现成的Valkey云解决方案，具有众多优势，包括：
- 包含
Redis Commanderweb UI。- 无限的消息大小和传输。
- 只需点击即可轻松更新。
- 由专用私有VM提供的最佳性能和强大安全性。
节省时间，简化生活：只需5分钟即可试用Stackhero的Valkey云托管解决方案！
如果您的应用程序在一段时间内未与 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)