Ошибка подключения к базе данных, в то время как работник сельдерея простаивает 24 часа - PullRequest
Новая
       61

Ошибка подключения к базе данных, в то время как работник сельдерея простаивает 24 часа

0 голосов
/ 08 мая 2019

У меня есть веб-приложение на основе Django, в котором я использую Kafka для обработки некоторых заказов.Теперь я использую Сельдерей, чтобы назначить Потребителя Кафки для каждой темы.Каждый Потребитель Кафки назначается теме Кафки в форме задач Кафки.Однако примерно через день, когда я отправляю задачу, я получаю следующую ошибку: 

    _mysql.connection.query(self, query)
_mysql_exceptions.OperationalError: (2006, 'MySQL server has gone away')

The above exception was the direct cause of the following exception:

Ниже показано, как выглядит мой файл tasks.py: 

@shared_task
def init_kafka_consumer(topic):
    try:
        if topic is None:
            raise Exception("Topic is none, unable to initialize kafka consumer")
        logger.info("Spawning new task to subscribe to topic")
        params = []
        params.append(topic)
        background_thread = Thread(target=sunscribe_consumer, args=params)
        background_thread.start()
    except Exception :
        logger.exception("An exception occurred while reading message from kafka")

def sunscribe_consumer(topic) :
    try:
        if topic is None:
            raise Exception("Topic is none, unable to initialize kafka consumer")
        conf = {'bootstrap.servers': "localhost:9092", 'group.id': 'test', 'session.timeout.ms': 6000,
                'auto.offset.reset': 'earliest'}
        c = Consumer(conf)
        logger.info("Subscribing consumer to topic "+str(topic[0]))
        c.subscribe(topic)
        # Read messages from Kafka
        try:
            while True:
                msg = c.poll(timeout=1.0)
                if msg is None:
                    continue
                if msg.error():
                    raise KafkaException(msg.error())
                else:
                    try:
                        objs = serializers.deserialize("json", msg.value())
                        for obj in objs:
                            order = obj.object
                            order = BuyOrder.objects.get(id=order.id) #Getting an error while accessing DB
                            if order.is_pushed_to_kafka :
                                return
                            order.is_pushed_to_kafka = True
                            order.save()
                            from web3 import HTTPProvider, Web3, exceptions
                            w3 = Web3(HTTPProvider(INFURA_MAIN_NET_ETH_URL))
                            processBuyerPayout(order,w3)
                    except Exception :
                        logger.exception("An exception occurred while de-serializing message")
        except Exception :
            logger.exception("An exception occurred while reading message from kafka")
        finally:
            c.close()
    except Exception :
        logger.exception("An exception occurred while reading message from kafka")

В любом случае я могу проверить, существует ли соединение с базой данных, как только получено задание, и если нет, я могу восстановить соединение?

1 Ответ

0 голосов
/ 09 мая 2019

Согласно https://github.com/celery/django-celery-results/issues/58#issuecomment-418413369 и комментарии выше, поставив этот код: 

from django.db import close_old_connections
close_old_connections()

, который закрывает старое соединение и открывает новое внутри вашей задачи, должно помочь.

...