Valkey: "Connection Closed by Server"エラーの回避

このドキュメントはPythonでの使用ガイドの一部です。完全なガイドはこちらからご覧いただけます：ValkeyをPythonで接続する方法。

👋 Stackheroのドキュメントへようこそ！

Stackheroは、数多くの利点を提供するValkeyクラウドソリューションをすぐに利用できる形で提供しています。

• Redis Commander Web UIが含まれています。
• メッセージサイズと転送が無制限。
• ワンクリックで簡単にアップデート。
• プライベートで専用のVMによる最適なパフォーマンスと強力なセキュリティ。

時間を節約し、生活を簡素化：StackheroのValkeyクラウドホスティングソリューションを試すのに5分しかかかりません！

アプリが一定期間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秒に1回は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)