Redis®*: 避免 Redis 和 Python 中的 "Connection Closed by Server" 错误
本文档属于使用 Python指南的一部分。您可以在此处查看完整指南：如何使用 Python 连接 Redis。
👋 欢迎来到 Stackhero 文档！
Stackhero 提供现成的 Redis 云 解决方案，具有众多优势，包括：
- 包含
Redis Commander网页界面。- 无限消息大小和传输。
- 只需点击即可轻松更新。
- 由专用私有 VM提供的最佳性能和强大安全性。
节省时间，简化生活：只需 5 分钟即可试用 Stackhero 的 Redis 云托管 解决方案！
您可能会遇到 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 功能时，确保您的应用程序调用 get_message() 或 listen() 的频率高于指定的 health_check_interval（在此示例中为每 10 秒）。您可以参考 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)