問題タブ [celerybeat]

ユーザーが登録してからちょうど 48 時間後にメールを送信したいのですが、セロリを使用してこれを達成するにはどうすればよいですか? メールを送信する定期的なタスクを作成する場合、そのタスクを実行する特定の時間を決定する必要があります。送信する必要のある電子メールがあるかどうかを確認するために、毎秒セロリ タスクを実行し続けたくありません。

セロリワーカーとrabbitmqをブローカーとして備えた3台のマシンがあり、1つのワーカーがビートフラグで実行されており、これらすべてがスーパーバイザーによって管理されており、セロリがこのようなエラーで死ぬことがあります。このエラーはビート ワーカーでのみ表示されますが、表示されると、すべてのマシンのワーカーが停止します。(セロリ==3.1.12、昆布==3.0.20)

次のように、1分に1回実行される定期的なセロリタスクがあります。

関数pollAPI()は、名前から推測できるように、API をポーリングします。問題は、API には公開されていないレート制限があり、その制限に達すると、エラー応答が返されることがあることです。その応答を取得できるようにしたいと思います。制限に達した場合は、定期的なタスクの間隔を動的に減らします (または、タスクをしばらく一時停止します)。これは可能ですか？

is_dueのメソッドの上書きに関するドキュメントを読みましたがschedules、ここで探している動作を実現するために何をすべきか正確にわかりません。誰でも助けてもらえますか？

したがって、私は VM (vagrant) を使用して開発しています。その中でセロリ ビートを開始すると、次のメッセージが表示されます。

その中でデートするTue Jul 15 09:25:11 UTC 2014と、私はポルトガルに住んでいて、ホストマシンが私にTer Jul 15 10:25:39 WEST 2014.

これを修正するための最良の方法は何ですか?

これをライブにしたらどうですか？

使用celery 3.1.12していますが、セットはありませんCELERY_TIME_ZONE。

次を使用して、新しいスケジュールされたタスクを Celery に追加しています。

そしてupdate_scheduled_tasks、新しいタスクが追加されたときにタスクを呼び出します。

これは、1 つの Celery ワーカーで問題なく動作します。新しいスケジュールされたタスクが追加されます。複数のワーカーを使用しても同じように機能しますか? Celery は構成の変更を複数のワーカーにどのように伝達しますか? これに関するドキュメントには何も見つかりませんでした。

PS私はFlaskを使用しているのでdjango-celery、私には選択肢がありません.

私はDjangoアプリでCeleryを使用していますが、ほとんどの場合、物事は順調に進んでいます。参考までに、私のセロリビート構成は次のとおりです。

1 ～ 2 日ごとにランダムに完全に同期しなくなることに気付きました。たとえば、5 分ごとに実行される最後のエントリは、通常、1:00、1:05、1:10、1:15 などのように実行されます。

これが今朝の動作です: 7:30, 7:35, 7:40, 7:46 , 7:50, 8:03, 8:16, 8:32, 8:50, 9:13, 9: 21, 9:26, 9:29, 9:32, 9:38, 9:47, 9:59, 10:08, 10:22, 10:24, 10:24, 10:24, 10:24, 10:25, 10:25, 10:25, 10:25, 10:25, 10:25, 10:25, 10:25, 10:25, 10:25, 10:25, 10:26, 10: 26、10 :30、10:35、10:40...

同期が外れて、一度にすべてを埋めようとして、それから自分自身を取り戻しているように見えます。最初はただのまぐれだと思っていましたが、あまりにも頻繁に発生しています。これには、時間通りに実行する必要があるミッション クリティカルなプロセスがあります。

これを防ぐ方法はありますか？人々がセロリで多くの成功を収めているように見えるので、私は本当に困惑しています.

セロリビートに、メッセージに渡す引数を返す呼び出し可能オブジェクトを与える方法はありますか? データベースからデータを取得して、セロリ キューに渡す引数を作成できるようにしたいと考えています。

私がやりたいことは次のようなものです：

現在、キューにメッセージを追加するマシンと、メッセージをキューから取り出して処理する別のマシンがあります。メッセージをキューに追加しているマシンは、その引数をハードコーディングしていますが、その情報をデータベースから引き出したいと思っています。

現在考えられる唯一の解決策は、キューにメッセージを追加しているマシンがセロリ自体を実行することです。処理サーバーの元のキューにメッセージを追加するコマンドを実行するための別のキューがあります。これは非常に複雑に思えますが、もっと簡単な解決策があるはずです。

私はウサギのmqでセロリとセロリビートを使用していますが、すべて正常に動作します。

しかし今、ブロードキャスト キューに定期的なタスクを作成しましたが、これは機能しません。

セロリconfig.py

task.py

ログファイルで、タスクがトリガーされていることがわかりますが、タスクは実行されていません

ブロードキャスト キューで定期スクリプトを開始するにはどうすればよいですか?

ありがとう

マルセル

を介して一度実行されるタスクcalculate_common_locationsCELERYBEAT_SCHEDULEがあります。このタスクは、データベース内の関数を呼び出すだけです。

これは のエントリですCELERYBEAT_SCHEDULE:

スケジュールには、1 日 1 回または10 秒ごとに実行されるタスクがさらに含まれています。これらのタスクは何度も再実行されないようです。セロリの花は、タスクが 20 回以上実行されたことを示しています。スケジュールどおりに開始された最初のものは、約 100 秒実行され、成功してから再び開始されます。

実行中の celerybeat は 1 つだけです。

これは、セロリが開始される方法です（supervisord経由）：

--concurrency=10スイッチなしでテストしました。データベース機能はまだ複数回実行されています。

この関数は、非常に頻繁に (1 秒間に数回) 挿入される大きなテーブル (> 100 万行) から読み取ります。Postgres ロックは、すべてのロックが許可されていることを示しています。

クエリがある時点で終了するため、タスクが再実行されている可能性はありますか?

次の場合は問題ありません。

• タスクは django シェルから (直接または経由で.delay())実行されます。
• タスクの内容は、軽量の SQL クエリ (select * from test) に置き換えられます。
• タスクの内容はsleep(100)に置き換えられます。

バージョン:

• セロリ==3.1.12
• psql (PostgreSQL) 9.3.5

私のアプリケーションには 3 つのビート タスクがあります。ビート タスクが実行されるとflower、タスクを管理するために使用します。

1 つのタスクにバグがあることがわかった場合は、タスク全体を停止してバグを修正する必要があります。次に、セロリを再起動します。

1 つのタスクを停止して、他のタスクを実行し続けることはできますか?