Здесь уже есть хорошие решения, но вот мой быстрый и грязный для django_redis, который, кажется, не включает функцию ping (хотя я использую более старую версию django и не могу использовать новейшие django_redis).
# assuming rs is your redis connection
def is_redis_available():
# ... get redis connection here, or pass it in. up to you.
try:
rs.get(None) # getting None returns None or throws an exception
except (redis.exceptions.ConnectionError,
redis.exceptions.BusyLoadingError):
return False
return True
Кажется, что все работает отлично. Обратите внимание, что если redis перезагружает и по-прежнему загружает файл .rdb, содержащий записи кэша на диске, тогда он будет бросать BusyLoadingError, хотя это базовый класс - ConnectionError, поэтому прекрасно его поймать.
Вы также можете просто исключить из redis.exceptions.RedisError
, который является базовым классом всех исключений redis.
Другой вариант, в зависимости от ваших потребностей, заключается в создании функций get и set, которые ловят исключения ConnectionError при настройке / получении значений. Затем вы можете продолжать или ждать или что-то еще, что вам нужно сделать (поднять новое исключение или просто выбросить более полезное сообщение об ошибке).
Это может не сработать, если вы абсолютно зависите от установки / получения значений кеша (для моих целей, если кеш отключен для всех, что мы обычно должны «продолжать»), и в этом случае это может иметь смысл иметь исключения и позволить программе / сценарию умереть и вернуть сервер / службу redis в состояние достижимости.