1

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

视图.py

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')

任务.py

@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)

鳕鱼.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()最后的循环中，我每秒都会更新它自己的“进度”状态，查看单链子任务状态。（参考1和2）
  4. 要知道发生了什么，用户会收到 javascript 函数的通知，该函数每 2 秒发出一次对python 函数getProgressCod()的 ajax 请求getProcessCod()
  5. getProcessCod()python 函数查看runFunctions()状态，当它是“成功”时，它会撤销（终止）runFunctions()执行。

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

4

1 回答 1

1

我已经解决getProgressCod()了删除 py 函数并在runCod(). 通过这种方式，我可以监视runFunctions()使用runCod()情况，当它成功终止时，我等待 5 秒钟以获取结果，然后我关闭任务并返回。我唯一剩下的疑问是，如果这种等待方法是否正确......这是我修改后的代码：

视图.py

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))

任务.py

@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)

鳕鱼.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);
}
于 2013-07-16T07:16:53.667 回答