11 
import celery
def temptask(n):
    header=list(tempsubtask.si(i) for i in range(n))
    callback=templink.si('printed at last?')
    r = celery.chord(celery.group(header))(callback)
    return r

@task()
def tempsubtask(i):
    print i    
    for x in range(i):
        time.sleep(2)
        current_task.update_state(
            state='PROGRESS', meta={'completed': x, 'total': i })

@task()
def templink(x):
    print 'this should be run at last %s'%x

#executing temptask
r = temptask(100)

tempsubtaskによって更新された進行状況にアクセスしたい。どうすればそれを達成できますか？

4

3 に答える 3

5

同様の質問がありました。ネット上のほとんどの例は時代遅れであり、ドキュメントはあまり役に立ちませんでしたが、ドキュメントにはソースへのリンクがあり、それを読んで私は役に立ちました。私の目的は、並列タスクをグループに編成することでした。グループは順番に実行する必要があります。そこで、タスクを個別に開始して割り当てる前に、タスクIDを生成することにしました。Celery4.3.0を使用しています

これが簡単な例です。

まず、実行をシーケンシャルにし、特定のグループの状態をチェックできるようにするためのダミータスクが必要でした。これはコールバックとして使用されるため、グループ内の他のすべてのタスクの後でのみ完了します。

@celery.task(bind=True, name="app.tasks.dummy_task")
def dummy_task( self, results=None, *args, **kwargs ):
    return results

ここでの私のコメントは、IDの割り当て方法を説明しています。

from celery.utils import uuid
from celery import group, chord, chain


# Generating task ids, 
# which can be saved to a db, sent to the client and so on
#
# This is done before executing any tasks

task_id_1 = uuid()
task_id_2 = uuid()

chord_callback_id_1 = uuid()
chord_callback_id_2 = uuid()

workflow_id = None


# Generating goups, using signatures
# the group may contain any number of tasks
group_1 = group(
        [
            celery.signature(
                    'app.tasks.real_task', 
                    args=(), 
                    kwargs = { 'email': some_email, 'data':some_data },
                    options = ( {'task_id': task_id_1 } )
                )
        ]
    )

group_2 = group(
        [
            celery.signature(
                    'app.tasks.real_task', 
                    args=(), 
                    kwargs = { 'email': some_email, 'data':some_data },
                    options = ( {'task_id': task_id_2 } )
                )
        ]
    )

# Creating callback task which will simply rely the result
# Using the task id, which has been generated before
# 
# The dummy task start after all tasks in this group are completed
# This way we know that the group is completed

chord_callback = celery.signature( 
        'app.tasks.dummy_task',
        options=( {'task_id': chord_callback_id_1 } )
    ) 

chord_callback_2 = celery.signature( 
        'app.tasks.dummy_task',
        options=( {'task_id': chord_callback_id_2 } )
    ) 


# we can monitor each step status
# by its chord callback id

# the id of the chord callback  
step1 = chord( group_1, body=chord_callback )
# the id of the chord callback  
step2 = chord( group_2, body=chord_callback_2 )

# start the workflow execution
# the steps will execute sequentially 
workflow = chain( step1, step2 )()


# the id of the last cord callback
workflow_id = workflow.id

# return any ids you need
print( workflow_id )

これにより、アプリ内のタスクのステータスを確認できます。

# This is a simplified example
# some code is omitted
from celery.result import AsyncResult


def task_status( task_id=None ):

    # PENDING
    # RECEIVED
    # STARTED
    # SUCCESS
    # FAILURE
    # REVOKED
    # RETRY

    task = AsyncResult(task_id)

    response = {
      'state': task.state,
    }

    return jsonify(response), 200
于 2019-04-24T08:28:28.280 に答える
4

何時間もグーグルした後、私はhttp://www.manasupo.com/2012/03/chord-progress-in-celery.htmlに出くわしました。そこにある解決策は、箱から出してすぐには機能しませんでしたが、似たようなことを試してみるようになりました。

from celery.utils import uuid
from celery import chord

class ProgressChord(chord):

    def __call__(self, body=None, **kwargs):
        _chord = self.type
        body = (body or self.kwargs['body']).clone()
        kwargs = dict(self.kwargs, body=body, **kwargs)
        if _chord.app.conf.CELERY_ALWAYS_EAGER:
            return self.apply((), kwargs)
        callback_id = body.options.setdefault('task_id', uuid())
        r= _chord(**kwargs)
        return _chord.AsyncResult(callback_id), r

celery.chordを実行する代わりに、次のようにProgressChordを使用します。

def temptask(n):
    header=list(tempsubtask.si(i) for i in range(n))
    callback=templink.si('printed at last?')
    r = celery.Progresschord(celery.group(header))(callback)
    return r

rの戻り値には、コールバックの非同期結果とグループ結果の両方を持つタプルが含まれていました。したがって、成功は次のようになりました。

In [3]: r
Out[3]: 
(<AsyncResult: bf87507c-14cb-4ac4-8070-d32e4ff326a6>,
 <GroupResult: af69e131-5a93-492d-b985-267484651d95 [4672cbbb-8ec3-4a9e-971a-275807124fae, a236e55f-b312-485c-a816-499d39d7de41, e825a072-b23c-43f2-b920-350413fd5c9e, e3f8378d-fd02-4a34-934b-39a5a735871d, c4f7093b-9f1a-4e5e-b90d-66f83b9c97c4, d5c7dc2c-4e10-4e71-ba2b-055a33e15f02, 07b1c6f7-fe95-4c1f-b0ba-6bc82bceaa4e, 00966cb8-41c2-4e95-b5e7-d8604c000927, e039c78e-6647-4c8d-b59b-e9baf73171a0, 6cfdef0a-25a2-4905-a40e-fea9c7940044]>)

ソースがどこにも見つからなかったので、[celery.chord][1]代わりに継承して上書きしました。[celery.task.chords.Chord][2]

于 2013-03-16T06:23:10.710 に答える
0

古い問題と私は、より良い最新の解決策を見つけるために数日を無駄にしました。現在のプロジェクトでは、グループの進行状況を個別に追跡し、最終的なコールバックでロックを解除する必要があります。

そして、現在の解決策ははるかに単純です（しかし推測するのは難しいです）、件名は最後にコメントされています：

@celery_app.task(name="_scheduler", track_started=True, ignore_result=False)
def _scheduler():
    lock = cache.lock("test_lock")
    if not lock.acquire(blocking=False):
        return {"Error": "Job already in progress"}

    lock_code = lock.local.token.decode("utf-8")

    tasks = []
    for x in range(100):
        tasks.append(calculator.s())
    
    _group = group(*tasks)
    _chord = chord(_group)(_get_results.s(token=lock_code))
    
    group_results = _chord.parent # This is actual group inside chord
    group_results.save() # I am saving it to usual results backend, and can track progress inside.
    
    return _chord # can return anything, I need only chord.

私はで働いていますCelery 5.1

于 2021-06-27T09:58:02.673 に答える