Valkey: Unikanie błędów "Connection Closed by Server"

Ta dokumentacja jest częścią przewodnika Używanie z Pythonem. Pełny przewodnik znajdziesz tutaj: Jak połączyć Valkey z Pythonem.

👋 Witamy w dokumentacji Stackhero!

Stackhero oferuje gotowe do użycia rozwiązanie Valkey cloud, które zapewnia wiele korzyści, w tym:

• Włączony web UI Redis Commander.
• Nieograniczony rozmiar wiadomości i transfery.
• Bezproblemowe aktualizacje za jednym kliknięciem.
• Optymalna wydajność i solidne zabezpieczenia dzięki prywatnej i dedykowanej VM.

Oszczędzaj czas i uprość sobie życie: wystarczy 5 minut, aby wypróbować rozwiązanie Valkey cloud hosting Stackhero!

Błąd redis.exceptions.ConnectionError: Connection closed by server może wystąpić, jeśli Twoja aplikacja nie wchodzi w interakcję z Valkey przez pewien czas, co prowadzi do automatycznego rozłączenia. Aby tego uniknąć, możesz ustawić parametr health_check_interval jak pokazano poniżej:

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
)

Podczas korzystania z funkcji Pub/Sub Valkey, biblioteka redis-py oczekuje, że funkcje takie jak get_message() lub listen() będą wywoływane częściej niż health_check_interval. W naszym przykładzie ustawiliśmy ten interwał na 10 sekund, więc upewnij się, że wywołujesz get_message() lub listen() co najmniej raz na 10 sekund (odwołaj się do oficjalnej dokumentacji redis-py).

Jeśli tego nie zrobisz, możesz napotkać ten sam błąd połączenia. Aby tego uniknąć, rozważ regularne wywoływanie check_health().

Oto jak można to zaimplementować:

import redis
import threading

# Połącz z 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
)

# Utwórz instancję PubSub
p = r.pubsub()

# Subskrybuj kanał "test"
p.subscribe('test')

# Utwórz funkcję, która będzie wywoływać `check_health` co 5 sekund
def valkey_auto_check(p):
  t = threading.Timer(5, valkey_auto_check, [p])
  t.start()
  p.check_health()

# Wywołaj funkcję valkey_auto_check
valkey_auto_check(p)