セロリのワーカーごとに別々のデータベース接続を作成する

Question

作成直後に作業を実行しているワーカーの間に、wierd mysqlの問題に取り組んでいます。

我々はこれまでのところで、時に労働者の処理開始、それらはすべて同じMySQLを取得する同時実行3. マイobeservationとセロリプロセスを開始ジャンゴ1.3、セロリ3.1.17、djorm-EXT-プール0.5

を使用

接続。以下のようにdb接続IDをログに記録します。

from django.db import connection 
connection.cursor() 
logger.info("Task %s processing with db connection %s", str(task_id), str(connection.connection.thread_id())) 

すべてのワーカーがタスクを取得すると、最初のタスクは正常に実行されますが、残りの2つは奇妙なMysqlエラーが発生します。それは、 "MySQLのサーバーがなくなった"エラー、またはDjangoが "DoesNotExist"エラーをスローする状態のいずれかです。 Djangoが照会しているオブジェクトは明らかに存在します。

このエラーが発生すると、各ワーカーは独自のデータベース接続を開始したが、問題が見つからない。

セロリのデフォルトの動作は何ですか？同じデータベース接続を共有するように設計されていますか？もしそうなら、プロセス間通信はどのように扱われますか？ 私は理想的には、各作業者ごとに異なるデータベース接続を推奨します。

下のリンクで動作しないコードを試しましたが、 Celery Worker Database Connection Pooling

また、以下で提案するセロリコードを修正しました。 https://github.com/celery/celery/issues/2453

質問を下す人のために、私にdownvoteの理由を教えてください。

Answer 1

セロリは、ワーカープロセスをforkする前に、MySQLデータベースにいくつかのクエリを作っていたマスター・プロセスの一環として、以下のコマンド

celery -A myproject worker --loglevel=debug --concurrency=3 -Q testqueue 

myproject.pyで開始されます。

メインプロセスのクエリフローの一部として、django ORMは、sqlalchemy接続プールが存在しない場合は作成します。その後、ワーカープロセスが作成されます。

セガはdjango fixupsの一部として既存の接続を閉じます。実際に

def close_database(self, **kwargs): 
    if self._close_old_connections: 
     return self._close_old_connections() # Django 1.6 
    if not self.db_reuse_max: 
     return self._close_database() 
    if self._db_recycles >= self.db_reuse_max * 2: 
     self._db_recycles = 0 
     self._close_database() 
    self._db_recycles += 1 

起こっことができるもの二股ときに未使用のDB接続とSQLAlchemyのプールオブジェクトが3ワーカープロセスにコピーされる、ということです。したがって、3つの異なるプールには、同じ接続ファイル記述子を指す3つの接続オブジェクトがあります。

db接続を要求されたときにタスクを実行しているワーカーは、現在使用されていないため、すべてのワーカーがsqlalchemyプールと同じ未使用接続を取得します。すべての接続が同じファイル記述子を指しているという事実は、MySQL接続がエラーを起こした原因となっています。

新しい接続はすべて新しいもので、同じソケットファイル記述子を指しません。

ソリューション：任意のインポートが行われる前にメインプロセスで

は

from django.db import connection 
connection.cursor() 

を追加します。即ち、djorm-ext-poolモジュールが追加される前である。

このように、すべてのdbクエリは、プール外でdjangoによって作成された接続を使用します。セロリdjango fixupが接続を閉じると、実際に接続が閉じられ、錬金術プールに戻って、錬金術プールに接続がなくなり、フォークされたすべての作業者に対処します。その後、作業者がdb接続を要求すると、sqlalchemyは新しく作成された接続の1つを返します。