0

Celery 2には、次のように設定されたTaskSetがありました。

for (item,jobId) in itemsAndJobs:
    tasks.append(waitForOutput.subtask((jobId,item)))
job = TaskSet(tasks)
result = job.apply_async()

その後、以下をチェックして、TaskSetが完了したかどうかを確認します。

job.ready() and job.successful()

これは正常に機能しました。waitForOutputタスクはゆっくりと進み、すべて完了するとジョブチェックが完了します。何回も問題なくチェックできました。

セロリ3では、変更するだけでこれをグループに変更するための迅速で汚い方法を試しました

TaskSet(tasks)

に

group(tasks)

すべてのwaitForOutputタスクが完了するまで待ってから、準備ができて成功したことを確認しない限り、これは機能しません。ready（）は常にfalseを返します。ロギングとデフォルトの30秒の再試行を追加しましたが、これが表示されます-

  • 5つのwaitForOutputジョブを開始します
  • ready（）を確認してください。waitForOutputジョブは完了していません。readyはfalseです。
  • 2つのwaitForOutputジョブが完了しました
  • check ready（）、readyはfalse、job.check_completed（）は2
  • check ready（）、readyはfalse、job.check_completed（）は0
  • 残りの3つのwaitForOutputジョブが完了しました
  • check ready（）、readyはfalse、job.check_completed（）は3

Celery 3コードを取得し、タスクからTaskSetをインポートして、グループの代わりにそれを使用した場合も、同じ動作が見られます。

グループを間違って使用しているだけだと言われたいです！

4

1 に答える 1

0

結果のバックエンドとしてRedisに移動することで、これを解決しました。結果のバックエンドとしてAMQPを使用する場合のバグのようです。

于 2012-08-23T15:20:11.993 に答える