我有这样的情况:
测试集可以以不同的顺序运行并使用diff。执行的时间量。每个worker只应执行一个没有重叠或并发的测试用例。每个工作程序仅从其自己的队列/使用者运行任务。
在以前的版本中,我有一个多处理解决方案,它工作正常。但是对于Celery,我不能从一组中为所有5VM添加所有100个测试用例,它只开始为VM alpha添加任务,并等到它们全部完成以启动下一个VM beta的任务,依此类推。
现在,当我尝试使用多处理为每个工作者创建单独的线程时:AssertionError:不允许守护进程生成子进程
问题是 - 如何同时为5名工人添加100个测试?
因此,每个工作者(来自alpha,beta,......)将同时运行自己的100个测试用例。
使用基于每个消费者的任务键可以解决此问题,例如:
app.control.add_consumer(
queue='alpha',
exchange = 'local',
exchange_type = 'direct',
routing_key = 'alpha.*',
destination = ['worker_for_alpha@HOSTNAME'])
因此,现在您可以使用密钥和队列名称向此使用者发送任何任务以获取单独的工作者:
@app.task(queue='alpha', routing_key = 'alpha.task_for_something')
def any_task(arg_1, arg_2):
do something with arg_1 and arg_2
现在,您可以将其扩展到单个工作人员的任意数量的工作人员或消费者。只需制作一个集合,然后逐个为多个工人\消费者提供它们。
另一个问题可以通过每个worker的--concurrency选项来解决。您可以将并发设置为5,以便在一个worker上同时拥有5个线程。或者使用唯一键和使用者(队列)为每个工作者在单独的线程上中断任务流。