問題タブ [celery]

誤ってキューに追加したものが多すぎて、フリーズしました。それ以外のことをすることはできません。

Djangoの特定のアプリに対してのみDEBUG=Falseを設定することは可能ですか？デバッグが有効になっている場合、Celeryには既知のメモリリークがあります。Celeryをデバッグせずにサービスとして実行してメモリリークを発生させない開発サーバーがありますが、Djangoアプリの残りの部分でデバッグを使用して、テスト時にエラーが表示されるようにします。

私はDjangoとCelery+RabbitMQを使用して、ユーザーがアップロードしているビデオのビデオ変換タスクを作成しています。これで、セロリにクエリを実行してステータスを取得する方法がわかりました。私の質問は、各タスクに関連付けられたtask_idをどこに保存するかですが、モデルまたはdjangoのキャッシュに保存する必要がありますか？

過去に同様の質問があったことは知っています。たとえば、このスタックオーバーフローの質問ですが、決定的な答えはまだありません。これに対する答えは開発者の好みによって異なることは知っていますが、誰かがさまざまなアプローチの長所/短所について教育できれば、それは素晴らしいことです。

繰り返しになりますが、task_idを使用して、フロントエンドからAJAXクエリを頻繁に実行し、ビデオ変換が完了したかどうかを確認します。

Celery を使用して、複数のデータ マイニング タスクを処理しています。これらのタスクの 1 つは、ユーザーごとに最大 10 の同時接続を許可するリモート サービスに接続します(つまり、グローバルに 10 接続を超えることはできますが、個々のジョブごとに 10 接続を超えることはできません)。

Token Bucket (レート制限)を探していると思いますが、その実装が見つからないようです。

タスクの再試行で問題が発生しました。テストタスクは次のようになります。

装飾されたタスクを再試行する方法に関するドキュメントがまったく見つかりません。見つけたのはこれだけです。

selfメソッドから渡される変数がないため、これは私の場合には機能しないようです。

編集：

私は今、無駄に次のことを試みています：

次のエラーが発生します。

TypeError（"再試行のkwargs引数を空にすることはできません。タスクは**kwargsを受け入れる必要があります。http：//bit.ly/cAx3Bgを参照してください"、）

私は、10,000フィートから以下で構成される分散マスターワーカーシステムを設計しています。

• WebベースのUI
• 構成可能なアルゴリズムのセットに従ってジョブを生成する役割を担うマスターコンポーネント
• 通常のPC、HPCクラスター、さらにはクラウドで実行されている一連のワーカー
• デジタルリポジトリ
• メッセージングベースのミドルウェア
• さまざまなカテゴリのタスク。実行時間は1秒未満から6時間までの範囲です。タスクは、データ/ IOに負担をかけるのではなく、計算に負担をかけます。タスクの量はそれほど多くないと予想されます（私が今見る限り）。おそらく最大で約100/分です。

厳密に言えば、Windowsエコシステムの外に移動する必要はありませんが、オプションを開いたままにするためのクロスプラットフォームソリューションの方が快適です（一部のタスクはWindowsのみです）。

私はメッセージングレイヤーとしてRabbitMQをほぼ採用しており、 Fedora-commonsは最も成熟した既製のリポジトリのようです。私が評価しているマスター/ワーカーロジックについては、次のとおりです。

• Javaベース：Grails + Postgres+ DOSGiまたはGridGainと Zookeeper
• Pythonベース：Django + Postgres + Celery
• .netベース：ASP.NET MVC + SQL Server + NServiceBus +リポジトリとしてのSharepointまたはZentity

私はさまざまなIoC/DIコンテナーを調べましたが、それらがタスク実行コンテナーに本当に最適であり、レイヤー/複雑さを追加するのに本当に最適であるとは思えません。しかし、多分私は間違っています。

現在、私はpythonソリューション（軽量に保つ）に傾倒していますが、特に.netスタックで人々が共有しなければならない経験/提案に興味があります。オープンソース/スケーラビリティ/復元力の機能はプラスのポイントです。

PS：より高度な将来の要件は、ユーザーが実行中のタスクに直接接続し（Web UIを使用）、その動作に影響を与える（リアルタイムステアリング）機能です。これを行うには、直接通信チャネルが必要になります（AMQPを介してこれを行うのは良い考えではないようです）。

celerydをデーモンとして実行していますが、正常に停止できない場合があります。TERMシグナルを送信し、キューにアイテムがある場合（この場合service celeryd stop）、celerydは新しいジョブの取得を停止し、すべてのワーカープロセスをシャットダウンします。ただし、親プロセスはシャットダウンしません。

AとBの2つの別々のワーカーマシンでcelerydを実行しているシナリオに遭遇しました。RabbitMQサーバーに約1000のメッセージがあり、Aをシャットダウンして、上記で説明した状況を経験しました。Bは引き続き機能しましたが、サーバーに約40のメッセージが残ったまま停止しました。しかし、Bを正しく止めることができました。

Bを再起動して、40個のアイテムがキューから削除されるかどうかを確認しましたが、そうではありませんでした。次に、私はAを激しく殺し、その後Bはタスクをつかんで完了しました。

私の結論は、親プロセスがその子のためにRabbitMQサーバーから40個のアイテムを予約したということです。子を正しく刈り取りますが、手動で強制終了しない限り、アイテムをRabbitMQに解放しません。

誰かが似たようなことを経験したことがありますか？

Celery2.2.2を実行しています

Celery を介して Django で RabbitMQ を使用しています。私は最も基本的なセットアップを使用しています：

Celery タスクをインポートし、1 年後に実行するようにキューに入れました。iPython シェルから:

RabbitMQ は、セロリ キューで次のメッセージを受け取りました。

次に、control-C を押してから「a」を押して中止し、RabbitMQ を強制終了しました。サーバーを再起動し、rabbitmqctl で確認すると、セロリ キューにメッセージがないと表示されます。

セロリ キューは耐久性がありました。メッセージが保持されなかったのはなぜですか? メッセージを永続化するにはどうすればよいですか?

私には次のようなタスクがあります：

次に、これを実行します。

タスクが完了したことがワーカーで確認できます。

[2011-02-19 20：18：43,851：INFO /MainProcess]タスクtestcelery.test[7598b170-2877-4d76-89a0-9bcc4c9f877e]は20.0225799084sで成功しました：「完了」

ただし、t.stateがPROGRESSからSUCCESSに変わることはありません。私は何が間違っているのですか？

多くのタスクを開始しましたが、完了していないタスク (763 タスク) があり、PENDING 状態ですが、システムは何も処理していません... セロリに task_id を与えてこのタスクを再試行することは可能ですか?