aio_pika, aio_pika и завершение работы клиента

import logging
import aio_pika
from aio_pika.exceptions import AMQPError
from app import RABBITMQ_USER, RABBITMQ_PASS, RABBITMQ_HOST, QUEUE_EVENT
import json
import threading
import asyncio
from app.utils.event_service import EventService
 
 
class RabbitConsumer:
    _url: str
    connection: aio_pika.abc.AbstractRobustConnection
    channel: aio_pika.abc.AbstractRobustChannel
 
    event_service: EventService
 
    def __init__(self):
        self._url = f"amqp://{RABBITMQ_USER}:{RABBITMQ_PASS}@{RABBITMQ_HOST}/"
 
    async def rabbitmq_consumer(self):
 
        self.event_service = EventService()
        self.loop = asyncio.get_event_loop()  # Сохраняем цикл событий
        self.stop_event = threading.Event()
        while not self.stop_event.is_set():
            try:
                _url = f"amqp://{RABBITMQ_USER}:{RABBITMQ_PASS}@{RABBITMQ_HOST}/"
                logging.info(f'Запуск RabbitMQ: {_url}')
                self.connection = await aio_pika.connect_robust(_url)
                async with self.connection:
                    self.channel = await self.connection.channel()
                    queue = await self.channel.declare_queue(QUEUE_EVENT, durable=True)
                    async for message in queue:
                        async with message.process():
                            try:
                                # Обработка сообщения
                                text = message.body.decode()
                                await self.event_service.event_worker(json.loads(text))
                            except Exception as e:
                                logging.error(f"Неизвестная ошибка при отправке сообщения: {e}")
                                print(e)
            except AMQPError as e:
                logging.error(f"Ошибка подключения к RabbitMQ: {e}")
            except Exception as e:
                logging.error(f"Неизвестная ошибка при подключении к RabbitMQ: {e}")
 
            # Повторяем подключение каждые 5 секунд в случае ошибки
            await asyncio.sleep(5)
            # Закрытие соединения и канала
 
    async def on_shutdown(self, consumer_thread):
        logging.info("Stop RabbitMQ")
        # Останавливаем поток
        self.stop_event.set()  # Устанавливаем флаг для остановки потребителя
        if self.channel:
            logging.info("Stop RabbitMQ Channel")
            try:
                await asyncio.wait_for(self.channel.close(), timeout=3)
            except asyncio.TimeoutError:
                print("Тайм-аут при закрытии канала")
                # await self.channel.close()
        if self.connection:
            logging.info("Stop RabbitMQ Connection")
            # try:
            #     run_in_executor(None, self.connection.close)
            # except asyncio.TimeoutError:
            #     print("Тайм-аут при закрытии канала")
            # await self.loop.run_in_executor(None, self.connection.close)

def start_rabbitmq_consumer():
    loop = asyncio.new_event_loop()  # Создаем новый event loop для потока
    asyncio.set_event_loop(loop)  # Устанавливаем новый event loop
    loop.run_until_complete(rabbit_consumer.rabbitmq_consumer())  # Запускаем асинхронную функцию в потоке
 
 
# Запуск потока для RabbitMQ
consumer_thread = threading.Thread(target=start_rabbitmq_consumer)
consumer_thread.start()
 
 
async def on_shutdown() -> None:
    ..... Код логики завершения работы бота ....
    logging.info("Stop Bot")
    await rabbit_consumer.on_shutdown(consumer_thread)
    # Ждем завершения потока
    logging.info("Join to RabbitMQ....")
    consumer_thread.join(timeout=2)

import logging
import aio_pika
from aio_pika.exceptions import AMQPError
import json
import asyncio
from app import RABBITMQ_USER, RABBITMQ_PASS, RABBITMQ_HOST, QUEUE_EVENT
from app.utils.event_service import EventService
 
class RabbitConsumer:
    _url: str
    connection: aio_pika.abc.AbstractRobustConnection
    channel: aio_pika.abc.AbstractRobustChannel
    event_service: EventService
    stop_event: asyncio.Event  # Асинхронное событие для остановки
 
    def __init__(self):
        self._url = f"amqp://{RABBITMQ_USER}:{RABBITMQ_PASS}@{RABBITMQ_HOST}/"
        self.stop_event = asyncio.Event()
 
    async def rabbitmq_consumer(self):
        self.event_service = EventService()
        while not self.stop_event.is_set():
            try:
                _url = f"amqp://{RABBITMQ_USER}:{RABBITMQ_PASS}@{RABBITMQ_HOST}/"
                logging.info(f'Запуск RabbitMQ: {_url}')
                self.connection = await aio_pika.connect_robust(_url)
                async with self.connection:
                    self.channel = await self.connection.channel()
                    queue = await self.channel.declare_queue(QUEUE_EVENT, durable=True)
                    async for message in queue:
                        async with message.process():
                            try:
                                text = message.body.decode()
                                await self.event_service.event_worker(json.loads(text))
                            except Exception as e:
                                logging.error(f"Неизвестная ошибка при отправке сообщения: {e}")
            except AMQPError as e:
                logging.error(f"Ошибка подключения к RabbitMQ: {e}")
            except Exception as e:
                logging.error(f"Неизвестная ошибка при подключении к RabbitMQ: {e}")
            
            # Повторяем подключение через 5 секунд при ошибке
            await asyncio.sleep(5)
 
    async def on_shutdown(self):
        logging.info("Stop RabbitMQ")
        self.stop_event.set()  # Устанавливаем флаг остановки
        if self.channel:
            logging.info("Stop RabbitMQ Channel")
            try:
                await asyncio.wait_for(self.channel.close(), timeout=3)
            except asyncio.TimeoutError:
                logging.error("Тайм-аут при закрытии канала")
        if self.connection:
            logging.info("Stop RabbitMQ Connection")
            try:
                await asyncio.wait_for(self.connection.close(), timeout=3)
            except asyncio.TimeoutError:
                logging.error("Тайм-аут при закрытии соединения")
 
# Запуск RabbitMQ в отдельном потоке с помощью asyncio.to_thread
async def start_rabbitmq_consumer():
    rabbit_consumer = RabbitConsumer()
    await rabbit_consumer.rabbitmq_consumer()
 
# Логика завершения работы бота
async def on_shutdown():
    logging.info("Stop Bot")
    # Закрываем RabbitMQ соединение
    await rabbit_consumer.on_shutdown()
    logging.info("Завершение работы бота")