redis-py не закрывает потоки при выходе

Question

Я использую redis-py 2.10.6 и redis 4.0.11.

Мое приложение использует redis как для db, так и для pubsub. Когда я выключаюсь, я часто получаю либо зависание, либо вылет. Последний обычно жалуется на плохой файловый дескриптор или ошибку ввода-вывода в файле (я их не использую), которая происходит при обработке обратного вызова pubsub, поэтому я предполагаю, что основная проблема такая же: почему-то я не t отключиться должным образом, и пул, используемый моим объектом redis.Redis, жив и работает.

Пример вывода первой ошибки (во время _read_from_socket):

redis.exceptions.ConnectionError: Ошибка при чтении из сокета: (9, 'Плохой дескриптор файла')

В других случаях трассировка стека четко показывает redis/connection.py -> redis/client.py -> threading.py, что доказывает, что redis не убивает используемые потоки.

Когда я запускаю приложение, я запускаю:

self.redis = redis.Redis(host=XXXX, port=XXXX)
self.pubsub = self.redis.pubsub()
subscriptions = {'chan1': self.cb1, 'chan2': self.cb2}  # cb1 and cb2 are functions
self.pubsub.subscribe(**subscriptions)
self.pubsub_thread = self.pubsub.run_in_thread(sleep_time=1)

Когда я хочу выйти из приложения, последняя инструкция, которую я выполняю в main, - это вызов функции в моем классе redis using, реализация которой:

self.pubsub.close()
self.pubsub_thread.stop()
self.redis.connection_pool.disconnect()

Насколько я понимаю, теоретически мне даже не нужно выполнять какие-либо из этих "закрывающих" вызовов, и тем не менее, с ними или без них, я все еще не могу гарантировать полное завершение работы.

У меня вопрос, как я могу гарантировать чистое завершение работы?

Answer 1

PubSubWorkerThreadпроверка класса на self._running.is_set()внутри петли.

Чтобы сделать «чистое отключение», вы должны позвонить self.pubsub_thread._running.clean()чтобы установить для события потока значение false, и оно остановится.

Проверьте, как это работает, здесь: https://redis.readthedocs.io/en/latest/_modules/redis/client.html?highlight=PubSubWorkerThread#

Answer 2

Если у кого-то возникнет этот вопрос, у меня возникла та же проблема, и я добавил daemon=True в качестве дополнительного параметра к вызову run_in_thread.

      self.pubsub_thread = self.pubsub.run_in_thread(sleep_time=1, deamon=True)

Это гарантирует, что поток не переживет завершение программы, и может обеспечить полное завершение работы, которое вы ищете.

Answer 3

Я столкнулся с той же проблемой, и это в основном вызвано неправильной обработкой завершения работы библиотекой redis. Во время очистки поток продолжает обрабатывать новые сообщения и не учитывает ситуации, когда сокет больше не доступен. Немного пощупав код, я не смог найти способ предотвратить дополнительную обработку, просто не дождавшись.

Поскольку это запускается во время фазы выключения и это средство для сторонней библиотеки, меня не слишком беспокоит sleep, но в идеале библиотеку следует обновить, чтобы предотвратить дальнейшие действия при завершении работы.

      self.pubsub_thread.stop()
time.sleep(0.5)
self.pubsub.reset()

Это может стоить журнала проблем или PR в библиотеке redis-py .