問題タブ [celery-task]

次のように、チェーン内でグループ（またはチャンク）を使用したいと思います。

タスクGROUP()のグループはどこにありますか。同様の質問が、リストをグループに返すCeleryタスクをチェーンする方法に投稿されました。ただし、グループからの出力をチェーン内の次のタスクに渡す方法については説明されていません。double()group(double(0),double(1),double(2))

セロリでタスクをデーモン化しようとしています。デーモン化せずにテストしましたが、非常にうまく機能しています。

しかし、チュートリアルのようにデーモン化することはできません ( http://docs.celeryproject.org/en/latest/tutorials/daemonizing.html#daemonizing ) 私は自分のファイルを持っています:

solr_desa.py

セロリconfig.py

また、 /etc/default/celeryd

https://github.com/celery/celery/blob/3.0/extra/generic-init.d/celerydのデフォルトの celeryd で実行しますが、タスクはキューに入れられただけですが、ワーカーがないように見えます:(

構成のどこに間違いがありますか? :(

Celery 3.0.12 を使用しています。

Q1、Q2 の 2 つのキューがあります。

通常、メイン タスクを Q1 に置き、サブタスクを呼び出して Q2 に移動します。サブタスクの結果を保存したくありません。したがって、私のサブタスクにはデコレータ @celery.task(ignore_results=True) があります。

メインタスクは、サブタスクが完了するまで待機する必要があります。結果を書かないので。使用できません: AsyncResult. 状態をバックエンドに保存せずにサブタスクが終了するまで待機するメインタスクで待機する方法はありますか? AsyncResults を使用したすべての試みは、成功の燃料ではありません (バックエンドに依存しています)。get() もバックエンドに依存しているようです。

コード内のストーリー全体:

Celery Flower を使用してアプリケーション全体を監視していますが、サブタスクが正常に終了していることがわかります。セロリはどのようにしてその状態を検出できますか? 私は彼らのコードを閲覧しましたが、彼らがどのように検出を行っているかを見つけることができませんでした.

mytask.apply_async(countdown=120):(デコレータを使用mytaskした単純なPythonメソッドです）を使用してCeleryタスクを実行するようにスケジュールしようとしています。@taskただし、それを呼び出すと、タスクは2分後に実行されるのではなく、すぐに実行され、のような結果が返されます<EagerResult: 2b41cba1-81ff-489c-81bb-f1ad994d4441>。私がここで何を見逃しているのか分かりますか？etaまた、の代わりにを使用してこれを実行しようとしましたcountdown。私のセロリの構成は以下のとおりです。

-- v3.0.12 (Chiastic Slide)
[Configuration]
-- broker: redis://localhost:6379/0
-- app: default:0x106090f50 (djcelery.loaders.DjangoLoader)
-- concurrency: 4 (processes)
-- events: OFF (enable -E to monitor this worker)
[Queues]
-- celery: exchange:celery(direct) binding:celery
それが役に立ったら、私はこの方法でセロリを始めています： python manage.py celeryd -l DEBUG -Bそして定期的なcrontasksが正しく機能するようにしています。

ありがとう

まず第一に、この質問を この質問の重複と見なさないでください

とをceleryとredisとして使用する環境をセットアップしました。私の質問は、セロリ ワーカーがクラッシュしたときに、セロリ ワーカーがバックアップされたときに、スケジュールされたすべてのタスクが再試行されるようにするにはどうすればよいかということです。 brokerresult_backend

を使用することに関するアドバイスを見たことがありますCELERY_ACKS_LATE = True。これにより、ブローカーは ACK を取得するまでタスクを再実行しますが、私の場合は機能しません。タスクをスケジュールするたびに、そのタスクはすぐにワーカーに送られ、スケジュールされた実行時間まで保持されます。いくつか例を挙げましょう：

次のようなタスクをスケジュールしています:res=test_task.apply_async(countdown=600)が、すぐにセロリ ワーカー ログに : のようなものが表示されますGot task from broker: test_task[a137c44e-b08e-4569-8677-f84070873fc0] eta:[2013-01-...]。セロリ ワーカーを強制終了すると、これらのスケジュールされたタスクが失われます。私の設定：

火曜日の午後 12 時から水曜日の午後 12 時まで 1 時間ごとに実行する定期的なcronタスクを作成しようとしています。celeryこれは 24 時間の期間ですが、2 つの異なる日にまたがっています。これを次のような単一のタスクとしてスケジュールすることは可能です @periodic_task(run_every=crontab(<an expression equivalent to stated above>))
か ?
crontab(minute='0',hour='12-23',day_of_week='tue')
crontab(minute='0',hour='0-11',day_of_week='wed')

ありがとう

いくつかの外部ライブラリを利用する Celery タスクを実装しました。

タスクコードが次のようなものだとしましょう:

私がインポートする2つのモジュールは基本的に次のとおりです。

と：

my_job_id実行されたタスクに応じて、各行にの正しい値を含むログ ファイルを取得したいと考えています。

次のようなものになります。

my_job_idの値をタスク ロガーに簡単に含めることができ、タスク関数内からログに記録した行 ([1][...other info..] My task executed with my_job_id=1や など[2][...other info..] My task executed with my_job_id=2) を正しく取得できます。

しかし、外部ライブラリはどうですか? 外部スクリプトによって生成されたログを「ラップ」し、要件に従ってフォーマットするエレガントな方法はありますか?

今日、Amazon SQS の毎月の請求書を受け取りました。キューへの 600.000 リクエストの使用を見て驚きました。

私がしているのは、毎分 1 つのタスクを実行することだけです。これはどのように合計 600.000 リクエストになりますか?

私はまだセロリに慣れていないので、根本的に間違ったことをしている可能性があります。ヒントはありますか？

特定のタスクを実行するdjangoセロリビューがあり、タスクが正常に完了した後、それをデータベースに書き込みます。

私はこれをやっています：

と、

しかし、タスクの実行が終了した後、if条件にはなりません。試してみましresult.ready()たが、うまくいきませんでした。

編集：上記の行は同じビューにあります：

e：＃データベースへの書き込み

celeryd-multi のドキュメントには、次の例があります。

(ここから: http://docs.celeryproject.org/en/latest/reference/celery.bin.celeryd_multi.html#examples )

上記の例のように、1 つのホスト上で複数のワーカーが同じキューを処理するのが良い理由の実用的な例は何ですか? それが同時実行の設定ではないでしょうか。

より具体的には、次の 2 行 (A と B) の間に実際的な違いはありますか?

A:

B:

この実際的な違いを理解していないために、タスク キューに関する貴重な知識が欠けているのではないかと心配しています。

ありがとう！