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)