我的python redis队列中有一个嵌套的作业结构。首先执行了RNCopy作业。完成此操作后,接下来会有3个依赖的注册工作。当所有这3个作业的计算完成时,我想触发作业,以将Websocket发送到我的前端。
我当前的尝试:
rncopy = redisqueue.enqueue(raw_nifti_copymachine, patientid, imagepath, timeout=6000)
t1c_reg = redisqueue.enqueue(modality_registrator, patientid, "t1c", timeout=6000, depends_on=rncopy)
t2_reg = redisqueue.enqueue(modality_registrator, patientid, "t2", timeout=6000, depends_on=rncopy)
fla_reg = redisqueue.enqueue(modality_registrator, patientid, "fla", timeout=6000, depends_on=rncopy)
notify = redisqueue.enqueue(print, patient_finished, patientid, timeout=6000, depends_on=(t1c_reg, t2_reg, fla_reg))
不幸的是,似乎从未将多职位依赖性功能合并到主人中。我看到目前在Git上有两个拉的请求。有能力可以使用的解决方法吗?
很抱歉未能提供可再现的例子。
新版本(rq> = 1.8(
您可以简单地使用depends_on
参数,通过列表或元组。
rncopy = redisqueue.enqueue(raw_nifti_copymachine, patientid, imagepath, timeout=6000)
t1c_reg = redisqueue.enqueue(modality_registrator, patientid, "t1c", timeout=6000, depends_on=rncopy)
t2_reg = redisqueue.enqueue(modality_registrator, patientid, "t2", timeout=6000, depends_on=rncopy)
fla_reg = redisqueue.enqueue(modality_registrator, patientid, "fla", timeout=6000, depends_on=rncopy)
notify = redisqueue.enqueue(first_wrapper, patient_finished, patientid,t2_reg.id,fla_reg.id, timeout=6000, depends_on=(t1c_reg, t2_reg, fla_reg))
# you can also use a list instead of a tuple:
# notify = redisqueue.enqueue(first_wrapper, patient_finished, patientid,t2_reg.id,fla_reg.id, timeout=6000, depends_on=[t1c_reg, t2_reg, fla_reg])
旧版本(RQ< 1.8(
我使用此解决方法:如果依赖项为 n ,我创建了真实功能的 n-1 包装器:每个包装器都取决于其他作业。
此解决方案有点偏僻,但起作用。
rncopy = redisqueue.enqueue(raw_nifti_copymachine, patientid, imagepath, timeout=6000)
t1c_reg = redisqueue.enqueue(modality_registrator, patientid, "t1c", timeout=6000, depends_on=rncopy)
t2_reg = redisqueue.enqueue(modality_registrator, patientid, "t2", timeout=6000, depends_on=rncopy)
fla_reg = redisqueue.enqueue(modality_registrator, patientid, "fla", timeout=6000, depends_on=rncopy)
notify = redisqueue.enqueue(first_wrapper, patient_finished, patientid,t2_reg.id,fla_reg.id, timeout=6000, depends_on=t1c_reg)
def first_wrapper(patient_finished, patientid,t2_reg_id,fla_reg_id):
queue = Queue('YOUR-QUEUE-NAME'))
queue.enqueue(second_wrapper, patient_finished, patientid, fla_reg_id, timeout=6000, depends_on=t2_reg_id)
def second_wrapper(patient_finished, patientid,fla_reg_id):
queue = Queue('YOUR-QUEUE-NAME'))
queue.enqueue(print, patient_finished, patientid, timeout=6000, depends_on=fla_reg_id)
一些警告:
我不会将队列对象传递给包装器,因为出现了一些序列化问题。因此,队列必须通过名称恢复...
出于相同的原因,我将Job.ID(而不是作业对象(传递给包装器。
我创建了'an&quot'rq-manager;解决与依赖性等多重和树的类似问题:https://github.com/crispydyne/rq-manager
具有多个依赖性的项目结构看起来像这样。
def simpleTask(x):
return 2*x
project = {'jobs':[
{
'blocking':True, # this job must finished before moving on.
'func':simpleTask,'args': 0
},
{
'blocking':True, # this job, and its child jobs, must finished before moving on.
'jobs':[ # these child jobs will run in parallel
{'func':simpleTask,'args': 1},
{'func':simpleTask,'args': 2},
{'func':simpleTask,'args': 3}],
},
{ # this job will only run when the blocking jobs above finish.
'func':simpleTask,'args': 4
}
]}
然后将其传递给经理完成。
from rq_manager import manager, getProjectResults
managerJob = q.enqueue(manager,project)
projectResults = getProjectResults(managerJob)
返回
projectResults = [0, [2, 4, 6], 8]
当抚养工作需要父母的结果时。我创建了执行第一个作业的函数,然后在项目中添加了其他作业。因此,就您的示例:
def firstTask(patientid,imagepath):
raw_nifti_result = raw_nifti_copymachine(patientid,imagepath)
moreTasks = {'jobs':[
{'func':modality_registrator,'args':(patientid, "t1c", raw_nifti_result)},
{'func':modality_registrator,'args':(patientid, "t2", raw_nifti_result)},
{'func':modality_registrator,'args':(patientid, "fla", raw_nifti_result)},
]}
# returning a dictionary with an "addJobs" will add those tasks to the project.
return {'result':raw_nifti_result, 'addJobs':moreTasks}
该项目看起来像这样:
project = {'jobs':[
{'blocking':True, # this job, and its child jobs, must finished before moving on.
'jobs':[
{
'func':firstTask, 'args':(patientid, imagepath)
'blocking':True, # this job must finished before moving on.
},
# "moreTasks" will be added here
]
}
{ # this job will only run when the blocking jobs above finish.
'func':print,'args': (patient_finished, patientid)
}
]}
如果最终作业需要上一个作业的结果,请设置"上议院"。旗帜。&quot" finaljob"将收到以前结果的数组,并带有其子作业结果的嵌套数组。
def finalJob(previousResults):
# previousResults = [
# raw_nifti_copymachine(patientid,imagepath),
# [
# modality_registrator(patientid, "t1c", raw_nifti_result),
# modality_registrator(patientid, "t2", raw_nifti_result),
# modality_registrator(patientid, "fla", raw_nifti_result),
# ]
# ]
return doSomethingWith(previousResults)
那么该项目看起来像这样
project = {'jobs':[
{
#'blocking':True, # Blocking not needed.
'jobs':[
{
'func':firstTask, 'args':(patientid, imagepath)
'blocking':True, # this job must finished before moving on.
},
# "moreTasks" will be added here
]
}
{ # This job will wait, since it needs the previous job's results.
'func':finalJob, 'previousJobArgs': True # it gets all the previous jobs results
}
]}
希望https://github.com/rq/rq/rq/issues/260可以实现,我的解决方案将被淘汰!