私は現在、いくつかのステップを含むワークフローに Redis を使用しています。各ステップで、ワーカーはペイロードをキューから取得し、完了すると次のキューにプッシュします。次のワーカーがさらに処理できるようになります。例外が発生した場合、タスクはワーカーによって特別なキューに入れられます。
したがって、アプリケーションを介したフローに関するアプリケーション ロジックは、ワーカー自体にあります。私は今、セロリに切り替えたいと思っています。
セロリではサブタスクを使用できることは理解していますが、例外やタイムアウトなどのさまざまな条件で特定のエラー処理を表現する方法がわかりません。別のキューを使用するか、サブタスクを使用する必要がありますか? コードではどのようになりますか?
ドキュメントをさらに徹底的に読み、さらにいくつかのテストを行ったところ、これは機能します。
問題は、タスクが次々に発生するようにタスクをつなぎ合わせることですが、同時にエラー状態を処理し、フローを「中断」して、単に中止するだけでなく、何か他のことを実行できるようにすることです。
タスクをlinkでつなぎ合わせることができ、追加のパラメーター *link_error* がそこにある場合、失敗に使用されます。読書から：
http://docs.celeryproject.org/en/latest/userguide/calling.html#linking-callbacks-errbacks 私はこれを作りました:
res = add.apply_async((2, 2), link=mul.s(16), link_error=onerror.s())
3 つのタスクは、add、mul、および onerror です。Add は 2 つの数値を加算し、mul は 2 つの数値を乗算します。したがって、これにより 2 と 2 が加算され、合計が次のステップ (mul) に引き継がれ、16 が乗算されます。 、add は例外をスローし、mul の代わりに onerror タスクが実行されます。onerror タスクは、ジョブの uuid を取得し、データベース バックエンドでジョブを検索できます (そのようなものが構成されている場合)。onerror タスクは、ジョブをアーカイブしたり、電子メールを送信したりできます。