我正尝试在我的Django应用程序中集成测试一个并发的芹菜任务。我想让任务在pytest集成测试期间在一个worker上并发运行,但我在执行这项工作时遇到了问题。for i in range(secs): print(f"\t{i + 1}")
return "DONE" 在脚本中并发运行任务可以很好地工作在conftest.p
我试图通过在pytest套件上运行.delay命令来测试任务: id, select_onlytasks.update_table.delay(id, select_only) # some code here that should be called as soon as task gets started
因此,任务可以运行,但是后面的代码可以同步
在supervisord下,我使用某个并发级别(例如4)运行celery worker:command=/home/httpd/wgusf-wotwgs1/app/bin/celery -A roles.frontend worker -c 4 -l info
directory=/home/httpd/wgusf-wotwgs1