芹菜链监控：简单方法

Question

使用django-celery-3.0.17，celery-3.0.21和django-1.5.1，我正在尝试监控链执行。我找到了一个解决方案，但对我来说似乎有点奇怪，所以如果可能的话，我正在寻找一个更简单的解决方案。这是我的代码：

views.朋友

def runCod(request):
    runFunTask = runFunctions.delay(shpId, codId, stepValues, bboxTuple);
    getRunFunStatus.delay(runFunTask)
    return render_to_response('tss/append_runCod.html',
                             {'runFunTask': runFunTask},
                              context_instance=RequestContext(request))

def getProgressCod(request):
    task = AsyncResult(taskId)
    currStep = task.result['step']
    totSteps = task.result['total']

    if task.status == 'SUCCESS':
        task.revoke() # Manually terminate the runFunctions task

    response = dumps({'status':task.status,'currStep':currStep,'totSteps':totSteps})
    return HttpResponse(response, mimetype='application/json')

tasks.朋友

@task()
def runFunctions(shpId, codId, stepValues, bboxTuple):
    # ... Code to define which functions to launch ...

    stepsToLaunch = [fun1, fun2, fun3, fun4, fun5]
    chainId = chain(stepsToLaunch).apply_async()
    chainAsyncObjects = [node for node in reversed(list(nodes(chainId)))]

    current_task.update_state(state="PROGRESS", meta={'step':1, 'total':numSteps})

    for t in range(10800): # The same max duration of a celery task
        for i, step in enumerate(chainAsyncObjects):
            currStep = i+1
            if step.state == 'PENDING':
                current_task.update_state(state="PROGRESS", meta={'step':currStep, 'total':numSteps})
                break
            if step.state == 'SUCCESS':
                if currStep == numSteps:
                    current_task.update_state(state="SUCCESS", meta={'step':currStep, 'total':numSteps})
                    # This task will be terminated (revoked) by getProgressCod()
            if step.state == 'FAILURE':
                return
    sleep(1)

cods.js

function getProgressCod(taskId){
    var aoiStatus, allStatus, currStep, totSteps;
    var interval = 2000; // Perform ajax call every tot milliseconds

    var refreshId = setInterval(function(){
        $.ajax({
            type: 'get',
            url: 'getProgressCod/',
            data:{'taskId': taskId,},
            success: function(response){},
        });
    }, interval);
}

这就是发生的事情：

1. runCod()启动异步任务runFunctions();
2. runFunctions()创建并启动了一系列子任务
3. 使用runFunctions()最终循环，我每秒都会更新自己的“PROGRESS”状态，查看单链子任务状态。 （参考1和2）
4. 为了知道发生了什么，用户通过getProgressCod() javascript函数通知，这使得每2秒对getProcessCod() python函数的ajax请求
5. getProcessCod() python函数查看runFunctions()状态，当它是'SUCCESS'时，它撤销（终止）runFunctions()执行。

我没有找到另一种方法，因为如果我在链的每个子任务在其最终循环内完成时返回runFunctions()，我无法向用户通知用户其'SUCCESS'状态，因为getProcessCod()将获得执行None的task.status对象

Answer 1

我已经解决了删除getProgressCod() py函数并在runCod()中插入if语句的问题。通过这种方式，我可以使用runFunctions()监视runCod()，当它成功终止时，我等待5秒钟以获得结果，然后我通过返回关闭任务。我唯一怀疑的是，如果这种等待的方法是正确的还是没有...这是我的修改后的代码：

views.朋友

def runCod(request):
    taskId = request.GET['taskId']
    if taskId != '': # If the task is already running
        task = AsyncResult(taskId)
        currStep = task.result['step']
        totSteps = task.result['total']
        response = dumps({'status':task.status,
                          'currStep':currStep,
                          'totSteps':totSteps})
        return HttpResponse(response, mimetype='application/json')
    else: # If the task must be started
    runFunTask = runFunctions.delay(shpId, codId, stepValues, bboxTuple);
    getRunFunStatus.delay(runFunTask)
    return render_to_response('tss/append_runCod.html',
                             {'runFunTask': runFunTask},
                              context_instance=RequestContext(request))

tasks.朋友

@task()
def runFunctions(shpId, codId, stepValues, bboxTuple):
    # ... Code to define which functions to launch ...
    stepsToLaunch = [fun1, fun2, fun3, fun4, fun5]
    chainId = chain(stepsToLaunch).apply_async()
    chainAsyncObjects = [node for node in reversed(list(nodes(chainId)))]

    current_task.update_state(state="PROGRESS", meta={'step':1, 'total':numSteps})

    for t in range(10800): # The same max duration of a celery task
        for i, step in enumerate(chainAsyncObjects):
            currStep = i+1
            if step.state == 'PENDING':
                current_task.update_state(state="PROGRESS", meta={'step':currStep, 'total':numSteps})
                break
            if step.state == 'SUCCESS':
                if currStep == numSteps:
                    current_task.update_state(state="SUCCESS", meta={'step':currStep, 'total':numSteps})
                    sleep(5) # Wait before stop this task, in order for javascript to get the result!
                    return

            if step.state == 'FAILURE':
                return
        sleep(1)

cods.js

function getProgressCod(taskId){
    var interval = 2000; // Perform ajax call every tot milliseconds

    var refreshId = setInterval(function(){
        $.ajax({
            type: 'get',
            url: 'runCod/',
            data:{'taskId': taskId,},
            success: function(response){},
        });
    }, interval);
}