Redis®*: Eviter l'erreur "Connection Closed by Server" avec Redis et Python

Cette documentation fait partie du guide Utilisation avec Python. Consultez le guide complet ici : Comment connecter Redis avec Python.

👋 Bienvenue sur la documentation de Stackhero !

Stackhero propose une solution Redis cloud prête à l'emploi qui offre de nombreux avantages, notamment :

• Interface web Redis Commander incluse.
• Taille des messages et transferts illimités.
• Mises à jour simplifiées en un clic.
• Performance optimale et sécurité renforcée grâce à une VM privée et dédiée.

Gagnez du temps et simplifiez-vous la vie : il suffit de 5 minutes pour essayer la solution Redis cloud hosting de Stackhero !

Vous pouvez rencontrer l'erreur redis.exceptions.ConnectionError: Connection closed by server, qui peut survenir en cas d'inactivité de votre application Python, entraînant la fermeture automatique de la connexion. Lorsque votre application tente de se reconnecter, cela peut échouer et générer cette erreur.

Pour éviter cela, il est conseillé de définir le paramètre health_check_interval dans votre connexion Redis comme ceci :

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
)

Lorsque vous utilisez la fonctionnalité Pub/Sub de Redis, assurez-vous que votre application appelle get_message() ou listen() plus fréquemment que l'intervalle défini par health_check_interval (dans cet exemple, toutes les 10 secondes). Vous pouvez consulter la documentation officielle de redis-py pour plus d'informations.

Si ces appels ne sont pas effectués dans l'intervalle, vous risquez toujours de rencontrer l'erreur "Connection closed by server". Une solution efficace consiste à utiliser régulièrement la fonction check_health().

Voici comment l'implémenter :

import redis
import threading

# Connexion à 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
)

# Création d'une instance PubSub
p = r.pubsub()

# S'abonner au canal "test"
p.subscribe('test')

# Créer une fonction qui appelle `check_health` toutes les 5 secondes
def redis_auto_check(p):
  t = threading.Timer(5, redis_auto_check, [p])
  t.start()
  p.check_health()

# Appeler la fonction redis_auto_check
redis_auto_check(p)