いくつかの注文を処理するためにKafkaを使用しているdjangoベースのWebアプリケーションがあります。現在、Celery Workersを使用して、Kafkaコンシューマーを各トピックに割り当てています。各Kafkaコンシューマーは、Kafkaタスクの形式でKafkaトピックに割り当てられます。ただし、1日ほど経った後、タスクを送信すると、次のエラーが発生します。
_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")
タスクを受信するとすぐにデータベース接続が存在するかどうかを確認でき、存在しない場合は接続を再確立できますか?
https://github.com/celery/django-celery-results/issues/58#issuecomment-418413369と上記のコメントによると、このコードを入力します。
from django.db import close_old_connections
close_old_connections()
古い接続を閉じて、タスク内で新しい接続を開くと役立つはずです。
この記事はインターネットから収集されたものであり、転載の際にはソースを示してください。
侵害の場合は、連絡してください[email protected]
コメントを追加