Valkey: 使用 Python

如何将 Valkey 与 Python 连接

👋 欢迎来到 Stackhero 文档！

Stackhero 提供现成的 Valkey 云解决方案，具有众多优势，包括：

包含 Redis Commander web UI。

无限的消息大小和传输。

只需点击即可轻松完成更新。

由私有和专用 VM提供的最佳性能和强大安全性。

节省时间并简化您的生活：只需 5 分钟即可试用 Stackhero 的 Valkey 云托管解决方案！

选择用于 Valkey 集成的 Python 库

为了无缝集成您的应用程序与 Valkey，建议使用 Python 的 redis 库。这个库非常适合连接到 Valkey，一个 Redis 的开源分支。您可以通过运行以下命令轻松安装：

pip install redis
pip freeze > requirements.txt

将 Python 连接到 Valkey

让我们探讨如何将您的 Python 应用程序连接到 Valkey。以下是使用默认设置的简单示例：

import 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
)

为了增强安全性，建议使用环境变量管理凭据。以下是操作方法：

import os
import redis

r = redis.from_url(
  os.environ.get("STACKHERO_VALKEY_URL_TLS"),
  health_check_interval=10,
  socket_connect_timeout=5,
  retry_on_timeout=True,
  socket_keepalive=True
)

确保您的环境变量包含如下条目：STACKHERO_VALKEY_URL_TLS=rediss://default:<password>@<XXXXXX>.stackhero-network.com:<PORT_TLS>。

在 Valkey 和 Python 中使用 Pub/Sub

在 Python 中利用 Valkey 的发布/订阅 (Pub/Sub) 功能非常简单。以下是一个示例：

import redis

# 连接到 Valkey
r = redis.from_url(
  'rediss://default:<password>@<XXXXXX>.stackhero-network.com:<PORT_TLS>',
  health_check_interval=10,
  socket_connect_timeout=5,
  socket_keepalive=True,
  retry_on_timeout=True
)

# 创建一个 PubSub 实例
p = r.pubsub()

# 订阅频道 "test"
p.subscribe('test')

# 向频道 "test" 发布消息
r.publish('test', '这是一条测试消息')

# 从频道 "test" 获取第一个可用消息
p.get_message()

# 取消订阅频道 "test"
p.unsubscribe('test')

Valkey Pub/Sub 与 Python 的高级示例

通过以下高级示例扩展您的 Pub/Sub 功能：

# 创建一个 PubSub 实例并忽略订阅消息
p = r.pubsub(ignore_subscribe_messages=True)

# 订阅多个频道
p.subscribe('test-1', 'test-2', ...)

# 取消订阅多个频道
p.unsubscribe('test-1', 'test-2', ...)

# 您还可以使用无参数的 "unsubscribe" 来断开所有已订阅频道的连接
p.unsubscribe()

# 使用模式订阅频道
p.psubscribe('my-*')

# 使用模式取消订阅频道
p.punsubscribe('my-*')

避免 "Connection Closed by Server" 错误

如果您的应用程序在一段时间内未与 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)