タスクの結果をデータベースに保存するようにdjangoceleryをセットアップしようとしています。

私は設定しました：

次に、データベースを同期して移行しました（エラーなし）。

Celery は機能しており、タスクは処理されます (結果を取得できます) が、管理者はタスクがないことを示しています。データベースには と の 2 つのテーブルがcelery_taskmetaありdjcelery_taskmetaます。1 つ目は結果を保持し、2 つ目は管理画面に表示されます。誰でも適切に構成する方法を知っていますか?

質問

私はセロリを使用して、次のようなタスクセットを起動します。

1. 並行して実行できるタスクのバッチを実行します。このバッチのタスクの数は、数万から数千までさまざまです。

2. これらのタスクの結果を1つの回答に集約してから、この回答を使用して何かを実行します。たとえば、データベースに保存したり、特別な結果ファイルに保存したりします。基本的に、タスクの実行が完了した後、次のシグネチャを持つ関数を呼び出す必要があります。

   /li>

今のところ、ステップ1はセロリキューで実行され、ステップ2はセロリの外で実行されます。

すべてのタスクが実行されるまで（数時間かかる場合があります）単一のスレッドを停止する必要があるため、このアプローチは面倒です。

どういうわけかステップ2をセロリにも移動したいと思います---基本的に、タスクセット全体にコールバックを追加するか（セロリでサポートされていないことがわかっている限り）、これらすべてのサブタスクの後に実行されるタスクを送信する必要があります。

誰かがそれを行う方法を知っていますか？django環境で使用しているので、データベースに状態を保存できます。

私の最近の調査結果を要約すると

コードはしません

コードを使用すると、次のように見えるコールバックを作成できるため、コードを直接使用することはできません。

追加のパラメーターをコールバックに渡す明確な方法はありません（特に、これらのコールバックをローカル関数にすることはできないため）。

データベースを使用する

を使用して結果を保存できますTaskSetMetaが、このエンティティにはステータスフィールドがありません---したがって、TaskSetMetaにシグナルを追加する場合でも、かなりのオーバーヘッドが発生する可能性のあるタスク結果をプールする必要があります。

Redis をブローカーとして Celery を使用していますが、キューは実際にはシリアル化されたタスクをアイテムとして含む redis リストであることがわかります。

私の質問は、呼び出しの結果として AsyncResult オブジェクトがある場合<task>.delay()、キュー内のアイテムの位置を特定する方法はありますか?

アップデート：

私は最終的に以下を使用して位置を取得することができます:

ただし、すべてのワーカーと通信する必要があるため、少し遅いです。

私はRabbitMQバックエンドでCeleryを使用しています。保留中のタスクの数を監視して取得するにはどうすればよいですか? 「celery events」と djcelery は、実行中のタスクと完了したタスクを示します。RabbitMQ を監視する必要がありますか? もしそうなら、どのように？

その中に3つのメソッドが定義されたtasks.pyファイルがあります。私のviews.pyファイルでは、ビューから呼び出されるこれらのメソッドをインポートしようとしています。

これらのタスクを呼び出すビューを呼び出すと、django は次のように戻ります

私のプロジェクトの構造は次のとおりです

これらのメソッドのインポートに問題がある理由はわかりません。誰かが私が間違っていることについての洞察を提供できれば、それは大歓迎です。

ありがとうございます！

私のdjangoプロジェクトでセロリをセットアップしようとしているときに、これら2つのアプリケーションの違いがわかりません。

もしあれば、2つの違いは何ですか？オンラインでチュートリアルを読むと、両方が使用されていることがわかりますが、どちらが自分に最適かわかりません。djcelery はセロリに似ているようですが、django 用に調整されていますか? ただし、djcelery は必要ですが、インストールされたアプリに celery を含める必要はありません。

ありがとうございました

非同期プロセスに対して manage.py celeryd を実行すると、すべてが期待どおりに機能しますが、起動するたびに奇妙な警告が表示されます。エラーは発生していませんが、消えたり、意味を理解するのに苦労しています。

ここにあります：

django 設定モジュールが私の wsgi に追加されているため、これは珍しいことであり、これ以外のすべてで機能します。設定を httpd.conf などに追加する必要がありますか?

ありがとう

同じマシン上の 2 つのシェルで、次のことを行います。

a) /home/user/prod/env/bin/python /home/user/prod/www/devcode/manage.py セロリワーカー -E

どうやら問題なくセロリが起動します

b) /home/user/prod/env/bin/activate && /home/user/prod/www/devcode/manage.py セロリの状態

生成:

エラー: 制限時間内にノードが応答しませんでした。

どうやら根本的なことを見逃していたようですが、2日間試した後、質問する必要があります:(

私はdjangoでセロリを使用しています。失敗したタスクを調べ、必要に応じて失敗したタスクのデータを変更し、再度送信するオプションをユーザーに提供する必要があります。私はこのスレッドを見てきました - Celery Storeing unrecoverable task failures for later resubmission。したがって、セロリはタスクの元の引数とkwargsを保存しないことを理解しており、それを処理する必要があります。私はそれで大丈夫です。しかし、チェーン「SubTask1 | SubTask2 | SubTask3」を送信するメインタスク「MainTask1」があり、SubTask2 が失敗した場合、SubTask2 が成功するまで SubTask3 が実行されないことがわかります。ただし、SubTask2 が最大再試行後に失敗した場合、SubTask3 は送信されません。

私の質問は -

1. SubTask2 が失敗した場合、その引数と kwargs を永続化できます。しかし、チェーン内の残りのタスクの情報を取得するにはどうすればよいでしょうか?

2. celery_taskmeta テーブルの「result」列と「meta」列には正確に何が格納されていますか?

3. テーブル celery_tasksetmeta はいつ入力されますか?

ありがとう、

私はこの6時間かそこらの間これに固執しています。ですので、どんな助けでも大歓迎です。

HerokuにデプロイされているDjangoアプリでdjceleryを使用しています。このアプリには、問題なく機能する定期的なタスクがあります。

今日は、前のアプリのクローンである別のHerokuアプリを立ち上げたので、本番環境を用意します。新しいHerokuデプロイメントは定期的なタスクを実行しませんが、代わりに、ログにこれを取得します。

これは私が立ち往生しているところです。2つのデプロイメントはまったく同じコード、同じHerokuアドオン、同じ構成を持っています。さらに、djceleryのDBモデルは両方のデプロイメントで空です。

私は何が欠けていますか？上記以外の要因はありますか？それが問題にならないように、定期的なタスクに明示的な名前を付けました。

どんな助けでも大歓迎です！