問題タブ [celery]

私はセロリを使用して、ニュースアグリゲーションサイトのRSSフィードを更新しています。フィードごとに1つの@taskを使用しますが、うまく機能しているようです。

ただし、うまく処理できるかどうかわからない詳細があります。すべてのフィードは@periodic_taskで1分ごとに更新されますが、新しいタスクが開始されたときにフィードが最後の定期的なタスクからまだ更新されている場合はどうなりますか？（たとえば、フィードが非常に遅い場合、またはオフラインで、タスクが再試行ループで保持されている場合）

現在、タスクの結果を保存し、次のようにステータスを確認しています。

たぶん、私が見逃したセロリのメカニズムを使用して同じ結果を達成するためのより洗練された/堅牢な方法がありますか？

何らかの理由で、Celery で新しいタスクを作成して実行すると、結果を返す際に問題が発生します。最初のタスクは完全に返されますが、後続のすべてのタスクの結果は常に保留中です。Celery ログを確認したところ、エラーなしで正しい結果が得られましたが、それを返すことができません。

それが役立つ場合は、バックエンドとして rabbitmq を実行しています。

一度に 1 つのタスクを実行する単純なキューを実装しようとしています。Celery を使用してメイン スレッドからタスクをオフロードし、Celery 構成で concurrency=1 を設定すると問題なく動作しますが、他のタスクにはより多くの同時ワーカーを使用したい場合があります。

Celery または RabbitMQ に、タスクに複数の同時ワーカーを使用しないように指示する方法はありますか (concurrency=1 を強制する場合を除く)? ドキュメントには何も見つかりませんが、これらのツールは線形キュー用に設計されていないのでしょうか?

ありがとう！

自社開発のタスクキューシステムをCeleryベースのタスクキューに変換したいのですが、現在持っている機能の1つが苦痛を引き起こしています。

現在、私のタスクキューは非常に粗く動作しています。ジョブ（データを生成して別のサーバーにアップロードする）を実行し、Noseのログキャプチャライブラリのバリアントを使用してログを収集してから、タスクのログを詳細な結果レコードとしてアプリケーションデータベースに保存します。

これを3つのタスクに分けたいと思います。

1. データを収集します
2. データをアップロードする
3. 結果のレポート（前の2つのタスクからのすべてのログを含む）

ここでの本当のキッカーは、ロギングコレクションです。現在、ログキャプチャを使用して、データの生成およびアップロードプロセス中に行われたログ呼び出しごとに一連のログレコードがあります。これらは診断目的で必要です。タスクが同じプロセスで実行されることが保証されていないことを考えると、Celeryタスクキューでこれをどのように実行するかは明確ではありません。

この問題に対する私の理想的な解決策は、先行タスク（1、2）中にすべてのログをキャプチャし、レポータータスク（3）で利用できるようにする、些細で理想的には低侵襲の方法です。

タスク定義をかなり粗くして、このすべての作業を1つのタスクにまとめるのが最善ですか？または、最後に収集するために、既存のキャプチャされたログを渡す方法はありますか？

FreeBSD の celeryd 用の rc スクリプトを一緒にハックしましたが、もっと良い方法があるに違いないと思わずにはいられません。 celerydは自分自身をデーモン化しませんし、sigterm への対応にも苦労しているようですので、作業に取りかかるのは難しいかもしれません。

これは、他の誰かが以前に解決した問題ですか?

データを Redis に保存するオブジェクトがあります。ブロックの必要性をできるだけ少なくする必要があるため、Celery を使用してタスクをオフロードすることにしました。セロリの外でオブジェクトを .save() しようとすると、Redis に接続され、データが正常に保存されます。ただし、Celery タスクからまったく同じことを実行しようとすると、実行されているように見えますが、Redis への接続、例外、エラー出力はなく、Redis サーバーには何も保存されません。以下の小さなコードで問題を再現しました。test.py:

Python コンソール出力は次のとおりです。

そして、ここにセロリの出力があります：

どんな助けでも素晴らしいでしょう！複数のpythonバージョンを使用して、複数のコンピューターで問題を再現しました。

2 つのセロリ ワーカーがあり、ワーカー 1 にはタスク A と B、ワーカー 2 にはタスク A、B、および C があります。タスク C を送信すると、タスク C を持つセロリ ワーカーでは実行されないようです。ワーカー 2 だけがタスク C を割り当てられるようにする方法はありますか?

ほぼ同期も機能します。基本的に、ほとんどのジョブのタスクキューにWebアプリの背後でデータアクセスと処理を委任したいと思います。セロリのタスクに合理的と見なすことができる最速のレイテンシーはどれくらいですか？

更新（明確化のため）

わかりやすくするために、スループットは（良いのですが）私にとって必要な問題ではないことを説明する必要があると思います。しばらくの間、その方向にスケーリングする必要はありません。レイテンシーは、現時点で評価する唯一の基準です。それが機能する唯一の方法である場合、私は使用することに満足してtask.applyいますが、少し作業を進めたいと思います。

結果が大きいCeleryで実行されたタスクを処理する最良の方法は何ですか? 数百メガバイトのデータを返すテーブル ダンプなどを考えています。

メッセージを結果データベースに詰め込むという素朴なアプローチは、ここでは役立たず、結果のバックエンドに AMQP を使用する場合はなおさらだと思います。ただし、遅延が問題になるこれらのいくつかがあります。エクスポートの特定のインスタンスによっては、返されるまでブロックして、タスク クライアントからエクスポート データを直接送信する必要がある場合があります (エクスポート コンテンツに対する HTTP 要求が着信しましたが、存在しませんが、その要求への応答 ... 時間がかかっても)

では、このためのタスクを作成する最良の方法は何ですか?

私はdjangoとceleryで次のことをしようとしています。これが私がしていることの簡略版です

エラーは返されませんが、pdf_fileはnullのままです。何か案は？

ありがとう！