Python中文网 - 问答频道, 解决您学习工作中的Python难题和Bug
Python常见问题
我被一个相对复杂的芹菜链配置所困扰，试图实现以下目标。假设有一系列任务，如下所示：
chain1 = chain(
DownloadFile.s("http://someserver/file.gz"), # downloads file, returns temp file name
UnpackFile.s(), # unpacks the gzip comp'd file, returns temp file name
ParseFile.s(), # parses file, returns list URLs to download
)
现在我想并行下载每个URL，所以我所做的是：
^{pr2}$最后，我想获取从ParseFile返回的每个下载文件（从DownloadFile返回一个临时文件名），并并行地在另一个任务链中运行它（例如，它将是chain的group）：
chains = []
for tmpfile in res:
chains.append(celery.chain(
foo.s(tmpfile),
bar.s(),
baz.s()
))
res2 = celery.group(*chains)()
res2_data = res2.get()
如果我在一个普通的Python进程（而不是另一个celry任务）中运行它，该方法可以正常工作，因为我可以等待来自chain1的结果，然后为每个下载的文件构造下载任务组和新链。在
然而，现在我想把所有这些东西打包到另一个Cerry任务中，方法是将它包装在另一个@app.task修饰函数中，结果发现您不能（或者确实不应该从任务内部调用.get()）等待另一个任务的完成，我没能找到一个解决方案“移植”这个工作流程在一个任务中运行。我试图将res1添加到chain1链中，但是芹菜抱怨{
有谁能给我个建议吗？谢谢！在
Tags： 文件namechaingrouptempreturnsfilecelery
热门问题
- 如何为此数据帧创建散点图？
- 如何为此编写Django模板
- 如何为此表达式编写正则表达式？
- 如何为步进电机选择合适的值？
- 如何为每15分钟间隔的日期时间行（在新列中）添加标签？
- 如何为每一列创建汇总表？
- 如何为每一组groupbyPandas做滚动“得到假人”
- 如何为每一行分别运行函数（python）？
- 如何为每一行生成一个随机数？
- 如何为每一轮将pytorch模型输出存储到numpy
- 如何为每个.py-fi文件创建单独的zip文件
- 如何为每个<li class=”“><a>找到最近的上述同级<li>？
- 如何为每个CSV列生成特定的文件？
- 如何为每个csv文件使用read_csv，即使它是空的？PythonPandas
- 如何为每个CSV文件创建单独的Pandas数据帧并给它们起有意义的名称？
- 如何为每个datetime和每个id创建一行？
- 如何为每个Django型号选择赋予不同的颜色
- 如何为每个Django模型实例安排一个周期性的芹菜任务？
- 如何为每个Django视图设置一个装饰器？
- 如何为每个for循环迭代分配变量
热门文章
- Python覆盖写入文件
- 怎样创建一个 Python 列表？
- Python3 List append()方法使用
- 派森语言
- Python List pop()方法
- Python Django Web典型模块开发实战
- Python input() 函数
- Python3 列表(list) clear()方法
- Python游戏编程入门
- 如何创建一个空的set?
- python如何定义(创建)一个字符串
- Python标准库 [The Python Standard Library by Ex
- Python网络数据爬取及分析从入门到精通（分析篇）
- Python3 for 循环语句
- Python List insert() 方法
- Python 字典(Dictionary) update()方法
- Python编程无师自通 专业程序员的养成
- Python3 List count()方法
- Python 网络爬虫实战 [Web Crawler With Python]
- Python Cookbook（第2版）中文版
实际上，在任务中调用
.get()是不好的。目标Celery是并行执行异步任务，因此您不应该等待结果。在解决问题的一种方法是将第一次处理的url结果存储在文件中或数据库中。在
我写了一个简短的例子来说明如何通过将结果写入文件。我选择了
json转储。在假设您的
urls中有一个urls列表。首先，启动异步处理所有那些group为chain的url。 所有这些任务都将处理url并将要下载的url列表存储在指定tmp目录中的文件中。在然后还启动
check_dir任务，该任务将检查目录中是否已写入文件，在这种情况下，处理每个文件并删除tmp目录中的相应文件。在使用我选择的参数，这个任务每隔30秒自动回复一次，而且永远不会结束（我想你有一个当前的任务要执行），所以你可以改变它，但它是给你一个如何管理的想法。在
我将其作为
main运行，但如果您愿意，也可以将其打包到另一个芹菜任务中。在应用程序_模块.py在
在任务.py在
^{pr2}$在主.py在
控制台输出：
相关问题 更多 >
编程相关推荐