他のすべてのタスクが終了した後にセロリのタスクを開始する必要があります

Question

セロリーが比較的新しいです。セロリのタスクの1つは、他のタスクがすべて終了した後に開始する必要があります。私はそれを使って遊びました。また、オンラインでいくつかのドキュメントを見つけました。私はgroupsとchordsなどを使って、完全に理解できないはずです。

私はまた、SO hereの別の質問を掲載しましたが、今までのところ納得のいく回答は見つかりませんでした。

最後に私はthis SO questionに出くわしました。私は、受け入れられた答えで何が起こっているのかを明確に（少なくとも概念的に）理解することができました。私は最後にまったく同じプログラムを複製しようとしたときしかし、それは以下のエラーを投げた：私は上記の質問に同様のコメントで同じを掲載している

EncodeError: <AsyncResult: cf5875f1-7f72-449c-9808-07c9c9459737> is not JSON serializable

、質問は非常にあると思われます私は今まで私のコメントに何の牽引力も受けていません。

誰かが助けてくれますか？

Answer 1

this SO answerで受け入れられた回答は、実際に私が欲しかったものを達成するのに役立ちました。 （おそらく、これはやっかいな方法でしたが、私の問題は解決しました）。その答えは、しかしとしてエラーを投げた：私は以下のように解決することができました

EncodeError: <AsyncResult: cf5875f1-7f72-449c-9808-07c9c9459737> is not JSON serializable

はユースケースでそう

（私はその答えのコメントでも、この変更を言及しています）答えにおける一部：我々はタスクオブジェクトのリストを渡しているamass.delay()の2番目のパラメータとして

tasks = [] 
for i in xrange(10): 
    tasks.append(power.delay(i, 2)) 

amass.delay([], tasks) 

。私はちょうど代わりに実際にタスクIDのリストを渡すためにこれを変更しました。だから、amass()は今のようになります。

tasks = [] 
for i in xrange(10): 
    x = power.delay(i, 2) 
    tasks.append(x.id) 

amass.delay([], tasks) 

と の下に表示し、それがエラーを解決し、私が達成しようとしていた仕事をしてくれたようamass()でそれぞれの変更を行いました。

@celery.task() 
def amass(results, tasks): 
    completed_tasks = [] 
    for task_id in tasks: 
     result = AsyncResult(task_id, app=celery) 
     if result.ready(): 
      completed_tasks.append(task_id) 
      #results.append(task.get()) did not need this so commented it out 

    # remove completed tasks 
    tasks = list(set(tasks) - set(completed_tasks)) 

    if len(tasks) > 0: 
     # resend the task to execute at least 1 second from now 
     amass.delay(results, tasks, countdown=1) 
    else: 
     # we done 
     print results 

Answer 2

場合、あなたはすでに次々に実行する必要があり、あなただけchordsを使用する必要があり、結果を結合したいタスクのリストを持っています。

他

あなたが（ないすぐ）セロリにタスクを追加し続けると、あなたがそれらを1つずつ実行したい場合は、単に同時実行= 1で1人の労働者を設定することで、あなたがこれを達成することができますが、タスクは逐次的にしか実行されません。

並列実行の問題もありましたが、最初に並行性= 1のワーカーを実行して解決しましたが、後で回避策を達成するためにredisロックメカニズムを使用しました。