Valkey: 使用 Python

如何將 Valkey 與 Python 連接

👋 歡迎來到 Stackhero 文件！

Stackhero 提供即用型的 Valkey cloud 解決方案，帶來多項好處，包括：

包含 Redis Commander 網頁介面。

無限制的訊息大小和傳輸。

只需一鍵即可輕鬆完成更新。

由 專用私有 VM 提供的最佳效能和強大 安全性。

節省時間並簡化您的生活：只需 5 分鐘即可嘗試 Stackhero 的 Valkey cloud hosting 解決方案！

選擇 Python 庫以整合 Valkey

要無縫整合您的應用程式與 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)